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 (v0.8.0 update): v0.7.0 silently dropped requests
    for emails not on `allowed_emails`. v0.8.0 (item #6) removed
    that gate from the request path; the admission gate is now
    `permission_state` on the freshly-provisioned `users` row,
    asserted in test_beta_access_vertical.py. The tests below
    confirm v0.8.0's open-request shape for both on-list and
    off-list emails.
  * 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 — v0.8.0 update
#
# v0.7.0 gated the OTC request endpoint on the `allowed_emails` table:
# emails not on the list got a silent drop (still 202, but no code).
# v0.8.0 (roadmap item #6) reverses this: the request endpoint
# accepts any valid email and sends a code. The admission gate moves
# to `permission_state` on the freshly-provisioned `users` row,
# which the next-tier tests in test_beta_access_vertical.py cover.
# The `allowed_emails` table stays in the schema as a fast-path
# bypass for admin convenience.
# ---------------------------------------------------------------------------


def test_otc_request_admits_emails_regardless_of_allowlist_population(app_with_fake_gitea):
    """v0.8.0: the OTC request path no longer consults `allowed_emails`.
    Whether the allowlist is empty or populated, every valid email
    receives a code; admission gates at `permission_state` post-verify.
    """
    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 with one specific email; the v0.7.0
        # gate would have engaged here.
        db.conn().execute("INSERT INTO allowed_emails (email) VALUES (?)", ("invited@example.com",))

        # The not-on-list email still gets a code under v0.8.0.
        r = client.post("/auth/otc/request", json={"email": "stranger@example.com"})
        assert r.status_code == 200
        assert len(_outbound_otc_codes("stranger@example.com")) == 1


def test_otc_request_admits_allowlisted_email(app_with_fake_gitea):
    """v0.8.0: still works for emails that happen to be on the legacy
    allowlist — the table is no longer consulted at request time but
    populated rows are admitted alongside everyone else (since the
    gate is now open at the request surface)."""
    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


# ---------------------------------------------------------------------------
# v0.18.0: envelope headers — Slice 2
#
# OTC mail goes through `build_envelope` and MUST land Date,
# Message-ID, and Auto-Submitted but MUST NOT carry a
# List-Unsubscribe header (the recipient explicitly requested the
# code; advertising a list semantic would be wrong).
# ---------------------------------------------------------------------------


def _last_otc_envelope():
    from app import email as email_mod
    otc = [e for e in email_mod.sent_envelopes() if e.get("kind") == "otc"]
    assert otc, "no OTC envelope in the buffer"
    return otc[-1]


def test_otc_envelope_sets_date_messageid_autosubmitted(app_with_fake_gitea):
    from fastapi.testclient import TestClient
    from email.utils import parsedate_to_datetime

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        client.post("/auth/otc/request", json={"email": "headers@example.com"})
        msg = _last_otc_envelope()["message"]
        # Date is RFC 5322 parseable.
        assert parsedate_to_datetime(msg["Date"]) is not None
        # Message-ID is bracketed and carries the From-address @-domain.
        mid = msg["Message-ID"]
        assert mid.startswith("<") and mid.endswith(">")
        # Auto-Submitted prevents auto-responder loops.
        assert msg["Auto-Submitted"] == "auto-generated"


def test_otc_envelope_has_no_list_unsubscribe(app_with_fake_gitea):
    """The recipient explicitly typed their email and asked for a
    code; the framework MUST NOT advertise a list semantic on this
    mail. Per the v0.18.0 proposal's tradeoff discussion."""
    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": "headers@example.com"})
        msg = _last_otc_envelope()["message"]
        assert msg["List-Unsubscribe"] is None
        assert msg["List-Unsubscribe-Post"] is None