"""Admin routes: users, AD view, PIN clearing, devices, group map, AD refresh, attempts.""" from __future__ import annotations from fakedir import FAKE_PASSWORD, FakeDirectory from fastapi.testclient import TestClient from pshelpers import ORIGIN_HEADERS, PIN, app_conn, needs_pg, pin_signin, register_device, set_pin, sign_in from psycopg.types.json import Jsonb from packetstation.main import create_app from packetstation.settings import Settings pytestmark = needs_pg def test_admin_only(pg_client: TestClient) -> None: sign_in(pg_client, "clerk.clara") for path in ("/api/admin/users", "/api/admin/devices", "/api/admin/groups", "/api/admin/signin-attempts"): r = pg_client.get(path) assert r.status_code == 403 and r.json()["code"] == "forbidden", path assert pg_client.post("/api/admin/ad/refresh").status_code == 403 def test_group_map_flags_missing_and_extra(pg_client: TestClient, fake_dir: FakeDirectory) -> None: sign_in(pg_client, "admin.ann") view = pg_client.get("/api/admin/groups").json() assert view["ad_reachable"] is True and view["checked_at"] assert len(view["groups"]) == 50 missing = [g["name"] for g in view["groups"] if g["in_ad"] is False] assert missing == ["PS-14-Electrical-Lead"] assert view["extra_in_ou"] == ["PS-Extra-Legacy"] first = view["groups"][0] assert first == {"name": "PS-Admins", "role": "admin", "dept_code": None, "in_ad": True} weld = next(g for g in view["groups"] if g["name"] == "PS-03-WeldParts-Lead") assert weld["role"] == "shop:03:lead" and weld["dept_code"] == "03" def test_ad_refresh_requests_a_reconcile_run(pg_client: TestClient, pg_settings: Settings) -> None: """POST /ad/refresh asks the worker for an ad-reconcile run; the counts are the reconcile's (test_reconcile).""" sign_in(pg_client, "admin.ann") r = pg_client.post("/api/admin/ad/refresh") assert r.status_code == 202, r.text assert r.json() == {"requested": True, "pending": 1, "job": "ad-reconcile"} assert pg_client.post("/api/admin/ad/refresh").json()["pending"] == 2 empty = pg_client.get("/api/admin/ad/reconcile").json() assert empty["run"] is None and empty["pending"] == 2 assert empty["report"] == { "groups_missing": [], "groups_extra": [], "users_seen": 0, "created": 0, "updated": 0, "deactivated": 0, "sessions_ended": 0, "emp_missing": [], "emp_duplicate": [], } with app_conn(pg_settings) as conn: assert conn.execute("SELECT count(*) FROM job_requests WHERE job = 'ad-reconcile'").fetchone()[0] == 2 acts = conn.execute("SELECT action FROM audit_events WHERE action LIKE 'ad.%' ORDER BY id").fetchall() assert [a[0] for a in acts] == ["ad.refresh_requested", "ad.refresh_requested"] conn.execute( "INSERT INTO job_runs (job, trigger, finished_at, status, counts) VALUES ('ad-reconcile', 'request'," " now(), 'ok', %s)", (Jsonb({"users_seen": 16, "created": 15, "groups_missing": ["PS-14-Electrical-Lead"], "x": 1}),), ) conn.execute("INSERT INTO job_runs (job, trigger, status) VALUES ('ad-reconcile', 'schedule', 'running')") got = pg_client.get("/api/admin/ad/reconcile").json() assert got["run"]["status"] == "ok" and got["run"]["trigger"] == "request" assert got["report"]["users_seen"] == 16 and got["report"]["created"] == 15 assert got["report"]["groups_missing"] == ["PS-14-Electrical-Lead"] and "x" not in got["report"] def test_user_detail_and_lookup(pg_client: TestClient, fake_dir: FakeDirectory) -> None: sign_in(pg_client, "multi.mia") # the user row exists pg_client.post("/api/auth/signout") sign_in(pg_client, "admin.ann") mia = next(u for u in pg_client.get("/api/admin/users", params={"q": "mia"}).json()) detail = pg_client.get(f"/api/admin/users/{mia['id']}").json() assert detail["groups"] == ["PS-06-CodeBldg2-Supervisor", "PS-Clerical"] and detail["groups_source"] == "ad" assert detail["note"] == "Roles come from AD groups. Ask IT to change them." and detail["sessions"] == [] fake_dir.down = True cached = pg_client.get(f"/api/admin/users/{mia['id']}").json() assert cached["groups_source"] == "cache" and set(cached["groups"]) == set(detail["groups"]) assert pg_client.get("/api/admin/ad/lookup", params={"login": "mia"}).status_code == 503 fake_dir.down = False found = pg_client.get("/api/admin/ad/lookup", params={"login": "norole.nick"}).json() assert found["found"] is True and found["roles"] == [] and found["user"]["emp_no"] == "8001" assert pg_client.get("/api/admin/ad/lookup", params={"login": "ghost"}).json()["found"] is False assert pg_client.get("/api/admin/users/00000000-0000-0000-0000-000000000000").status_code == 404 def test_clear_pin(pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory) -> None: set_pin(pg_client, "weld.andy") andy = str(fake_dir.user("weld.andy").guid) sign_in(pg_client, "admin.ann") r = pg_client.request("DELETE", f"/api/admin/users/{andy}/pin", json={"reason": " "}) assert r.status_code == 400 and r.json()["code"] == "reason_required" r = pg_client.request("DELETE", f"/api/admin/users/{andy}/pin", json={"reason": "Forgot it, ticket 4411"}) assert r.status_code == 204 assert pg_client.request("DELETE", f"/api/admin/users/{andy}/pin", json={"reason": "again"}).status_code == 404 with app_conn(pg_settings) as conn: row = conn.execute( "SELECT user_id, subject_id, it_reason FROM audit_events WHERE action = 'pin.cleared'" ).fetchone() assert str(row[0]) == str(fake_dir.user("admin.ann").guid) and row[1] == andy assert row[2] == "Forgot it, ticket 4411" sign_in(pg_client, "weld.andy") assert pg_client.get("/api/account").json()["pin"] == {"set": False, "disabled": False} assert any(e["action"] == "pin.cleared" for e in pg_client.get("/api/account/access-log").json()) def test_devices_and_revoke_ends_pin_sessions( pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory ) -> None: device = register_device(pg_client) set_pin(pg_client, "admin.ann") assert pin_signin(pg_client, "1001").status_code == 200 # Admin is a password session only r = pg_client.get("/api/admin/users") assert r.status_code == 403 and r.json()["code"] == "password_session_required" with TestClient(create_app(pg_settings, directory=fake_dir), headers=ORIGIN_HEADERS) as admin: sign_in(admin, "admin.ann") listed = admin.get("/api/admin/devices").json() assert len(listed) == 1 and listed[0]["id"] == device["id"] and listed[0]["active_sessions"] == 1 assert listed[0]["registered_by"]["display_name"] == "Wes Super" assert admin.delete(f"/api/admin/devices/{device['id']}").status_code == 204 assert admin.get("/api/admin/devices").json()[0]["revoked_at"] is not None assert admin.get("/api/auth/me").status_code == 200 # the password session is not on the device assert pg_client.get("/api/auth/me").json()["code"] == "no_session" assert pg_client.get("/api/auth/context").json()["device"] is None r = pin_signin(pg_client, "1001") assert r.status_code == 403 and r.json()["code"] == "device_required" with app_conn(pg_settings) as conn: assert conn.execute("SELECT revoked_reason FROM sessions WHERE kind = 'pin'").fetchone()[0] == "device_revoked" assert conn.execute("SELECT count(*) FROM audit_events WHERE action = 'device.revoked'").fetchone()[0] == 1 def test_signin_attempts_list(pg_client: TestClient) -> None: sign_in(pg_client, "clerk.clara", "wrong") sign_in(pg_client, "admin.ann", FAKE_PASSWORD) rows = pg_client.get("/api/admin/signin-attempts").json() assert len(rows) == 1 and rows[0]["account"] == "clerk.clara" and rows[0]["reason"] == "bad_password" assert set(rows[0]) == {"id", "at", "account", "ip", "kind", "reason", "user_id", "display_name"} assert FAKE_PASSWORD not in str(rows) and PIN not in str(rows)