"""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 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_counts_and_flags(pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory) -> None: sign_in(pg_client, "admin.ann") res = pg_client.post("/api/admin/ad/refresh").json() assert res == { "groups_checked": 49, "groups_missing": ["PS-14-Electrical-Lead"], "users_seen": 16, "created": 15, "updated": 1, "deactivated": 0, "emp_missing": ["noemp.nora"], "emp_duplicate": ["9001"], } users = {u["sam_account"]: u for u in pg_client.get("/api/admin/users").json()} assert "norole.nick" not in users and len(users) == 16 assert users["dup.dave"]["flags"]["emp_duplicate"] and users["dup.dina"]["flags"]["emp_duplicate"] assert users["noemp.nora"]["flags"]["emp_missing"] and users["noemp.nora"]["emp_no"] is None assert users["disabled.dan"]["active"] is False and users["disabled.dan"]["flags"]["disabled"] assert users["nested.ned"]["roles"] == ["shop:10:worker"] assert users["both.bo"]["roles"] == ["shop:03:supervisor"] assert users["multi.mia"]["roles"] == ["clerical", "shop:06:supervisor"] assert [u["sam_account"] for u in pg_client.get("/api/admin/users", params={"q": "dup"}).json()] == [ "dup.dave", "dup.dina", ] # Clara signs in elsewhere, then AD disables her; the next refresh deactivates her and ends the session with TestClient(create_app(pg_settings, directory=fake_dir), headers=ORIGIN_HEADERS) as other: sign_in(other, "clerk.clara") fake_dir.user("clerk.clara").disabled = True fake_dir.users.remove(fake_dir.user("wh.wade")) # gone from AD fake_dir.user("qc.quinn").groups = [] # still in AD, in no group res = pg_client.post("/api/admin/ad/refresh").json() assert res["deactivated"] == 2 and res["users_seen"] == 14 assert other.get("/api/auth/me").json()["code"] == "no_session" users = {u["sam_account"]: u for u in pg_client.get("/api/admin/users").json()} assert users["clerk.clara"]["active"] is False and users["wh.wade"]["active"] is False assert users["qc.quinn"]["active"] is True and users["qc.quinn"]["roles"] == [] with app_conn(pg_settings) as conn: 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", "ad.user_deactivated", "ad.user_deactivated", "ad.refresh"] def test_user_detail_and_lookup(pg_client: TestClient, fake_dir: FakeDirectory) -> None: sign_in(pg_client, "admin.ann") pg_client.post("/api/admin/ad/refresh") 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)