rfc-app/backend/tests/test_beta_access_vertical.py

"""End-to-end integration tests for v0.8.0's open beta-access request
flow (§6.1 / §14.1, roadmap item #6).

The release replaces v0.3.0's `allowed_emails` allowlist as the
admission gate. Any valid email can sign in via the v0.7.0 OTC flow;
a fresh user lands in `permission_state='pending'` until an admin
grants access. The first-OTC flow captures first name, last name,
and a free-text "why I should be included in the beta" via a new
`POST /api/auth/me/beta-request` endpoint.

The tests prove:

  * A fresh OTC user lands `permission_state='pending'` with empty
    profile fields, and the verify-response carries `needs_profile=true`.
  * `POST /api/auth/me/beta-request` populates the three fields and
    leaves the row in `pending`.
  * A pending user is refused write endpoints (representative
    samples: propose RFC, post discussion thread). The refusal is
    403 (not 401 — they're authenticated, just not granted).
  * An admin-grant flow promotes pending → granted. v0.8.0 doesn't
    ship an admin UI for this (deferred to item #7 / v0.9.0), so
    the test flips the column directly via DB and asserts that
    `require_contributor` now admits the user.
  * A grandfathered user (existing row pre-migration, default
    `permission_state='granted'`) is unaffected — write endpoints
    accept them.
  * The `/auth/otc/request` endpoint accepts any email — the
    v0.7.0 allowlist gate is gone from this path. The `allowed_emails`
    table stays in the schema; the admin UI from v0.3.0 continues to
    manage it for the fast-path bypass deployments may use.
"""
from __future__ import annotations

import pytest

