PacketStation/tests/server/test_admin_api.py

"""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)