PacketStation/tests/server/test_admin_roles.py

"""Every phase 9 Admin endpoint as every role: an Admin password session gets through, an Admin PIN session and every
other role get 403, and no session gets 401."""

from __future__ import annotations

from collections.abc import Iterator
from typing import Any

import pytest
from fakedir import FakeDirectory
from fastapi.testclient import TestClient
from pshelpers import ORIGIN_HEADERS, app_conn, needs_pg, pin_signin, register_device, set_pin, sign_in

from packetstation.domain.checklists import DEFAULT_TEMPLATES
from packetstation.main import create_app
from packetstation.settings import Settings

pytestmark = needs_pg

OTHER_ROLES = ("clerk.clara", "clerk.leo", "wh.wade", "qc.quinn", "pm.paula", "weld.andy", "lead.lou", "sup.wes")


def tiny_pdf(text: str = "No stamp here") -> bytes:
    import io

    from reportlab.pdfgen.canvas import Canvas

    buf = io.BytesIO()
    c = Canvas(buf)
    c.drawString(40, 40, text)
    c.save()
    return buf.getvalue()


@pytest.fixture(autouse=True)
def restore_group_map() -> Iterator[None]:
    """Adding a department rebuilds the process-wide group map; put the seed map back afterwards."""
    from packetstation.auth import groups

    yield
    groups.use_departments(groups.DEPARTMENTS)


def endpoints(andy: str, ck_version: int) -> list[tuple[str, str, dict[str, Any], int]]:
    """(method, path, request kwargs, the Admin's status)."""
    return [
        ("GET", "/api/admin/departments", {}, 200),
        ("PATCH", "/api/admin/departments/03", {"json": {"name": "Weld Parts"}}, 200),
        (
            "POST",
            "/api/admin/departments",
            {"json": {"code": "15", "slug": "Machining", "name": "Machining", "kind": "shop", "stations": ["Mill 1"]}},
            201,
        ),
        ("GET", "/api/admin/users", {}, 200),
        ("PATCH", f"/api/admin/users/{andy}/profile", {"json": {"welder_stamp": "A12"}}, 200),
        ("POST", "/api/admin/users/import", {"json": {"csv": "employee_id,welder_stamp\n3003,A13\n"}}, 200),
        ("GET", "/api/admin/checklists", {}, 200),
        ("PUT", "/api/admin/checklists", {"json": {"value": DEFAULT_TEMPLATES, "version": ck_version}}, 200),
        ("GET", "/api/admin/security/settings", {}, 200),
        ("PUT", "/api/admin/security/settings", {"json": {"values": {}, "version": 0}}, 200),
        ("GET", "/api/admin/security/alerts", {}, 200),
        ("GET", "/api/admin/security/integrity", {}, 200),
        ("GET", "/api/admin/access-log", {}, 200),
        ("GET", "/api/admin/access-log.csv", {}, 200),
        ("GET", "/api/admin/presence", {}, 200),
        ("GET", "/api/admin/directory-flags", {}, 200),
        ("POST", "/api/admin/ad/refresh", {}, 202),
        ("GET", "/api/admin/ad/reconcile", {}, 200),
    ]


def call(client: TestClient, method: str, path: str, kw: dict[str, Any]) -> Any:
    return client.request(method, path, **kw)


def test_every_new_admin_endpoint_as_every_role(
    pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory
) -> None:
    andy = str(fake_dir.user("weld.andy").guid)
    sign_in(pg_client, "weld.andy")  # the user row exists
    pg_client.post("/api/auth/signout")
    with app_conn(pg_settings) as conn:
        row = conn.execute("SELECT version FROM settings WHERE key = 'checklist_templates'").fetchone()
    eps = endpoints(andy, int(row[0]) if row else 0)

    # Anonymous
    with TestClient(create_app(pg_settings, directory=fake_dir), headers=ORIGIN_HEADERS) as anon:
        for method, path, kw, _ in eps:
            r = call(anon, method, path, kw)
            assert r.status_code == 401, (method, path, r.status_code)

    # Every other role
    for sam in OTHER_ROLES:
        assert sign_in(pg_client, sam).status_code == 200, sam
        for method, path, kw, _ in eps:
            r = call(pg_client, method, path, kw)
            assert r.status_code == 403 and r.json()["code"] == "forbidden", (sam, method, path, r.status_code)
        pg_client.post("/api/auth/signout")

    # An Admin on a PIN session
    register_device(pg_client)
    set_pin(pg_client, "admin.ann")
    assert pin_signin(pg_client, "1001").status_code == 200
    for method, path, kw, _ in eps:
        r = call(pg_client, method, path, kw)
        assert r.status_code == 403 and r.json()["code"] == "password_session_required", (method, path)
    pg_client.post("/api/auth/signout")

    # An Admin on a password session
    assert sign_in(pg_client, "admin.ann").status_code == 200
    for method, path, kw, status in eps:
        r = call(pg_client, method, path, kw)
        assert r.status_code == status, (method, path, r.status_code, r.text)

    # Every refusal to a signed-in person was audited once per path (no bodies)
    with app_conn(pg_settings) as conn:
        n = conn.execute(
            "SELECT count(*) FROM audit_events WHERE action = 'policy.denied' AND detail->>'code' = 'forbidden'"
        ).fetchone()[0]
        assert n == len(OTHER_ROLES) * len(eps)
        details = conn.execute("SELECT detail FROM audit_events WHERE action = 'policy.denied' LIMIT 1").fetchone()[0]
        assert set(details) == {"method", "path", "code"}


def stamp_endpoints() -> list[tuple[str, str, dict[str, Any]]]:
    out = []
    for base in ("/api/stamp-verify", "/api/admin/stamp-verify"):
        out.append(("POST", base, {"json": {"text": "0123456789abcdef"}}))
        out.append(("POST", f"{base}/pdf", {"files": {"file": ("a.pdf", tiny_pdf(), "application/pdf")}}))
    return out


def test_stamp_check_roles(pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory) -> None:
    """Change C9: the stamp check is open to Admin, Q.C. and every Clerical level, at both paths, on a password
    session. Every other role gets 403."""
    eps = stamp_endpoints()
    with TestClient(create_app(pg_settings, directory=fake_dir), headers=ORIGIN_HEADERS) as anon:
        for method, path, kw in eps:
            assert call(anon, method, path, kw).status_code == 401, path
    for sam in ("admin.ann", "qc.quinn", "clerk.clara", "clerk.leo"):
        assert sign_in(pg_client, sam).status_code == 200, sam
        for method, path, kw in eps:
            r = call(pg_client, method, path, kw)
            assert r.status_code == 200, (sam, path, r.status_code, r.text)
        pg_client.post("/api/auth/signout")
    for sam in ("wh.wade", "pm.paula", "weld.andy", "lead.lou", "sup.wes"):
        assert sign_in(pg_client, sam).status_code == 200, sam
        for method, path, kw in eps:
            r = call(pg_client, method, path, kw)
            assert r.status_code == 403 and r.json()["code"] == "forbidden", (sam, path, r.status_code)
        pg_client.post("/api/auth/signout")
    # A PIN session: the password session rule stays
    register_device(pg_client)
    set_pin(pg_client, "admin.ann")
    assert pin_signin(pg_client, "1001").status_code == 200
    r = call(pg_client, eps[0][0], eps[0][1], eps[0][2])
    assert r.status_code == 403 and r.json()["code"] == "password_session_required", r.text