from test_propose_vertical import (  # noqa: F401 — fixtures land via import
    FakeGitea,
    app_with_fake_gitea,
    provision_user_row,
    sign_in_as,
    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 from every OTC envelope in the test buffer."""
    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


# ---------------------------------------------------------------------------
# Fresh OTC sign-in lands pending with empty fields
# ---------------------------------------------------------------------------


def test_fresh_otc_user_lands_pending_with_empty_profile(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()

        # Request + verify the OTC.
        r = client.post("/auth/otc/request", json={"email": "newcomer@example.com"})
        assert r.status_code == 200, r.text
        code = _outbound_otc_codes("newcomer@example.com")[-1]

        r = client.post("/auth/otc/verify", json={"email": "newcomer@example.com", "code": code})
        assert r.status_code == 200, r.text
        body = r.json()
        # The verify response carries the new fields v0.8.0 added.
        assert body["needs_profile"] is True
        assert body["user"]["permission_state"] == "pending"

        # The row reflects the same: pending state, no profile yet.
        row = db.conn().execute(
            "SELECT permission_state, first_name, last_name, beta_request_reason FROM users WHERE email = ? COLLATE NOCASE",
            ("newcomer@example.com",),
        ).fetchone()
        assert row is not None
        assert row["permission_state"] == "pending"
        assert row["first_name"] is None
        assert row["last_name"] is None
        assert row["beta_request_reason"] is None

        # /api/auth/me surfaces the same shape.
        me = client.get("/api/auth/me").json()
        assert me["authenticated"] is True
        assert me["user"]["permission_state"] == "pending"
        assert me["user"]["needs_profile"] is True
        assert me["user"]["first_name"] == ""
        assert me["user"]["last_name"] == ""
        assert me["user"]["beta_request_reason"] == ""


# ---------------------------------------------------------------------------
# beta-request endpoint captures the fields and leaves state pending
# ---------------------------------------------------------------------------


def test_beta_request_populates_fields_keeps_state_pending(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()
        # Sign in the fresh user via the full OTC flow.
        client.post("/auth/otc/request", json={"email": "alice@example.com"})
        code = _outbound_otc_codes("alice@example.com")[-1]
        client.post("/auth/otc/verify", json={"email": "alice@example.com", "code": code})

        # Submit the capture form.
        r = client.post(
            "/api/auth/me/beta-request",
            json={
                "first_name": "Alice",
                "last_name": "Liddell",
                "beta_request_reason": "I want to help write the RFCs.",
            },
        )
        assert r.status_code == 200, r.text

        # The row reflects the captured fields; state stays pending.
        row = db.conn().execute(
            "SELECT permission_state, first_name, last_name, beta_request_reason FROM users WHERE email = ? COLLATE NOCASE",
            ("alice@example.com",),
        ).fetchone()
        assert row["permission_state"] == "pending"
        assert row["first_name"] == "Alice"
        assert row["last_name"] == "Liddell"
        assert row["beta_request_reason"] == "I want to help write the RFCs."

        # /api/auth/me now reports needs_profile=false (fields are set).
        me = client.get("/api/auth/me").json()
        assert me["user"]["permission_state"] == "pending"
        assert me["user"]["needs_profile"] is False
        assert me["user"]["first_name"] == "Alice"


def test_beta_request_refuses_anonymous(app_with_fake_gitea):
    """The endpoint requires authentication — an anonymous caller can't
    file a request without first signing in via OTC."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        client.cookies.clear()
        r = client.post(
            "/api/auth/me/beta-request",
            json={"first_name": "A", "last_name": "B", "beta_request_reason": "Hi"},
        )
        assert r.status_code == 401


def test_beta_request_refuses_granted_user(app_with_fake_gitea):
    """A grandfathered (already granted) user has no business filing a
    beta request. The endpoint refuses with 409 so the client can
    distinguish the failure from "we don't know you" (401)."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=1, login="grandfathered", role="contributor")
        sign_in_as(
            client,
            user_id=1,
            gitea_login="grandfathered",
            display_name="Grandfathered",
            role="contributor",
        )
        r = client.post(
            "/api/auth/me/beta-request",
            json={"first_name": "G", "last_name": "F", "beta_request_reason": "x"},
        )
        assert r.status_code == 409


# ---------------------------------------------------------------------------
# Pending user is refused write endpoints; admin grant promotes them
# ---------------------------------------------------------------------------


def test_pending_user_is_refused_write_endpoints(app_with_fake_gitea):
    """A pending user can read everything anonymous can read, but every
    write-shaped endpoint refuses with 403. The refusal shape mirrors
    the v0.6.0 / item #4 audit's anon-401 — both are "no contributor
    capability"; pending is the authenticated-but-ungranted variant.

    Representative samples: propose RFC, post discussion thread.
    """
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        # Sign in via fresh OTC — lands pending.
        client.post("/auth/otc/request", json={"email": "pending@example.com"})
        code = _outbound_otc_codes("pending@example.com")[-1]
        client.post("/auth/otc/verify", json={"email": "pending@example.com", "code": code})

        # Reads work — every anonymous surface stays reachable.
        assert client.get("/api/health").status_code == 200
        assert client.get("/api/rfcs").status_code == 200
        assert client.get("/api/philosophy").status_code == 200

        # Propose — write-shaped, refused with 403.
        r = client.post(
            "/api/rfcs/propose",
            json={"title": "T", "slug": "t", "pitch": "p", "tags": []},
        )
        assert r.status_code == 403
        # The error body mentions the review state so a UI surface can
        # render the right message — but the test asserts only on the
        # status code (the body shape is the FastAPI default detail).


def test_admin_grant_promotes_pending_to_granted(app_with_fake_gitea):
    """v0.8.0 doesn't ship an admin UI for this — it's deferred to
    item #7 / v0.9.0. For this release, an admin gesture is an
    `UPDATE users SET permission_state='granted' WHERE email=?`. The
    test flips the column directly via DB and asserts the
    `require_contributor` gate now admits the user.
    """
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        _reset_outbound()
        # Sign in a fresh OTC user — lands pending.
        client.post("/auth/otc/request", json={"email": "promoted@example.com"})
        code = _outbound_otc_codes("promoted@example.com")[-1]
        client.post("/auth/otc/verify", json={"email": "promoted@example.com", "code": code})

        # Before the grant: propose refused with 403.
        r = client.post(
            "/api/rfcs/propose",
            json={"title": "T", "slug": "t-pre", "pitch": "p", "tags": []},
        )
        assert r.status_code == 403

        # The admin gesture (v0.8.0 shape — direct UPDATE; v0.9.0 will
        # ship a UI). The test stamps `permission_decided_by` and
        # `permission_decided_at` as the v0.9.0 admin UI will, so the
        # column population exercises the schema slot. user_id=99 is
        # a placeholder admin row — provision it so the FK resolves.
        provision_user_row(user_id=99, login="adminuser", role="admin")
        db.conn().execute(
            """
            UPDATE users
               SET permission_state = 'granted',
                   permission_decided_by = 99,
                   permission_decided_at = datetime('now')
             WHERE email = ?
            """,
            ("promoted@example.com",),
        )

        # The next request reads the fresh column from the DB. The
        # propose endpoint reaches the route body now (it then refuses
        # for a different reason — the slug 't-prop' will fail
        # the slug-format check or hit a mock-gitea path — but the
        # status code is _not_ 403/401, which is the v0.8.0 assertion).
        r = client.post(
            "/api/rfcs/propose",
            json={"title": "Title", "slug": "tprop", "pitch": "Pitch text.", "tags": []},
        )
        assert r.status_code != 403, r.text
        assert r.status_code != 401, r.text


def test_grandfathered_user_is_unaffected_by_migration(app_with_fake_gitea):
    """An existing `users` row at migration time has
    `permission_state='granted'` via the column default. The
    grandfathered user passes write endpoints without filing a
    beta request and without the admin UI. v0.6.0 (anon-write
    audit) is the v0.6.0 contract; v0.8.0 widens the gate but
    does not break this case.
    """
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=5, login="oldhand", role="contributor")
        # provision_user_row uses INSERT OR REPLACE INTO users with
        # the column list it knows; permission_state is not in that
        # list, so it picks up the column default ('granted') on
        # insert. Confirm directly.
        row = db.conn().execute(
            "SELECT permission_state FROM users WHERE id = 5"
        ).fetchone()
        assert row["permission_state"] == "granted"

        sign_in_as(
            client,
            user_id=5,
            gitea_login="oldhand",
            display_name="Old Hand",
            role="contributor",
        )

        # Propose is write-shaped; the call should not refuse on
        # the permission_state gate. (Subsequent failure modes —
        # e.g. mock-gitea wiring — are not the v0.8.0 concern; this
        # test asserts on the gate, not the propose body's success.)
        r = client.post(
            "/api/rfcs/propose",
            json={"title": "Title", "slug": "gf-slug", "pitch": "Pitch.", "tags": []},
        )
        assert r.status_code != 403, r.text
        assert r.status_code != 401, r.text


# ---------------------------------------------------------------------------
# /auth/otc/request accepts any email — the v0.7.0 allowlist gate is gone
# ---------------------------------------------------------------------------


def test_otc_request_accepts_any_email_regardless_of_allowlist(app_with_fake_gitea):
    """v0.7.0 silently dropped OTC requests for emails not on the
    `allowed_emails` table. v0.8.0 reverses this: the request
    endpoint sends a code to any valid email; admission gates at
    `permission_state` post-verify instead. The `allowed_emails`
    table stays in the schema as a fast-path bypass for
    deployments that want to pre-mark known-good emails (the v0.9.0
    admin user-management page will collapse the two surfaces).
    """
    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 so the v0.7.0
        # gate would have engaged. v0.8.0 ignores it for the request
        # path.
        db.conn().execute("INSERT INTO allowed_emails (email) VALUES (?)", ("known@example.com",))

        # An email NOT on the allowlist still gets a code under v0.8.0.
        r = client.post("/auth/otc/request", json={"email": "stranger@example.com"})
        assert r.status_code == 200
        codes = _outbound_otc_codes("stranger@example.com")
        assert len(codes) == 1, "OTC code must be sent regardless of allowlist state"

        # The row is there and the user can complete sign-in (and will
        # land in 'pending' per the other tests).
        row = db.conn().execute(
            "SELECT 1 FROM otc_codes WHERE email = ?",
            ("stranger@example.com",),
        ).fetchone()
        assert row is not None


def test_allowlist_table_still_present_in_schema(app_with_fake_gitea):
    """The schema migration leaves the `allowed_emails` table in
    place — the admin UI from v0.3.0 still manages it for the
    fast-path bypass deployments may use. This is a regression net
    for "did the v0.8.0 cleanup accidentally drop the table"."""
    from fastapi.testclient import TestClient
    from app import db

    app, _fake = app_with_fake_gitea
    with TestClient(app):
        # The table accepts inserts (i.e. it exists) — no schema check
        # gymnastics needed.
        db.conn().execute("INSERT INTO allowed_emails (email) VALUES (?)", ("kept@example.com",))
        row = db.conn().execute(
            "SELECT email FROM allowed_emails WHERE email = ?",
            ("kept@example.com",),
        ).fetchone()
        assert row is not None