rfc-app/backend/tests/test_admin_create_user_invite_vertical.py

"""End-to-end integration tests for v0.17.0's admin-create user +
invite-email + claim-flow vertical (roadmap item #16, §6.1).

The release lands three halves of the same surface:

  * **Admin-create user** at `POST /api/admin/users`. The admin types
    email, first/last name, role, and an optional custom message. The
    framework provisions the invitee `users` row (granted, with the
    chosen role) and writes a `user_invite_tokens` row carrying the
    bcrypt-hashed opaque token. The "pending invite" discriminator is
    the active `user_invite_tokens` row joined on `invited_user_id`,
    not a NULL column on `users` (the existing `last_seen_at` column
    is NOT NULL). An invite email dispatches via the existing SMTP
    relay.

  * **Pending-invite admin listing** at `GET /api/admin/users/invites`.
    Lists active (not claimed, not expired) invites for the admin's
    "I sent these but they haven't been claimed yet" view.

  * **Claim** at `POST /api/invites/claim`. The invitee POSTs the token
    they got via email; the framework verifies, marks the row claimed,
    signs them in (skipping OTC on first sign-in per the roadmap), and
    returns a `needs_passcode` hint for the frontend to route to the
    passcode-set screen.

The tests prove:

  * The happy path: admin creates → invite row + email envelope land →
    invitee claims with the token → session is established.
  * Non-admin caller is refused 403.
  * Self-invite is refused 422.
  * Duplicate email is refused 409.
  * Owner-grant by non-owner is refused 422.
  * Malformed role is refused 422 (pydantic regex).
  * Custom message over 500 chars is refused 422 (pydantic max_length).
  * Claim with valid token: signs in + marks row claimed.
  * Claim with expired token: HTTP 410.
  * Claim with already-claimed token: HTTP 410.
  * Claim with unknown token: HTTP 400.
  * The admin-create gesture writes a `permission_events` row with
    event_kind='user_invited'.
  * The user listing surfaces the `pending_invite` field for invited-
    but-not-yet-claimed users, and clears it after claim.
"""
from __future__ import annotations

import json

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_invite_envelopes(to_address: str | None = None) -> list[dict]:
    """Pull the invite-kind envelopes off the shared notifier buffer.

    Mirrors the OTC code-extraction helper in
    test_admin_users_vertical.py — invite emails land in the same
    `_SENT` buffer with `kind='invite'`.
    """
    from app import email as email_mod
    out = []
    for env in email_mod.sent_envelopes():
        if env.get("kind") != "invite":
            continue
        if to_address is not None and env["to"] != to_address:
            continue
        out.append(env)
    return out


def _extract_claim_url(envelope: dict) -> str:
    """Pull the claim URL out of the invite email body."""
    for line in envelope["body"].splitlines():
        line = line.strip()
        if line.startswith("http") and "/invites/claim" in line:
            return line
    raise AssertionError(f"no claim URL in envelope body: {envelope['body']!r}")


def _extract_claim_token(envelope: dict) -> str:
    """Pull the `token` query-string param out of the claim URL."""
    from urllib.parse import urlparse, parse_qs
    url = _extract_claim_url(envelope)
    qs = parse_qs(urlparse(url).query)
    return qs["token"][0]


# ---------------------------------------------------------------------------
# Admin create + invite — happy path
# ---------------------------------------------------------------------------


