Ben Stull
1456c8b73f
Wave 5. Roadmap item #16. Folds in #21 Part C Amplitude wiring inline.
From the v0.9.0 /admin/users surface, an admin can create a fresh
users row, assign a role (admin/owner/granted-beta-user), include
an optional custom message, and dispatch an invite email carrying
a single-use opaque claim token (256-bit CSPRNG, bcrypt-at-rest,
7-day TTL). The invitee clicks through, lands at /invites/claim,
optionally checks "trust this device for 30 days", and is signed
in without going through OTC (the token in the email is itself
proof of email control).
Backend: migration 019_user_invite_tokens.sql (auto-applied);
backend/app/invites.py (create + claim + list module);
backend/app/email_invite.py (sibling of email_otc); POST
/api/admin/users + GET /api/admin/users/invites in api_admin;
POST /api/invites/claim in main.py's oauth_router (shares
trust-device cookie helper with /auth/otc/verify). 15 new tests in
test_admin_create_user_invite_vertical.py. permission_events row
with event_kind='user_invited' for the audit trail.
Frontend: Admin.jsx Create-user-invite modal + (pending invite)
badge in UserRow; InviteClaim.jsx /invites/claim landing page.
App.jsx route registration. api.js helpers.
Design choices documented in the CHANGELOG: immediate-send (no
admin-review queue); no bulk-invite (deferred); OTC skipped on
first sign-in (token = proof of email control); no users table
changes (discriminator is active user_invite_tokens row, not a
new first_sign_in_at column); refusal cases (self-invite 422,
duplicate email 409, non-owner admin granting owner 422).
Amplitude wiring (inline, #21 Part C):
- USER_INVITED from CreateUserInviteModal { target_user_id,
initial_role, custom_message_chars }
- INVITE_CLAIMED from InviteClaim { invited_by_admin_id,
initial_role, needs_passcode, trust_device }
- identify() BEFORE the claim event with properties:
claim_method 'admin-invite', invited_at (setOnce),
invited_by_admin_id (setOnce), initial_role (setOnce) —
so the Amplitude user record is created with the OHM
user_id from the very first event the invitee fires
Subagent ο shipped the feature on feature/v0.17.0-admin-create-user
(41b0c6a). Driver-side integration squash-merged into main,
hand-resolved 5 files (VERSION, package.json, CHANGELOG —
strict-descending to 0.17.0 → 0.16.0 → 0.15.0; App.jsx — both
new routes kept; api_admin.py — both per-user additive fields
+ pending-invite query both kept). Added inline Amplitude wiring
in CreateUserInviteModal + InviteClaim. 252 backend tests pass
(33 new across #12 + #16 surfaces). Frontend build verified green.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
650 lines
24 KiB
Python
650 lines
24 KiB
Python
"""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
|