rfc-app/backend/tests/test_otc_vertical.py

"""End-to-end integration tests for the v0.7.0 email/OTC sign-in
vertical (§6.2).

The release replaces the Gitea OAuth gesture as the primary human
sign-in path. The tests prove:

  * `/auth/otc/request` is rate-limited per-email — back-to-back
    requests inside `OTC_REQUEST_COOLDOWN_SECONDS` are refused with
    429 (the loud-failure shape the spec calls out).
  * The happy path: request → code lands in the outbound buffer →
    verify with the code → session cookie surfaces an authenticated
    user via `/api/auth/me`.
  * Expired codes refuse with 400.
  * Already-consumed codes refuse with 400 on re-use.
  * Wrong codes refuse with 400.
  * Allowlist gate: when `allowed_emails` is populated and the email
    isn't on it, the response is still 202 (no leak), but no email
    lands in the outbound buffer and verify finds no matching code.
  * Migration link: an existing OAuth-era user (with a `users.email`
    row) is linked by email on first OTC sign-in — `gitea_id` is
    preserved.
  * Provisioning path: an unrecognized email creates a fresh
    contributor row with `gitea_id = NULL`.

The Gitea bot user + token are still required at process construction
(every test harness sets the same `GITEA_*` env vars); the OTC flow
itself never reaches Gitea. The fakes from `test_propose_vertical`
remain in scope so the rest of the app boots cleanly.
"""
from __future__ import annotations

import pytest

from test_propose_vertical import (  # noqa: F401
    FakeGitea,
    app_with_fake_gitea,
    provision_user_row,
    tmp_env,
)


def _reset_outbound():
    from app import email as email_mod
    email_mod.reset_sent_envelopes()


def _outbound_otc_codes(to_address: str | None = None) -> list[str]:
    """Pluck the `code` line out of every OTC email in the test buffer.

    The OTC mailer stamps `kind='otc'` on the envelope so the §15.4
    notification mailer's envelopes (the unsubscribe-footer shape)
    don't accidentally satisfy the assertion. Each envelope's body
    carries the code on its own indented line; this helper extracts
    just that token so the test reads the same way the user would
    read the email.
    """
    from app import email as email_mod
    out = []
    for env in email_mod.sent_envelopes():
        if env.get("kind") != "otc":
            continue
        if to_address is not None and env["to"] != to_address:
            continue
        for line in env["body"].splitlines():
            tok = line.strip()
            if tok.isdigit() and len(tok) == 6:
                out.append(tok)
                break
    return out


# ---------------------------------------------------------------------------
# Happy path
# ---------------------------------------------------------------------------


