spb/hfmarketdata

1"""Admin API: session-only access, invite flow → accept → login, tier/status updates, keys, last-admin guard,2user deletion, global usage, audit filters."""3from __future__ import annotations45import os6from urllib.parse import parse_qs, urlparse78from fastapi.testclient import TestClient910JSON = {"Content-Type": "application/json"}111213def token_of(link: str) -> str:14    return parse_qs(urlparse(link).query)["token"][0]151617def fresh_email(tag: str) -> str:18    return f"{tag}-{os.getpid()}-{os.urandom(3).hex()}@example.com"192021def test_admin_requires_session_and_role(client, make_user, admin_web):22    """A6: an admin's API key does NOT open the admin area; plain users get FORBIDDEN."""23    _, key, _, _ = make_user()24    r = client.get("/v1/admin/users", headers={"Authorization": f"Bearer {key}"})25    assert r.status_code == 403 and r.json()["error"]["code"] == "SESSION_REQUIRED"26    assert client.get("/v1/admin/users").status_code == 40127    _, admin_key, _, _ = make_user(role="admin")28    r = client.get("/v1/admin/users", headers={"Authorization": f"Bearer {admin_key}"})29    assert r.status_code == 403 and r.json()["error"]["code"] == "SESSION_REQUIRED"30    web, _, _, _ = admin_web31    assert web.get("/v1/admin/users").status_code == 20032    # a signed-in plain user → FORBIDDEN33    _, _, email, pw = make_user(with_key=False)34    plain = TestClient(web.app)35    plain.post("/v1/auth/login", json={"email": email, "password": pw}, headers=JSON)36    r = plain.get("/v1/admin/users")37    assert r.status_code == 403 and r.json()["error"]["code"] == "FORBIDDEN"383940def test_invite_flow(admin_web):41    web, _, _, _ = admin_web42    email = fresh_email("tim")43    r = web.post("/v1/admin/users", json={"email": email, "name": "Tim", "tier": "high_usage"}, headers=JSON)44    assert r.status_code == 201, r.text45    d = r.json()["data"]46    assert d["user"]["status"] == "invited" and d["user"]["tier"] == "high_usage" and d["user"]["keys_active"] == 147    assert d["key"]["prefix"].startswith("hfmd_live_") and "key" not in d["key"]   # admins never see the full key here48    assert d["invitation"]["delivered"] is False and d["invitation"]["queued"] is False49    assert "/accept-invite?token=" in d["invitation"]["link"]50    uid = d["user"]["id"]51    # duplicate invite → 40952    r = web.post("/v1/admin/users", json={"email": email, "name": "Tim"}, headers=JSON)53    assert r.status_code == 409 and r.json()["error"]["code"] == "EMAIL_TAKEN"54    # detail shows the pending link, the keys and a 7-day usage series55    r = web.get(f"/v1/admin/users/{uid}")56    assert r.status_code == 20057    det = r.json()["data"]58    assert det["pending_invite_link"] == d["invitation"]["link"] and len(det["keys"]) == 159    assert det["usage"]["range"] == "7d" and len(det["usage"]["points"]) == 168 and det["keys"][0]["scopes"] == ["data"]60    # user cannot log in yet (no password)61    guest = TestClient(web.app)62    r = guest.post("/v1/auth/login", json={"email": email, "password": "whatever-long-enough"}, headers=JSON)63    assert r.status_code == 40164    # re-send voids the first token65    r = web.post(f"/v1/admin/users/{uid}/invite", json={}, headers=JSON)66    assert r.status_code == 20067    link2 = r.json()["data"]["invitation"]["link"]68    r = guest.post("/v1/auth/accept-invite", json={"token": token_of(d["invitation"]["link"]), "password": "tim-passphrase-123"}, headers=JSON)69    assert r.status_code == 400 and r.json()["error"]["code"] == "INVALID_TOKEN"70    # accept with the fresh one → active + session71    r = guest.post("/v1/auth/accept-invite", json={"token": token_of(link2), "password": "tim-passphrase-123"}, headers=JSON)72    assert r.status_code == 200 and r.json()["data"]["status"] == "active" and r.json()["data"]["email_verified"] is True73    me = guest.get("/v1/me").json()["data"]74    assert me["user"]["email"] == email and me["limits"]["tier"] == "high_usage" and me["user"]["keys_active"] == 175    guest.cookies.clear()76    r = guest.post("/v1/auth/login", json={"email": email, "password": "tim-passphrase-123"}, headers=JSON)77    assert r.status_code == 20078    guest.cookies.clear()79    # signing up again with an invited-then-active address → uniform 202, nothing changed80    r = guest.post("/v1/auth/signup", json={"email": email, "name": "Tim", "password": "another-passphrase"}, headers=JSON)81    assert r.status_code == 202 and "debug_link" not in r.json()["data"]82    assert guest.post("/v1/auth/login", json={"email": email, "password": "tim-passphrase-123"}, headers=JSON).status_code == 200838485def test_signup_on_invited_address_resends_invitation(admin_web):86    web, _, _, _ = admin_web87    email = fresh_email("luca")88    web.post("/v1/admin/users", json={"email": email, "name": "Luca"}, headers=JSON)89    guest = TestClient(web.app)90    r = guest.post("/v1/auth/signup", json={"email": email, "name": "Luca", "password": "luca-passphrase-123"}, headers=JSON)91    assert r.status_code == 202 and r.json()["data"]["status"] == "verification_sent"     # uniform wording92    assert "/accept-invite?token=" in r.json()["data"]["debug_link"]                       # …but it IS the invitation939495def test_create_active_user_update_tier_and_disable(admin_web):96    web, _, _, _ = admin_web97    email = fresh_email("ada")98    r = web.post("/v1/admin/users", json={"email": email, "name": "Ada", "password": "ada-passphrase-123"}, headers=JSON)99    assert r.status_code == 201 and r.json()["data"]["user"]["status"] == "active" and r.json()["data"]["invitation"] is None100    uid = r.json()["data"]["user"]["id"]101    # admin creates a key for her (shown once) with a lifetime, it honours the user tier then the override102    r = web.post(f"/v1/admin/users/{uid}/keys", json={"name": "handover", "expires_in_days": 90, "note": "handed over by mail"}, headers=JSON)103    assert r.status_code == 201 and r.json()["data"]["expires_at"] and r.json()["data"]["note"] == "handed over by mail"104    key = r.json()["data"]["key"]105    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).headers["X-RateLimit-Limit-Requests"] == "120"106    r = web.patch(f"/v1/admin/users/{uid}", json={"tier": "high_usage"}, headers=JSON)107    assert r.status_code == 200 and r.json()["data"]["tier"] == "high_usage"108    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).headers["X-RateLimit-Limit-Requests"] == "600"109    r = web.patch(f"/v1/admin/users/{uid}", json={"tier": "platinum"}, headers=JSON)110    assert r.status_code == 422111    # disable → key refused, login refused112    r = web.patch(f"/v1/admin/users/{uid}", json={"status": "disabled"}, headers=JSON)113    assert r.status_code == 200114    r = web.get("/v1/status", headers={"Authorization": f"Bearer {key}"})115    assert r.status_code == 403 and r.json()["error"]["code"] == "ACCOUNT_DISABLED"116    guest = TestClient(web.app)117    r = guest.post("/v1/auth/login", json={"email": email, "password": "ada-passphrase-123"}, headers=JSON)118    assert r.status_code == 403 and r.json()["error"]["code"] == "ACCOUNT_DISABLED"119    assert web.post(f"/v1/admin/users/{uid}/reset-password", json={}, headers=JSON).status_code == 409120    # re-enable and revoke via admin121    web.patch(f"/v1/admin/users/{uid}", json={"status": "active"}, headers=JSON)122    kid = web.get(f"/v1/admin/users/{uid}").json()["data"]["keys"][0]["id"]123    r = web.delete(f"/v1/admin/users/{uid}/keys/{kid}", headers=JSON)124    assert r.status_code == 200 and r.json()["data"]["status"] == "revoked"125    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).status_code == 401126    # reset link for the user127    r = web.post(f"/v1/admin/users/{uid}/reset-password", json={}, headers=JSON)128    assert r.status_code == 200 and "/reset-password?token=" in r.json()["data"]["reset"]["link"]129    assert web.get(f"/v1/admin/users/{uid}").json()["data"]["pending_reset_link"] == r.json()["data"]["reset"]["link"]130    assert web.get("/v1/admin/users/999999").json()["error"]["code"] == "USER_NOT_FOUND"131132133def test_last_admin_guard_and_self_protection(admin_web, make_user):134    """A19."""135    from accounts.models import User136    from core.db import session137    from sqlalchemy import select, update138    web, admin_id, _, _ = admin_web139    # an admin cannot touch their own role / status140    r = web.patch(f"/v1/admin/users/{admin_id}", json={"role": "user"}, headers=JSON)141    assert r.status_code == 403 and r.json()["error"]["code"] == "FORBIDDEN"142    assert web.patch(f"/v1/admin/users/{admin_id}", json={"status": "disabled"}, headers=JSON).status_code == 403143    assert web.patch(f"/v1/admin/users/{admin_id}", json={"name": "Root"}, headers=JSON).status_code == 200144    assert web.delete(f"/v1/admin/users/{admin_id}", headers=JSON).status_code == 403145    # make our admin the ONLY active admin, then try to demote another (now demoted) admin… and ourselves through the service146    other_id, _, _, _ = make_user(role="admin", with_key=False)147    with session() as s:148        s.execute(update(User).where(User.role == "admin", User.status == "active", User.id.notin_([admin_id, other_id])).values(role="user"))149    r = web.patch(f"/v1/admin/users/{other_id}", json={"role": "user"}, headers=JSON)150    assert r.status_code == 200                                    # two admins → demoting one is fine151    assert web.get("/v1/admin/users?limit=1").json()["meta"]["admins_active"] == 1152    from accounts import service153    from core.errors import ApiError154    with session() as s:155        me = s.get(User, admin_id)156        for kwargs in ({"role": "user"}, {"status": "disabled"}):157            try:158                service.update_user(s, me, actor="test", **kwargs)159                assert False, "last admin must be protected"160            except ApiError as exc:161                assert exc.code == "LAST_ADMIN" and exc.status == 409162        try:163            service.delete_account(s, me, actor="test")164            assert False165        except ApiError as exc:166            assert exc.code == "LAST_ADMIN"167        s.rollback()168    # promoting someone lifts the guard169    web.patch(f"/v1/admin/users/{other_id}", json={"role": "admin"}, headers=JSON)170    assert web.get("/v1/admin/users?limit=1").json()["meta"]["admins_active"] == 2171172173def test_role_change_signs_the_user_out(admin_web, make_user, signin):174    """A11: bumping session_version on role change."""175    web, _, _, _ = admin_web176    uid, _, email, pw = make_user(with_key=False)177    user_web = TestClient(web.app)178    signin(user_web, email, pw)179    assert user_web.get("/v1/me").status_code == 200180    assert web.patch(f"/v1/admin/users/{uid}", json={"role": "admin"}, headers=JSON).status_code == 200181    assert user_web.get("/v1/me").status_code == 401182183184def test_admin_delete_user(admin_web, make_user, signin):185    web, _, _, _ = admin_web186    uid, key, email, pw = make_user()187    user_web = TestClient(web.app)188    signin(user_web, email, pw)189    r = web.delete(f"/v1/admin/users/{uid}", headers=JSON)190    assert r.status_code == 200 and r.json()["data"]["status"] == "deleted" and r.json()["data"]["email"].endswith("@deleted.invalid")191    assert user_web.get("/v1/me").status_code == 401192    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).status_code == 401193    assert web.patch(f"/v1/admin/users/{uid}", json={"tier": "high_usage"}, headers=JSON).status_code == 409194    assert web.post(f"/v1/admin/users/{uid}/invite", json={}, headers=JSON).status_code == 409195    assert web.get(f"/v1/admin/users?status=deleted").json()["data"][0]["status"] == "deleted"196197198def test_list_users_pagination_and_search(admin_web):199    web, _, _, _ = admin_web200    emails = [fresh_email("page") for _ in range(3)]201    for e in emails:202        web.post("/v1/admin/users", json={"email": e, "name": "Paged Person"}, headers=JSON)203    r = web.get("/v1/admin/users?search=paged&limit=2")204    assert r.status_code == 200205    body = r.json()206    assert len(body["data"]) == 2 and body["meta"]["next_cursor"] and "admins_active" in body["meta"]207    r2 = web.get(f"/v1/admin/users?search=paged&limit=2&cursor={body['meta']['next_cursor']}")208    assert r2.status_code == 200 and len(r2.json()["data"]) >= 1209    ids = {u["id"] for u in body["data"]} | {u["id"] for u in r2.json()["data"]}210    assert len(ids) >= 3211    assert web.get("/v1/admin/users?limit=0").status_code == 422212213214def test_global_usage_and_audit(admin_web, make_user):215    web, admin_id, _, _ = admin_web216    uid, key, _, _ = make_user()217    web.get("/v1/_test/frame?n=100", headers={"Authorization": f"Bearer {key}"})218    import time219220    from ratelimit import usage221    usage.fold(now_s=time.time() + 120)     # force the current minute to fold222    r = web.get("/v1/admin/usage?days=7&top=5")223    assert r.status_code == 200224    d = r.json()["data"]225    assert d["per_day"] and d["per_day"][-1]["requests"] >= 1 and d["totals"]["requests"] >= 1226    top = [t for t in d["top"] if t["user"] and t["user"]["id"] == uid]227    assert top and top[0]["rows"] >= 100 and top[0]["user"]["email"]228    r = web.get("/v1/admin/audit?limit=5")229    assert r.status_code == 200230    rows = r.json()["data"]231    assert rows and {"id", "ts", "actor", "action", "target", "meta"} <= set(rows[0])232    r = web.get("/v1/admin/audit?action=key.")233    assert all(x["action"].startswith("key.") for x in r.json()["data"])234    # B6: filters by actor / target235    r = web.get(f"/v1/admin/audit?actor=user:{admin_id}")236    assert r.json()["data"] and all(x["actor"] == f"user:{admin_id}" for x in r.json()["data"])237    r = web.get(f"/v1/admin/audit?target=user:{uid}")238    assert r.json()["data"] and all(x["target"] == f"user:{uid}" for x in r.json()["data"])239    assert web.get("/v1/admin/audit?target=user:0").json()["data"] == []240