Ben Stull
426 lines
16 KiB
Python
426 lines
16 KiB
Python
"""End-to-end integration tests for v0.9.0's admin user-management page
|
|
and new-beta-request notifications (roadmap item #7, §6.1 / §15).
|
|
|
|
The release lands two halves of the same surface:
|
|
|
|
* **Admin notification on new beta request.** When a pending user
|
|
submits `POST /api/auth/me/beta-request`, every owner/admin
|
|
receives a `new_beta_request` notification (the §15 substrate
|
|
insert lands the row; the §15.4 email path dispatches subject to
|
|
the recipient's `email_admin_actionable` toggle).
|
|
|
|
* **Admin user-management surface** at `/admin/users`. The
|
|
`GET /api/admin/users` listing carries every user with their
|
|
permission_state, profile fields, sign-up reason, and decision
|
|
audit. The new `POST /api/admin/users/<id>/permission` endpoint
|
|
flips the column and writes a `permission_events` row.
|
|
|
|
The tests prove:
|
|
|
|
* The first beta-request submission fans a `new_beta_request`
|
|
row out to every admin/owner (and not to the requester
|
|
themselves). The row carries the captured profile in
|
|
`payload.extras`.
|
|
* Re-submitting the form from the same pending user doesn't
|
|
re-fan (we only notify on the row's first complete state).
|
|
* `GET /api/admin/users` carries the v0.9.0 columns
|
|
(permission_state, first/last/reason, decided_by).
|
|
* `POST /api/admin/users/<id>/permission` flips the state,
|
|
stamps decided_by/at, and writes a `permission_events` row.
|
|
* The endpoint refuses self-flip (422) and refuses non-admin
|
|
callers (403).
|
|
* The endpoint accepts only the three valid states (422 on
|
|
anything else).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
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]:
|
|
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
|
|
|
|
|
|
def _provision_pending_user(client, email: str) -> int:
|
|
"""Sign in a fresh OTC user (lands `pending`) and return their user_id."""
|
|
from app import db
|
|
_reset_outbound()
|
|
client.post("/auth/otc/request", json={"email": email})
|
|
code = _outbound_otc_codes(email)[-1]
|
|
client.post("/auth/otc/verify", json={"email": email, "code": code})
|
|
row = db.conn().execute(
|
|
"SELECT id FROM users WHERE email = ? COLLATE NOCASE", (email,)
|
|
).fetchone()
|
|
return row["id"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Admin notification on beta-request submission
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_beta_request_submission_notifies_every_admin(app_with_fake_gitea):
|
|
"""First-time submission of a beta-request fans a notification out
|
|
to every owner and admin. The requester themselves never receives
|
|
a row (filtered out by user_id even if they happened to be in the
|
|
admin set, which they aren't in practice — fresh OTC users are
|
|
`contributor`+`pending`)."""
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
# Provision two admins and one owner so the fan-out has multiple
|
|
# targets. The OWNER_GITEA_LOGIN-derived ownership doesn't fire
|
|
# here (no OAuth round-trip in this path); we seed the role
|
|
# directly.
|
|
provision_user_row(user_id=10, login="ownerzero", role="owner")
|
|
provision_user_row(user_id=11, login="admin_one", role="admin")
|
|
provision_user_row(user_id=12, login="admin_two", role="admin")
|
|
provision_user_row(user_id=13, login="contrib_one", role="contributor")
|
|
|
|
# Sign in a fresh OTC user → permission_state='pending'.
|
|
requester_id = _provision_pending_user(client, "newbie@example.com")
|
|
|
|
# Capture-form submit.
|
|
r = client.post(
|
|
"/api/auth/me/beta-request",
|
|
json={
|
|
"first_name": "Newt",
|
|
"last_name": "Newcomer",
|
|
"beta_request_reason": "I want to write the Human RFC.",
|
|
},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
|
|
# Every owner + admin gets a `new_beta_request` notification.
|
|
# The contributor (id=13) does not. The requester (whoever id
|
|
# they got) does not.
|
|
rows = db.conn().execute(
|
|
"""
|
|
SELECT recipient_user_id, event_kind, actor_user_id, payload
|
|
FROM notifications
|
|
WHERE event_kind = 'new_beta_request'
|
|
"""
|
|
).fetchall()
|
|
recipients = sorted(r["recipient_user_id"] for r in rows)
|
|
assert recipients == [10, 11, 12], f"unexpected recipients: {recipients}"
|
|
# Actor is the requester (§15.9: never the bot).
|
|
for r in rows:
|
|
assert r["actor_user_id"] == requester_id
|
|
import json as _json
|
|
extras = _json.loads(r["payload"])
|
|
assert extras["requester_first_name"] == "Newt"
|
|
assert extras["requester_last_name"] == "Newcomer"
|
|
assert extras["requester_email"] == "newbie@example.com"
|
|
|
|
|
|
def test_beta_request_resubmit_does_not_re_notify(app_with_fake_gitea):
|
|
"""Once a user has completed the capture form, re-submitting it
|
|
(the endpoint is idempotent for pending users) must not re-fan a
|
|
fresh notification to every admin — that would carpet-bomb the
|
|
inbox on every typo correction."""
|
|
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=20, login="adminzero", role="admin")
|
|
_provision_pending_user(client, "carpet@example.com")
|
|
|
|
body = {
|
|
"first_name": "Carpet",
|
|
"last_name": "Bomb",
|
|
"beta_request_reason": "first draft",
|
|
}
|
|
r1 = client.post("/api/auth/me/beta-request", json=body)
|
|
assert r1.status_code == 200
|
|
|
|
# Re-submit with edited reason — endpoint accepts (idempotent
|
|
# update), but the admin inbox stays at one row.
|
|
body2 = dict(body, beta_request_reason="cleaner final draft")
|
|
r2 = client.post("/api/auth/me/beta-request", json=body2)
|
|
assert r2.status_code == 200
|
|
|
|
rows = db.conn().execute(
|
|
"SELECT COUNT(*) AS n FROM notifications WHERE event_kind = 'new_beta_request'"
|
|
).fetchone()
|
|
assert rows["n"] == 1
|
|
|
|
|
|
def test_beta_request_notification_is_admin_actionable_category(app_with_fake_gitea):
|
|
"""The §15.4 category mapping must route `new_beta_request` to the
|
|
admin-actionable bucket so the email gate consults
|
|
`email_admin_actionable` (and skips for non-admin recipients).
|
|
"""
|
|
from app import email as email_mod
|
|
|
|
assert email_mod.category_for("new_beta_request", "structural") == "admin-actionable"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# /api/admin/users — listing carries the v0.9.0 columns
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_admin_users_listing_carries_permission_columns(app_with_fake_gitea):
|
|
"""The Users tab consumes this shape — confirm every required
|
|
column is on the response."""
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
# Seed an admin and a pending user with all the v0.8.0 columns
|
|
# populated. Direct-DB insert avoids the OTC dance (which would
|
|
# overwrite the cookie); the test above proves the capture
|
|
# pathway end-to-end and this one just exercises the listing
|
|
# surface's shape.
|
|
provision_user_row(user_id=30, login="ben", role="owner")
|
|
db.conn().execute(
|
|
"""
|
|
INSERT INTO users (id, gitea_id, gitea_login, email,
|
|
display_name, avatar_url, role,
|
|
permission_state, first_name, last_name,
|
|
beta_request_reason)
|
|
VALUES (31, NULL, NULL, 'pendinguser@example.com',
|
|
'pendinguser', '', 'contributor',
|
|
'pending', 'Penn', 'Ding', 'I want in.')
|
|
"""
|
|
)
|
|
|
|
sign_in_as(
|
|
client, user_id=30, gitea_login="ben",
|
|
display_name="Ben", role="owner",
|
|
)
|
|
|
|
r = client.get("/api/admin/users")
|
|
assert r.status_code == 200
|
|
items = r.json()["items"]
|
|
assert isinstance(items, list)
|
|
pending = next(
|
|
(i for i in items if i["email"] == "pendinguser@example.com"), None,
|
|
)
|
|
assert pending is not None
|
|
assert pending["permission_state"] == "pending"
|
|
assert pending["first_name"] == "Penn"
|
|
assert pending["last_name"] == "Ding"
|
|
assert pending["beta_request_reason"] == "I want in."
|
|
assert pending["permission_decided_at"] is None
|
|
assert pending["permission_decided_by_login"] is None
|
|
# Pending bucket is listed first (sort order).
|
|
assert items[0]["permission_state"] == "pending"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# /api/admin/users/<id>/permission — the flip endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_permission_flip_grant_promotes_pending_to_granted(app_with_fake_gitea):
|
|
"""The end-to-end gesture: a fresh OTC user lands pending, an admin
|
|
flips them to granted via the endpoint, the row reflects the new
|
|
state + decided_by/at, and a `permission_events` audit row lands."""
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
# Pending user.
|
|
pending_id = _provision_pending_user(client, "flip@example.com")
|
|
|
|
# Admin acting on them.
|
|
provision_user_row(user_id=40, login="adminflipper", role="admin")
|
|
sign_in_as(
|
|
client, user_id=40, gitea_login="adminflipper",
|
|
display_name="Admin Flipper", role="admin",
|
|
)
|
|
|
|
r = client.post(
|
|
f"/api/admin/users/{pending_id}/permission",
|
|
json={"state": "granted"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["permission_state"] == "granted"
|
|
assert body["changed"] is True
|
|
|
|
# Row reflects the new state + decision stamp.
|
|
row = db.conn().execute(
|
|
"SELECT permission_state, permission_decided_by, permission_decided_at "
|
|
"FROM users WHERE id = ?",
|
|
(pending_id,),
|
|
).fetchone()
|
|
assert row["permission_state"] == "granted"
|
|
assert row["permission_decided_by"] == 40
|
|
assert row["permission_decided_at"] is not None
|
|
|
|
# Audit row landed in permission_events.
|
|
events = db.conn().execute(
|
|
"""
|
|
SELECT actor_user_id, subject_user_id, event_kind
|
|
FROM permission_events
|
|
WHERE event_kind = 'permission_granted'
|
|
"""
|
|
).fetchall()
|
|
assert len(events) == 1
|
|
assert events[0]["actor_user_id"] == 40
|
|
assert events[0]["subject_user_id"] == pending_id
|
|
|
|
|
|
def test_permission_flip_revoke_promotes_granted_to_revoked(app_with_fake_gitea):
|
|
"""Revoke is the symmetric gesture. Used when an account earned a
|
|
grant then later lost it (§6.1 / `revoked` state)."""
|
|
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=50, login="goner", role="contributor")
|
|
# Default permission_state is 'granted' via the column default.
|
|
provision_user_row(user_id=51, login="adminrevoker", role="admin")
|
|
sign_in_as(
|
|
client, user_id=51, gitea_login="adminrevoker",
|
|
display_name="Admin Revoker", role="admin",
|
|
)
|
|
|
|
r = client.post(
|
|
"/api/admin/users/50/permission",
|
|
json={"state": "revoked"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
|
|
row = db.conn().execute(
|
|
"SELECT permission_state FROM users WHERE id = 50"
|
|
).fetchone()
|
|
assert row["permission_state"] == "revoked"
|
|
|
|
events = db.conn().execute(
|
|
"SELECT event_kind FROM permission_events "
|
|
"WHERE event_kind = 'permission_revoked' AND subject_user_id = 50"
|
|
).fetchall()
|
|
assert len(events) == 1
|
|
|
|
|
|
def test_permission_flip_refuses_self(app_with_fake_gitea):
|
|
"""Symmetric to set_mute / set_role: an admin can't self-flip.
|
|
The state-change channel for one's own grant is somebody else's
|
|
hand."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
provision_user_row(user_id=60, login="selfflipper", role="admin")
|
|
sign_in_as(
|
|
client, user_id=60, gitea_login="selfflipper",
|
|
display_name="Self Flipper", role="admin",
|
|
)
|
|
r = client.post(
|
|
"/api/admin/users/60/permission",
|
|
json={"state": "revoked"},
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
|
|
def test_permission_flip_refuses_non_admin(app_with_fake_gitea):
|
|
"""The endpoint is admin-only (§17 admin/* requires require_admin).
|
|
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=70, login="target", role="contributor")
|
|
provision_user_row(user_id=71, login="contrib", role="contributor")
|
|
sign_in_as(
|
|
client, user_id=71, gitea_login="contrib",
|
|
display_name="Contrib", role="contributor",
|
|
)
|
|
r = client.post(
|
|
"/api/admin/users/70/permission",
|
|
json={"state": "granted"},
|
|
)
|
|
assert r.status_code == 403
|
|
|
|
client.cookies.clear()
|
|
r = client.post(
|
|
"/api/admin/users/70/permission",
|
|
json={"state": "granted"},
|
|
)
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_permission_flip_refuses_invalid_state(app_with_fake_gitea):
|
|
"""Pydantic regex pattern refuses anything outside the three
|
|
canonical states with 422."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
provision_user_row(user_id=80, login="targetx", role="contributor")
|
|
provision_user_row(user_id=81, login="adminx", role="admin")
|
|
sign_in_as(
|
|
client, user_id=81, gitea_login="adminx",
|
|
display_name="Admin X", role="admin",
|
|
)
|
|
r = client.post(
|
|
"/api/admin/users/80/permission",
|
|
json={"state": "banished"},
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
|
|
def test_permission_flip_no_op_when_state_already_matches(app_with_fake_gitea):
|
|
"""An admin flipping a granted user to granted gets 200 with
|
|
`changed: false` — no audit row, no decided_at update."""
|
|
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=90, login="alreadygranted", role="contributor")
|
|
provision_user_row(user_id=91, login="adminN", role="admin")
|
|
sign_in_as(
|
|
client, user_id=91, gitea_login="adminN",
|
|
display_name="Admin N", role="admin",
|
|
)
|
|
|
|
before_events = db.conn().execute(
|
|
"SELECT COUNT(*) AS n FROM permission_events"
|
|
).fetchone()["n"]
|
|
|
|
r = client.post(
|
|
"/api/admin/users/90/permission",
|
|
json={"state": "granted"},
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["changed"] is False
|
|
|
|
after_events = db.conn().execute(
|
|
"SELECT COUNT(*) AS n FROM permission_events"
|
|
).fetchone()["n"]
|
|
assert after_events == before_events
|