def test_admin_create_user_invite_happy_path(app_with_fake_gitea):
    """Admin creates → user row + invite-token row + email envelope all
    land; the response carries the created ids and the inviter is the
    admin who issued the gesture."""
    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=100, login="adminzero", role="admin")
        sign_in_as(
            client, user_id=100, gitea_login="adminzero",
            display_name="Admin Zero", role="admin",
            email="adminzero@test",
        )
        _reset_outbound()

        r = client.post(
            "/api/admin/users",
            json={
                "email": "invitee@example.com",
                "first_name": "Inv",
                "last_name": "Tee",
                "role": "contributor",
                "custom_message": "We chatted at the conference — welcome!",
            },
        )
        assert r.status_code == 200, r.text
        body = r.json()
        assert body["ok"] is True
        assert body["email"] == "invitee@example.com"
        assert body["role"] == "contributor"
        assert body["invite_id"] > 0
        assert body["invited_user_id"] > 0

        # User row exists with the chosen role + granted. The "pending
        # invite" discriminator is the active `user_invite_tokens` row,
        # not a NULL column on `users` — see the invites.create_invite
        # docstring for the reasoning.
        row = db.conn().execute(
            "SELECT role, permission_state, first_name, last_name "
            "FROM users WHERE email = ? COLLATE NOCASE",
            ("invitee@example.com",),
        ).fetchone()
        assert row is not None
        assert row["role"] == "contributor"
        assert row["permission_state"] == "granted"
        assert row["first_name"] == "Inv"
        assert row["last_name"] == "Tee"

        # Invite-token row exists with the matching ids and the custom
        # message persisted verbatim.
        invite = db.conn().execute(
            "SELECT email, role, custom_message, created_by_admin_id, "
            "invited_user_id, claimed_at FROM user_invite_tokens WHERE id = ?",
            (body["invite_id"],),
        ).fetchone()
        assert invite is not None
        assert invite["email"] == "invitee@example.com"
        assert invite["role"] == "contributor"
        assert invite["custom_message"] == "We chatted at the conference — welcome!"
        assert invite["created_by_admin_id"] == 100
        assert invite["invited_user_id"] == body["invited_user_id"]
        assert invite["claimed_at"] is None

        # Email envelope landed with the invite kind and embeds the
        # custom message + claim URL. The inviter display name comes
        # off the DB row (which provision_user_row sets to
        # login.capitalize()), not the sign_in_as cookie payload.
        envelopes = _outbound_invite_envelopes(to_address="invitee@example.com")
        assert len(envelopes) == 1
        env = envelopes[0]
        assert "Adminzero" in env["subject"] or "Adminzero" in env["body"]
        assert "We chatted at the conference — welcome!" in env["body"]
        # Claim URL is well-formed.
        url = _extract_claim_url(env)
        assert "/invites/claim?token=" in url

        # `permission_events` row landed with event_kind='user_invited'.
        ev = db.conn().execute(
            "SELECT actor_user_id, subject_user_id, event_kind, details "
            "FROM permission_events WHERE event_kind = 'user_invited'"
        ).fetchall()
        assert len(ev) == 1
        assert ev[0]["actor_user_id"] == 100
        assert ev[0]["subject_user_id"] == body["invited_user_id"]
        details = json.loads(ev[0]["details"])
        assert details["email"] == "invitee@example.com"
        assert details["role"] == "contributor"


# ---------------------------------------------------------------------------
# Refusals on the admin-create endpoint
# ---------------------------------------------------------------------------


def test_admin_create_user_invite_refuses_non_admin(app_with_fake_gitea):
    """A contributor caller is refused 403; an anonymous caller 401."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=110, login="contrib", role="contributor")
        sign_in_as(
            client, user_id=110, gitea_login="contrib",
            display_name="Contrib", role="contributor",
        )
        r = client.post(
            "/api/admin/users",
            json={"email": "x@y.com", "role": "contributor"},
        )
        assert r.status_code == 403, r.text

        client.cookies.clear()
        r = client.post(
            "/api/admin/users",
            json={"email": "x@y.com", "role": "contributor"},
        )
        assert r.status_code == 401


def test_admin_create_user_invite_refuses_self_email(app_with_fake_gitea):
    """An admin trying to invite their own email is refused 422 —
    self-invite is the wrong channel; the role-change endpoint exists
    for self-edits."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=120, login="adm", role="admin")
        # Manually set the admin's email since provision_user_row's
        # fixture uses login@test; this is what we'll try to self-invite.
        from app import db
        db.conn().execute(
            "UPDATE users SET email = ? WHERE id = ?",
            ("selfinviter@example.com", 120),
        )
        sign_in_as(
            client, user_id=120, gitea_login="adm",
            display_name="Adm", role="admin",
            email="selfinviter@example.com",
        )
        r = client.post(
            "/api/admin/users",
            json={
                "email": "selfinviter@example.com",
                "role": "contributor",
            },
        )
        assert r.status_code == 422, r.text
        assert "yourself" in r.json()["detail"].lower()