def test_otc_request_then_verify_signs_in_a_fresh_user(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()

        # Request: 202 + a single OTC envelope to the requested address.
        r = client.post("/auth/otc/request", json={"email": "newcomer@example.com"})
        assert r.status_code == 200, r.text
        codes = _outbound_otc_codes("newcomer@example.com")
        assert len(codes) == 1
        code = codes[0]

        # Verify: 200 + session cookie + me-shape now reads authenticated.
        r = client.post("/auth/otc/verify", json={"email": "newcomer@example.com", "code": code})
        assert r.status_code == 200, r.text
        me = client.get("/api/auth/me").json()
        assert me["authenticated"] is True
        assert me["user"]["email"] == "newcomer@example.com"
        # Fresh provisioning: no gitea linker. The display name is the
        # local part of the email per §6.2.
        assert me["user"]["role"] == "contributor"
        assert me["user"]["display_name"] == "newcomer"

        # The `users` row reflects the same: gitea_id NULL, email set.
        from app import db
        row = db.conn().execute(
            "SELECT gitea_id, email FROM users WHERE email = ? COLLATE NOCASE",
            ("newcomer@example.com",),
        ).fetchone()
        assert row is not None
        assert row["gitea_id"] is None
        assert row["email"] == "newcomer@example.com"


# ---------------------------------------------------------------------------
# Failure modes on verify
# ---------------------------------------------------------------------------


def test_otc_verify_refuses_wrong_code(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        client.post("/auth/otc/request", json={"email": "alice@example.com"})
        r = client.post("/auth/otc/verify", json={"email": "alice@example.com", "code": "000000"})
        assert r.status_code == 400


def test_otc_verify_refuses_consumed_code(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        client.post("/auth/otc/request", json={"email": "alice@example.com"})
        code = _outbound_otc_codes("alice@example.com")[-1]
        # First verify succeeds.
        r1 = client.post("/auth/otc/verify", json={"email": "alice@example.com", "code": code})
        assert r1.status_code == 200
        # Drop the session cookie so the re-verify reads as fresh.
        client.cookies.clear()
        # Second verify with the same code is refused — `consumed_at`
        # stamped on the row blocks the replay.
        r2 = client.post("/auth/otc/verify", json={"email": "alice@example.com", "code": code})
        assert r2.status_code == 400


def test_otc_verify_refuses_expired_code(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        client.post("/auth/otc/request", json={"email": "alice@example.com"})
        code = _outbound_otc_codes("alice@example.com")[-1]
        # Backdate the row's expires_at to the past. The TTL setting is
        # an env var (default 10 min); rather than waiting, the test
        # rewrites the row.
        db.conn().execute(
            "UPDATE otc_codes SET expires_at = datetime('now', '-1 minute') WHERE email = ?",
            ("alice@example.com",),
        )
        r = client.post("/auth/otc/verify", json={"email": "alice@example.com", "code": code})
        assert r.status_code == 400


# ---------------------------------------------------------------------------
# Rate limiting
# ---------------------------------------------------------------------------


def test_otc_request_rate_limited_per_email(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        r1 = client.post("/auth/otc/request", json={"email": "alice@example.com"})
        assert r1.status_code == 200
        # Cooldown defaults to 60s; the second back-to-back call is
        # refused with a loud 429.
        r2 = client.post("/auth/otc/request", json={"email": "alice@example.com"})
        assert r2.status_code == 429
        # The buffer still has exactly one envelope — the rate-limited
        # call didn't double-send.
        assert len(_outbound_otc_codes("alice@example.com")) == 1


def test_otc_request_cooldown_is_per_email_not_global(app_with_fake_gitea):
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        r1 = client.post("/auth/otc/request", json={"email": "alice@example.com"})
        assert r1.status_code == 200
        # Different email, fresh cooldown.
        r2 = client.post("/auth/otc/request", json={"email": "bob@example.com"})
        assert r2.status_code == 200


# ---------------------------------------------------------------------------
# Allowlist gate
# ---------------------------------------------------------------------------


def test_otc_request_silently_drops_when_email_not_on_allowlist(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        # Populate the allowlist so the gate turns on.
        db.conn().execute("INSERT INTO allowed_emails (email) VALUES (?)", ("invited@example.com",))

        r = client.post("/auth/otc/request", json={"email": "stranger@example.com"})
        # Still 202 — the allowlist's state is not leaked to callers.
        assert r.status_code == 200
        # But no email was sent, and no row landed in otc_codes.
        assert _outbound_otc_codes("stranger@example.com") == []
        row = db.conn().execute(
            "SELECT 1 FROM otc_codes WHERE email = ?",
            ("stranger@example.com",),
        ).fetchone()
        assert row is None


def test_otc_request_admits_allowlisted_email(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        db.conn().execute("INSERT INTO allowed_emails (email) VALUES (?)", ("invited@example.com",))
        r = client.post("/auth/otc/request", json={"email": "invited@example.com"})
        assert r.status_code == 200
        assert len(_outbound_otc_codes("invited@example.com")) == 1


# ---------------------------------------------------------------------------
# Migration path — link by email to an OAuth-era user
# ---------------------------------------------------------------------------


def test_otc_links_to_existing_oauth_user_by_email(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        # Seed an OAuth-era row. `provision_user_row` writes
        # email=<login>@test, so we sign in via OTC with the matching
        # email and expect the same `users.id` to come back.
        provision_user_row(user_id=42, login="legacyuser", role="contributor")
        existing = db.conn().execute(
            "SELECT id, gitea_id FROM users WHERE id = ?", (42,)
        ).fetchone()
        assert existing["gitea_id"] == 42  # OAuth linker is set.

        r = client.post("/auth/otc/request", json={"email": "legacyuser@test"})
        assert r.status_code == 200
        code = _outbound_otc_codes("legacyuser@test")[-1]
        r = client.post("/auth/otc/verify", json={"email": "legacyuser@test", "code": code})
        assert r.status_code == 200

        # /api/auth/me reports the linked user — same id, original role.
        me = client.get("/api/auth/me").json()
        assert me["authenticated"] is True
        assert me["user"]["id"] == 42
        assert me["user"]["role"] == "contributor"

        # gitea_id is preserved on the linked row — the migration path
        # doesn't disturb the OAuth linker.
        row = db.conn().execute(
            "SELECT gitea_id FROM users WHERE id = ?", (42,)
        ).fetchone()
        assert row["gitea_id"] == 42


def test_otc_provisions_fresh_user_when_email_matches_no_one(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        r = client.post("/auth/otc/request", json={"email": "newperson@example.com"})
        assert r.status_code == 200
        code = _outbound_otc_codes("newperson@example.com")[-1]
        r = client.post("/auth/otc/verify", json={"email": "newperson@example.com", "code": code})
        assert r.status_code == 200

        # A fresh row landed with NULL gitea_id (no OAuth linker).
        row = db.conn().execute(
            "SELECT id, gitea_id, gitea_login, role FROM users WHERE email = ? COLLATE NOCASE",
            ("newperson@example.com",),
        ).fetchone()
        assert row is not None
        assert row["gitea_id"] is None
        assert row["gitea_login"] is None
        assert row["role"] == "contributor"


# ---------------------------------------------------------------------------
# Re-request invalidates prior code
# ---------------------------------------------------------------------------


def test_otc_re_request_invalidates_prior_unused_code(app_with_fake_gitea, monkeypatch):
    from fastapi.testclient import TestClient

    # Drop the cooldown so the second request lands instead of 429ing.
    monkeypatch.setenv("OTC_REQUEST_COOLDOWN_SECONDS", "0")
    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        client.post("/auth/otc/request", json={"email": "alice@example.com"})
        first = _outbound_otc_codes("alice@example.com")[-1]
        client.post("/auth/otc/request", json={"email": "alice@example.com"})
        second = _outbound_otc_codes("alice@example.com")[-1]
        assert first != second

        # The old code is invalidated — verify with `first` now refuses.
        r = client.post("/auth/otc/verify", json={"email": "alice@example.com", "code": first})
        assert r.status_code == 400

        # The new code still works.
        r = client.post("/auth/otc/verify", json={"email": "alice@example.com", "code": second})
        assert r.status_code == 200