def test_admin_create_user_invite_refuses_duplicate_email(app_with_fake_gitea):
    """An admin trying to invite an email that already maps to a users
    row is refused 409 — the existing role / grant gestures are the
    right surface for an existing user."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=130, login="adminD", role="admin")
        provision_user_row(user_id=131, login="existingone", role="contributor")
        sign_in_as(
            client, user_id=130, gitea_login="adminD",
            display_name="Admin D", role="admin",
        )
        # provision_user_row sets email to <login>@test, so:
        r = client.post(
            "/api/admin/users",
            json={
                "email": "existingone@test",
                "role": "contributor",
            },
        )
        assert r.status_code == 409, r.text


def test_admin_create_user_invite_owner_grant_refused_for_non_owner(app_with_fake_gitea):
    """An admin (not owner) trying to invite a fresh user as `owner` is
    refused 422 — §6.1's owner-zero is the only bootstrap path."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=140, login="adminNoOwner", role="admin")
        sign_in_as(
            client, user_id=140, gitea_login="adminNoOwner",
            display_name="Admin", role="admin",
        )
        r = client.post(
            "/api/admin/users",
            json={
                "email": "wouldbeowner@example.com",
                "role": "owner",
            },
        )
        assert r.status_code == 422, r.text


def test_admin_create_user_invite_owner_can_invite_as_owner(app_with_fake_gitea):
    """A sitting owner can invite a fresh user as `owner` — the §6.1
    role-grant channel. Sanity check that the owner-grant path itself
    works, paired with the refusal above."""
    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=150, login="ownerzero", role="owner")
        sign_in_as(
            client, user_id=150, gitea_login="ownerzero",
            display_name="Owner Zero", role="owner",
        )
        r = client.post(
            "/api/admin/users",
            json={
                "email": "newowner@example.com",
                "role": "owner",
            },
        )
        assert r.status_code == 200, r.text
        row = db.conn().execute(
            "SELECT role FROM users WHERE email = ? COLLATE NOCASE",
            ("newowner@example.com",),
        ).fetchone()
        assert row["role"] == "owner"


def test_admin_create_user_invite_refuses_malformed_role(app_with_fake_gitea):
    """The pydantic regex refuses any role outside the §6.1 set."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=160, login="adminR", role="admin")
        sign_in_as(
            client, user_id=160, gitea_login="adminR",
            display_name="Admin R", role="admin",
        )
        r = client.post(
            "/api/admin/users",
            json={
                "email": "ok@example.com",
                "role": "superuser",
            },
        )
        assert r.status_code == 422


def test_admin_create_user_invite_refuses_long_custom_message(app_with_fake_gitea):
    """Custom message over the 500-char ceiling is refused 422."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=170, login="adminM", role="admin")
        sign_in_as(
            client, user_id=170, gitea_login="adminM",
            display_name="Admin M", role="admin",
        )
        r = client.post(
            "/api/admin/users",
            json={
                "email": "ok@example.com",
                "role": "contributor",
                "custom_message": "x" * 501,
            },
        )
        assert r.status_code == 422


# ---------------------------------------------------------------------------
# Claim flow
# ---------------------------------------------------------------------------


def test_claim_with_valid_token_signs_in_and_marks_claimed(app_with_fake_gitea):
    """End-to-end: admin creates → invitee posts the token to
    /api/invites/claim → session lands + row marked claimed +
    last_seen_at stamps on the user row (the pending-invite
    discriminator clears)."""
    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=200, login="adminC", role="admin")
        sign_in_as(
            client, user_id=200, gitea_login="adminC",
            display_name="Admin C", role="admin",
        )
        _reset_outbound()

        r = client.post(
            "/api/admin/users",
            json={
                "email": "claimant@example.com",
                "first_name": "Clai",
                "last_name": "Mant",
                "role": "contributor",
            },
        )
        assert r.status_code == 200
        invite_id = r.json()["invite_id"]
        invited_user_id = r.json()["invited_user_id"]

        env = _outbound_invite_envelopes("claimant@example.com")[0]
        token = _extract_claim_token(env)

        # The invitee's request is anonymous (they have no session
        # yet). We clear the admin's session cookie to simulate this.
        client.cookies.clear()

        r = client.post(
            "/api/invites/claim",
            json={"token": token},
        )
        assert r.status_code == 200, r.text
        body = r.json()
        assert body["ok"] is True
        assert body["user"]["id"] == invited_user_id
        assert body["user"]["role"] == "contributor"
        assert body["user"]["permission_state"] == "granted"
        # The user has no passcode set yet → frontend should route to
        # passcode-set per the roadmap.
        assert body["needs_passcode"] is True

        # Row marked claimed; last_seen_at populated.
        invite = db.conn().execute(
            "SELECT claimed_at, claimed_by_user_id FROM user_invite_tokens "
            "WHERE id = ?",
            (invite_id,),
        ).fetchone()
        assert invite["claimed_at"] is not None
        assert invite["claimed_by_user_id"] == invited_user_id

        user_row = db.conn().execute(
            "SELECT last_seen_at FROM users WHERE id = ?",
            (invited_user_id,),
        ).fetchone()
        assert user_row["last_seen_at"] is not None


def test_claim_with_expired_token_returns_410(app_with_fake_gitea):
    """A token whose `expires_at` has passed surfaces as HTTP 410."""
    from fastapi.testclient import TestClient
    from app import db, invites

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=210, login="adminE", role="admin")
        sign_in_as(
            client, user_id=210, gitea_login="adminE",
            display_name="Admin E", role="admin",
        )
        _reset_outbound()

        # Create the invite, then back-date the expires_at to the past.
        r = client.post(
            "/api/admin/users",
            json={
                "email": "expired@example.com",
                "role": "contributor",
            },
        )
        assert r.status_code == 200
        invite_id = r.json()["invite_id"]
        db.conn().execute(
            "UPDATE user_invite_tokens SET expires_at = datetime('now', '-1 day') "
            "WHERE id = ?",
            (invite_id,),
        )
        env = _outbound_invite_envelopes("expired@example.com")[0]
        token = _extract_claim_token(env)

        client.cookies.clear()
        r = client.post("/api/invites/claim", json={"token": token})
        assert r.status_code == 410, r.text
        assert "expired" in r.json()["detail"].lower()


def test_claim_with_already_claimed_token_returns_410(app_with_fake_gitea):
    """Re-claiming an already-consumed token surfaces as HTTP 410."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=220, login="adminA", role="admin")
        sign_in_as(
            client, user_id=220, gitea_login="adminA",
            display_name="Admin A", role="admin",
        )
        _reset_outbound()

        r = client.post(
            "/api/admin/users",
            json={
                "email": "twice@example.com",
                "role": "contributor",
            },
        )
        assert r.status_code == 200
        env = _outbound_invite_envelopes("twice@example.com")[0]
        token = _extract_claim_token(env)

        client.cookies.clear()
        # First claim succeeds.
        r = client.post("/api/invites/claim", json={"token": token})
        assert r.status_code == 200
        # Second claim, with the same token, refuses with 410.
        client.cookies.clear()
        r = client.post("/api/invites/claim", json={"token": token})
        assert r.status_code == 410, r.text
        assert "already" in r.json()["detail"].lower()


def test_claim_with_unknown_token_returns_400(app_with_fake_gitea):
    """A token that doesn't match any active invite is HTTP 400."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        # No invite ever created; the token is whatever the attacker
        # types in. The endpoint should refuse without disclosing
        # whether the token "looked" right.
        r = client.post(
            "/api/invites/claim",
            json={"token": "totally-made-up-token-string-that-is-not-real"},
        )
        assert r.status_code == 400, r.text


# ---------------------------------------------------------------------------
# Pending-invite admin listing
# ---------------------------------------------------------------------------


def test_pending_invites_listing_shows_active_invites_only(app_with_fake_gitea):
    """The `GET /api/admin/users/invites` listing filters to active
    invites — claimed and expired rows do not surface here (the admin
    user-listing carries the per-row pending-invite badge for the
    living rows; once claimed, the badge clears)."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=300, login="adminL", role="admin")
        sign_in_as(
            client, user_id=300, gitea_login="adminL",
            display_name="Admin L", role="admin",
        )
        _reset_outbound()

        # Create three invites: one stays pending, one we'll claim, one
        # we'll back-date to expired.
        for email in ("alive@ex.co", "claimed@ex.co", "expired@ex.co"):
            r = client.post(
                "/api/admin/users",
                json={"email": email, "role": "contributor"},
            )
            assert r.status_code == 200

        # Claim the middle one.
        env = _outbound_invite_envelopes("claimed@ex.co")[0]
        token_claim = _extract_claim_token(env)

        # Expire the third one.
        from app import db
        db.conn().execute(
            "UPDATE user_invite_tokens SET expires_at = datetime('now', '-1 day') "
            "WHERE email = 'expired@ex.co'"
        )

        # The admin's session is still on the cookie. Claim works
        # anonymously; we clear and restore.
        admin_cookie = client.cookies.get("rfc_session")
        client.cookies.clear()
        r = client.post("/api/invites/claim", json={"token": token_claim})
        assert r.status_code == 200
        client.cookies.set("rfc_session", admin_cookie)

        r = client.get("/api/admin/users/invites")
        assert r.status_code == 200, r.text
        items = r.json()["items"]
        emails = sorted(i["email"] for i in items)
        assert emails == ["alive@ex.co"]


def test_pending_invite_badge_clears_after_claim(app_with_fake_gitea):
    """The `/api/admin/users` listing surfaces `pending_invite` while
    the invite is unclaimed; after the invitee claims, the row's
    pending_invite is null."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=310, login="adminB", role="admin")
        sign_in_as(
            client, user_id=310, gitea_login="adminB",
            display_name="Admin B", role="admin",
        )
        _reset_outbound()

        r = client.post(
            "/api/admin/users",
            json={"email": "badgey@ex.co", "role": "contributor"},
        )
        assert r.status_code == 200
        invited_id = r.json()["invited_user_id"]

        # Before claim — pending_invite is populated.
        r = client.get("/api/admin/users")
        assert r.status_code == 200
        row = next(u for u in r.json()["items"] if u["id"] == invited_id)
        assert row["pending_invite"] is not None
        assert row["pending_invite"]["invite_id"] > 0

        # Claim.
        env = _outbound_invite_envelopes("badgey@ex.co")[0]
        token = _extract_claim_token(env)
        admin_cookie = client.cookies.get("rfc_session")
        client.cookies.clear()
        r = client.post("/api/invites/claim", json={"token": token})
        assert r.status_code == 200
        client.cookies.set("rfc_session", admin_cookie)

        # After claim — pending_invite is null.
        r = client.get("/api/admin/users")
        assert r.status_code == 200
        row = next(u for u in r.json()["items"] if u["id"] == invited_id)
        assert row["pending_invite"] is None


def test_pending_invites_listing_admin_only(app_with_fake_gitea):
    """The listing requires admin/owner; contributor gets 403."""
    from fastapi.testclient import TestClient

    app, _fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=320, login="contribL", role="contributor")
        sign_in_as(
            client, user_id=320, gitea_login="contribL",
            display_name="Contrib L", role="contributor",
        )
        r = client.get("/api/admin/users/invites")
        assert r.status_code == 403