Phase 2: core domain, policy engine, packet API, Clerical screen

Migration 0003: projects, aliases, renumbers, units, packets and every child table, change orders, assets, documents, ITPs, station limits, department weld flag. Policy package: Principal, PacketState, Action rules, custody, preconditions, CO hold, order of checks, can/blocked/holder on reads. docs/POLICY.md with the tightenings table for Seth. Write pipeline: FOR UPDATE, CAS on status and dept, If-Match row versions, server-assigned numbers, field CAS, version bump, packet log, audit, NOTIFY, signatures, IT reasons, filed lock and HMAC seal. API: /api/packets (header, transitions, work orders, pages, docs, move, remove, BOM, shorts, welds, heat tags, heat table, stage, checklist, clock, warehouse stage, Q.C., change orders, log, seal), /api/projects, /api/units, /api/assets, /_blob. Frontend: Clerical home, projects and units, packet page with pdf.js and QR codes, hand draft, heat map, send to warehouse, department or Q.C. Tests: policy matrix, API, concurrency, schema, projects; Playwright Clerical draft scenario. Verified on docker-1 with the CI stack and the Samba DC. Co-Authored-By: Claude Fable 5.1 <[email protected]>
2026-10-16 09:31:45 -06:00
parent 97b0973130
commit 37d9a0cb50
83 changed files with 16490 additions and 55 deletions
@@ -0,0 +1,630 @@
+"""The policy matrix: every role against every status for every transition, then the rules one by one.
+
+The expected table below is written from the rule table in the phase 2 design contract, not from the code.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+from types import MappingProxyType
+from typing import Any
+
+import pytest
+
+from packetstation.policy import (
+    RULES,
+    STATUSES,
+    Action,
+    PacketState,
+    Principal,
+    can,
+    check,
+    holder_name,
+    holds,
+    principal_from_roles,
+    works_on,
+)
+from packetstation.policy.preconditions import INPUT_ONLY, MESSAGES
+
+A = Action
+WELDER = "00000000-0000-0000-0000-00000000a11d"
+OTHER = "00000000-0000-0000-0000-0000000000b0"
+IT = "Ticket 4411: fix a typo"
+
+# Every input a step can need. The default state satisfies every state precondition.
+FULL_EXTRA: dict[str, Any] = {
+    "reason": "Needs another look at the fit-up.",
+    "station_id": 1,
+    "welder_id": WELDER,
+    "dept": "06",
+    "routed_to": "databook",
+    "book": True,
+    "spool": True,
+    "location": "Rack 4",
+}
+
+
+def P(roles: list[str], user_id: str = OTHER, limits: tuple[int, ...] = ()) -> Principal:
+    return principal_from_roles(user_id, roles, station_limits=limits)
+
+
+ROLES: dict[str, Principal] = {
+    "admin": P(["admin"]),
+    "clerical": P(["clerical"]),
+    "clerical_sup": P(["clerical_supervisor"]),
+    "warehouse": P(["warehouse"]),
+    "warehouse_sup": P(["warehouse_supervisor"]),
+    "qc": P(["qc"]),
+    "qc_sup": P(["qc_supervisor"]),
+    "pm": P(["pm"]),
+    "worker03": P(["shop:03:worker"]),
+    "lead03": P(["shop:03:lead"]),
+    "sup03": P(["shop:03:supervisor"]),
+    "worker06": P(["shop:06:worker"], user_id=WELDER),
+    # the assigned welder, but limited to a station that is not the packet's
+    "worker03_limited": P(["shop:03:worker"], user_id=WELDER, limits=(99,)),
+    "welder": P(["shop:03:worker"], user_id=WELDER),
+}
+
+
+def state(status: str, **kw: Any) -> PacketState:
+    base = PacketState(
+        id=1,
+        status=status,
+        dept="03",
+        station_id=1,
+        welder_id=WELDER,
+        wh_skipped=True,
+        routed_to="databook",
+        vclosed=True,
+        removed=False,
+        source_kind="hand",
+        weld_dept=True,
+        co_hold=False,
+        drawing_attached=True,
+        heat_map=True,
+        bom_total=3,
+        bom_ready=3,
+        welds=2,
+        checklist_complete=True,
+        filler_set=True,
+        open_shorts=1,
+    )
+    return dataclasses.replace(base, **kw)
+
+
+CL = {"clerical", "clerical_sup"}
+WH = {"warehouse", "warehouse_sup"}
+QC = {"qc", "qc_sup"}
+LEADS = {"lead03", "sup03"}
+SUP = {"sup03"}
+
+# action -> (from statuses, roles allowed). From the contract rule table.
+EXPECTED: dict[Action, tuple[set[str], set[str]]] = {
+    A.SEND_TO_WAREHOUSE: ({"draft"}, CL),
+    A.RELEASE_TO_SHOP: ({"draft"}, CL),
+    A.SEND_DRAFT_TO_QC: ({"draft"}, CL),
+    A.WH_SEND_BACK: ({"warehouse"}, WH),
+    A.RETURN_TO_CLERICAL: ({"ready"}, LEADS),
+    A.WH_RELEASE: ({"warehouse"}, WH),
+    A.SEND_BACK_TO_WAREHOUSE: ({"ready"}, LEADS),
+    A.ASSIGN: ({"ready"}, LEADS),
+    A.UNASSIGN: ({"assigned"}, LEADS),
+    A.PULL_BACK: ({"in_progress"}, LEADS),
+    A.START: ({"assigned"}, {*LEADS, "welder"}),
+    A.SUBMIT: ({"assigned", "in_progress"}, {*LEADS, "welder"}),
+    A.RETURN_TO_STATION: ({"review"}, SUP),
+    A.SEND_TO_QC: ({"review"}, SUP),
+    A.ROUTE_FROM_REVIEW: ({"review"}, SUP),
+    A.QC_REJECT: ({"qc"}, QC),
+    A.QC_ACCEPT: ({"qc"}, QC),
+    A.QC_ACCEPT_AND_ROUTE: ({"qc"}, QC),
+    A.FILE: ({"complete"}, CL),
+    A.ADMIN_REOPEN: ({"filed"}, {"admin"}),
+}
+
+
+def expected(role: str, status: str, action: Action) -> str:
+    """'' when the step passes, else the expected code."""
+    froms, allowed = EXPECTED[action]
+    if status not in froms:
+        return "wrong_status"
+    if role in allowed:
+        return ""
+    return "it_reason_required" if role == "admin" else "forbidden"
+
+
+MATRIX = [(r, s, a) for r in ROLES for s in STATUSES for a in EXPECTED]
+
+
+def test_expected_table_covers_every_user_transition() -> None:
+    transitions = {a for a, r in RULES.items() if r.to_status is not None and not r.system_only}
+    assert transitions == set(EXPECTED)
+
+
+@pytest.mark.parametrize(("role", "status", "action"), MATRIX, ids=[f"{r}-{s}-{a}" for r, s, a in MATRIX])
+def test_matrix(role: str, status: str, action: Action) -> None:
+    d = check(ROLES[role], state(status), action, extra=FULL_EXTRA)
+    want = expected(role, status, action)
+    assert d.ok is (want == ""), (d.code, d.why)
+    assert d.code == want
+    if d.ok:
+        assert d.signature == RULES[action].signature
+        assert d.admin_bypass is False
+
+
+# ---------- custody ----------
+
+HOLDERS: dict[str, set[str]] = {
+    "draft": {"clerical", "clerical_sup"},
+    "warehouse": {"warehouse", "warehouse_sup"},
+    "ready": {"lead03", "sup03"},
+    "assigned": {"lead03", "sup03", "worker03", "welder"},
+    "in_progress": {"lead03", "sup03", "worker03", "welder"},
+    "review": {"lead03", "sup03"},
+    "qc": {"qc", "qc_sup"},
+    "complete": {"clerical", "clerical_sup"},
+    "filed": set(),
+}
+
+
+@pytest.mark.parametrize("status", STATUSES)
+def test_custody_per_status(status: str) -> None:
+    got = {name for name, p in ROLES.items() if holds(p, state(status))}
+    assert got == HOLDERS[status]
+
+
+def test_holder_name() -> None:
+    assert holder_name(state("draft")) == "Clerical"
+    assert holder_name(state("warehouse")) == "Warehouse"
+    assert holder_name(state("qc")) == "Q.C."
+    assert holder_name(state("complete")) == "Clerical (data book)"
+    assert holder_name(state("filed")) == "the data book (filed)"
+    assert holder_name(state("ready"), "Weld Parts") == "Weld Parts"
+    assert holder_name(state("assigned"), "Weld Parts", "Stall 1") == "Weld Parts · Stall 1"
+    assert holder_name(state("review"), "Weld Parts", "Stall 1") == "Weld Parts"
+    assert holder_name(state("ready")) == "03"
+
+
+def test_works_on_station_limits() -> None:
+    s = state("assigned", station_id=4)
+    assert works_on(P(["shop:03:worker"]), s)
+    assert works_on(P(["shop:03:worker"], limits=(4, 5)), s)
+    assert not works_on(P(["shop:03:worker"], limits=(5,)), s)
+    assert not works_on(P(["shop:06:worker"]), s)
+    # leads and supervisors ignore station limits
+    assert works_on(P(["shop:03:lead"], limits=(5,)), s)
+    assert works_on(P(["shop:03:supervisor"], limits=(5,)), s)
+    # a limited worker with no station on the packet does not work on it
+    assert not works_on(P(["shop:03:worker"], limits=(5,)), state("assigned", station_id=None))
+
+
+@pytest.mark.parametrize("action", [A.START, A.SUBMIT])
+def test_start_and_submit_need_the_assigned_welder_for_a_worker(action: Action) -> None:
+    s = state("assigned")
+    assert check(P(["shop:03:worker"], user_id=WELDER), s, action).ok
+    other = check(P(["shop:03:worker"], user_id=OTHER), s, action)
+    assert (other.ok, other.code) == (False, "forbidden")
+    assert other.why == "Your access doesn't allow this step."
+    unassigned = check(P(["shop:03:worker"], user_id=WELDER), state("assigned", welder_id=None), action)
+    assert unassigned.code == "forbidden"
+    limited = check(P(["shop:03:worker"], user_id=WELDER, limits=(2,)), s, action)
+    assert limited.code == "forbidden"
+    # a lead does not need to be the welder
+    assert check(P(["shop:03:lead"], user_id=OTHER), s, action).ok
+    assert check(P(["shop:03:lead"], user_id=OTHER), state("assigned", welder_id=None), action).ok
+
+
+def test_worker_who_is_not_the_welder_still_holds_for_edits() -> None:
+    s = state("in_progress")
+    worker = P(["shop:03:worker"], user_id=OTHER)
+    assert check(worker, s, A.EDIT_WELDS).ok
+    assert check(worker, s, A.CLOCK).ok
+    assert check(P(["shop:03:worker"], limits=(9,)), s, A.EDIT_WELDS).code == "forbidden"
+
+
+def test_leads_write_shop_fields_only_while_they_hold() -> None:
+    lead = ROLES["lead03"]
+    for status in ("ready", "assigned", "in_progress", "review"):
+        assert check(lead, state(status), A.EDIT_WELDS).ok
+    d = check(ROLES["lead03"], state("qc"), A.EDIT_WELDS)
+    assert d.code == "wrong_status"
+    other_dept = check(P(["shop:06:lead"]), state("review"), A.EDIT_BOM)
+    assert other_dept.code == "forbidden"
+    assert other_dept.why == "Your access doesn't allow changing the materials on this packet. It's with 03 right now."
+    named = check(P(["shop:06:lead"]), state("review"), A.EDIT_BOM, extra={"holder_name": "Weld Parts"})
+    assert named.why.endswith("It's with Weld Parts right now.")
+
+
+# ---------- change-order hold ----------
+
+CO_HOLD_ACTIONS = {a for a, r in RULES.items() if r.co_hold}
+
+
+def test_co_hold_set_is_the_contract_set() -> None:
+    assert {
+        A.WH_RELEASE,
+        A.SUBMIT,
+        A.SEND_TO_QC,
+        A.ROUTE_FROM_REVIEW,
+        A.QC_ACCEPT,
+        A.QC_ACCEPT_AND_ROUTE,
+        A.FILE,
+    } == CO_HOLD_ACTIONS
+
+
+@pytest.mark.parametrize("action", list(EXPECTED))
+def test_co_hold_blocks_exactly_the_co_hold_actions(action: Action) -> None:
+    froms, allowed = EXPECTED[action]
+    status = sorted(froms)[0]
+    role = sorted(allowed)[0]
+    why = "Change order 2 must be verified first by: WH."
+    d = check(ROLES[role], state(status, co_hold=True, co_hold_why=why), action, extra=FULL_EXTRA)
+    if action in CO_HOLD_ACTIONS:
+        assert (d.ok, d.code, d.why) == (False, "co_hold", why)
+    else:
+        assert d.ok, (d.code, d.why)
+
+
+def test_co_hold_applies_to_admin() -> None:
+    d = check(ROLES["admin"], state("warehouse", co_hold=True), A.WH_RELEASE, it_reason=IT)
+    assert (d.code, d.why) == ("co_hold", "A change order holds this step.")
+
+
+# ---------- admin bypass ----------
+
+
+def test_admin_bypass_needs_an_it_reason() -> None:
+    admin = ROLES["admin"]
+    d = check(admin, state("draft"), A.SEND_TO_WAREHOUSE)
+    assert (d.ok, d.code) == (False, "it_reason_required")
+    short = check(admin, state("draft"), A.SEND_TO_WAREHOUSE, it_reason="too sh")
+    assert short.code == "it_reason_required"
+    ok = check(admin, state("draft"), A.SEND_TO_WAREHOUSE, it_reason=IT)
+    assert ok.ok and ok.admin_bypass
+
+
+def test_admin_bypass_passes_custody_and_is_marked() -> None:
+    admin = ROLES["admin"]
+    for status in ("ready", "assigned", "in_progress", "review"):
+        d = check(admin, state(status), A.EDIT_WELDS, it_reason=IT)
+        assert d.ok and d.admin_bypass
+    d = check(admin, state("in_progress"), A.SUBMIT, it_reason=IT)
+    assert d.ok and d.admin_bypass and d.signature == "Submit work to the supervisor"
+
+
+def test_admin_who_also_holds_is_not_a_bypass() -> None:
+    both = P(["admin", "clerical"])
+    d = check(both, state("draft"), A.EDIT_HEADER)
+    assert d.ok and not d.admin_bypass
+
+
+def test_admin_reopen_is_admin_only_and_not_a_bypass() -> None:
+    d = check(ROLES["admin"], state("filed"), A.ADMIN_REOPEN, extra=FULL_EXTRA)
+    assert d.ok and not d.admin_bypass and d.signature == "Reopen a filed packet"
+    for name, p in ROLES.items():
+        if name != "admin":
+            assert check(p, state("filed"), A.ADMIN_REOPEN, extra=FULL_EXTRA).code == "forbidden"
+
+
+def test_admin_set_databook_at_needs_no_reason() -> None:
+    d = check(ROLES["admin"], state("filed"), A.SET_DATABOOK_AT)
+    assert d.ok and d.admin_bypass
+
+
+def test_itp_reopen_qc_supervisor_or_admin_with_it_reason() -> None:
+    extra = {"reason": "Wrong line signed by mistake."}
+    assert check(ROLES["qc_sup"], state("complete"), A.ITP_REOPEN, extra=extra).ok
+    assert check(ROLES["qc"], state("complete"), A.ITP_REOPEN, extra=extra).code == "forbidden"
+    assert check(ROLES["admin"], state("complete"), A.ITP_REOPEN, extra=extra).code == "it_reason_required"
+    assert check(ROLES["admin"], state("complete"), A.ITP_REOPEN, extra=extra, it_reason=IT).ok
+
+
+# ---------- filed lock and removed ----------
+
+
+@pytest.mark.parametrize("action", list(RULES))
+def test_filed_lock(action: Action) -> None:
+    s = state("filed")
+    for name, p in ROLES.items():
+        d = check(p, s, action, extra={**FULL_EXTRA, "dept": "03"}, it_reason=IT)
+        if action is A.SET_DATABOOK_AT and name in {"admin", "clerical", "clerical_sup", "qc", "qc_sup"}:
+            assert d.ok, (name, d.code)
+        elif action is A.ADMIN_REOPEN and name == "admin":
+            assert d.ok
+        else:
+            assert not d.ok, (name, action)
+            if not RULES[action].from_status and not RULES[action].system_only and name == "admin":
+                assert d.code == "filed"
+                assert d.why == "This packet is filed and locked. IT can reopen it if a correction is needed."
+
+
+@pytest.mark.parametrize("action", list(RULES))
+def test_removed_refuses_everything(action: Action) -> None:
+    for status in STATUSES:
+        s = state(status, removed=True)
+        for p in [*ROLES.values(), Principal(user_id=None, system="visual-feed")]:
+            d = check(p, s, action, extra=FULL_EXTRA, it_reason=IT)
+            assert (d.ok, d.code) == (False, "removed")
+
+
+def test_unknown_status_is_wrong_status() -> None:
+    for action in RULES:
+        d = check(ROLES["admin"], state("shipped"), action, extra=FULL_EXTRA, it_reason=IT)
+        assert d.code == "wrong_status"
+    allowed, blocked = can(ROLES["clerical"], state("shipped"))
+    assert allowed == [] and blocked == {}
+
+
+# ---------- preconditions ----------
+
+# (action, status, role, state change or extra change, precondition name)
+PRECONDITION_CASES: list[tuple[Action, str, str, dict[str, Any], dict[str, Any], str]] = [
+    (A.SEND_TO_WAREHOUSE, "draft", "clerical", {"heat_map": None}, {}, "heat_map_answered"),
+    (A.SEND_TO_WAREHOUSE, "draft", "clerical", {"drawing_attached": False}, {}, "drawing_attached"),
+    (A.SEND_DRAFT_TO_QC, "draft", "clerical", {"drawing_attached": False}, {}, "drawing_attached"),
+    (A.RELEASE_TO_SHOP, "draft", "clerical", {}, {"dept": ""}, "target_dept_given"),
+    (A.WH_RELEASE, "warehouse", "warehouse", {"bom_ready": 2}, {}, "bom_all_pulled_or_short"),
+    (A.ASSIGN, "ready", "lead03", {}, {"welder_id": None}, "welder_set"),
+    (A.SUBMIT, "in_progress", "welder", {"checklist_complete": False}, {}, "checklist_complete"),
+    (A.SUBMIT, "in_progress", "welder", {"welds": 0}, {}, "welds_recorded"),
+    (A.SUBMIT, "in_progress", "welder", {"filler_set": False}, {}, "filler_set"),
+    (A.SEND_TO_QC, "review", "sup03", {}, {"spool": False}, "qc_book_spool_location"),
+    (A.SEND_TO_QC, "review", "sup03", {}, {"location": "  "}, "qc_book_spool_location"),
+    (A.ROUTE_FROM_REVIEW, "review", "sup03", {}, {"dept": "03"}, "target_dept_given"),
+    (A.QC_ACCEPT, "qc", "qc", {"qc_missing": ("weld exam",)}, {}, "qc_missing_empty"),
+    (A.QC_ACCEPT, "qc", "qc", {}, {"routed_to": "scrap"}, "route_given"),
+    (A.QC_ACCEPT_AND_ROUTE, "qc", "qc", {}, {"dept": "WH"}, "target_dept_given"),
+    (A.FILE, "complete", "clerical", {"vclosed": False}, {}, "vclosed_if_databook"),
+    (A.WH_SEND_BACK, "warehouse", "warehouse", {}, {"reason": "short"}, "reason_given"),
+    (A.QC_REJECT, "qc", "qc", {}, {"reason": None}, "reason_given"),
+    (A.REMOVE, "draft", "clerical", {}, {"reason": "  x  "}, "reason_given"),
+    (A.WH_RECEIVE_SHORT, "ready", "warehouse", {"open_shorts": 0}, {}, "short_open"),
+]
+
+
+@pytest.mark.parametrize(
+    ("action", "status", "role", "change", "extra", "name"),
+    PRECONDITION_CASES,
+    ids=[f"{c[0]}-{c[5]}" for c in PRECONDITION_CASES],
+)
+def test_each_precondition_fails_alone(
+    action: Action, status: str, role: str, change: dict[str, Any], extra: dict[str, Any], name: str
+) -> None:
+    s = state(status, **change)
+    x = {**FULL_EXTRA, **extra}
+    assert check(ROLES[role], state(status), action, extra=FULL_EXTRA).ok
+    d = check(ROLES[role], s, action, extra=x)
+    want = "reason_required" if name == "reason_given" else "precondition"
+    assert (d.ok, d.code) == (False, want)
+    if name == "qc_missing_empty":
+        assert d.why == "Q.C. still needs: weld exam."
+    elif name == "target_dept_given" and x["dept"] in ("", "WH"):
+        assert d.why == "Choose a department."
+    else:
+        assert d.why == MESSAGES[name]
+    # never bypassed by admin
+    a = check(ROLES["admin"], s, action, extra=x, it_reason=IT)
+    assert (a.ok, a.code, a.why) == (False, want, d.why)
+
+
+def test_every_named_precondition_is_tested() -> None:
+    used = {n for r in RULES.values() for n in r.preconditions}
+    assert used <= {c[5] for c in PRECONDITION_CASES}
+
+
+def test_filler_and_welds_only_in_weld_departments() -> None:
+    s = state("in_progress", weld_dept=False, welds=0, filler_set=False)
+    assert check(ROLES["welder"], s, A.SUBMIT).ok
+
+
+def test_vclosed_only_needed_for_the_data_book_route() -> None:
+    assert check(ROLES["clerical"], state("complete", routed_to="shipping", vclosed=False), A.FILE).ok
+
+
+def test_co_staged_if_dwg_change() -> None:
+    from packetstation.policy.preconditions import co_staged_if_dwg_change
+
+    s = state("draft")
+    assert co_staged_if_dwg_change(s, {}) is None
+    assert co_staged_if_dwg_change(s, {"dwg_change": True, "staged": True}) is None
+    assert co_staged_if_dwg_change(s, {"dwg_change": True}) == MESSAGES["co_staged_if_dwg_change"]
+
+
+# ---------- specific rules ----------
+
+
+@pytest.mark.parametrize("status", [s for s in STATUSES if s != "filed"])
+def test_wh_receive_short_passes_in_every_non_filed_status(status: str) -> None:
+    assert check(ROLES["warehouse"], state(status), A.WH_RECEIVE_SHORT).ok
+    assert check(ROLES["warehouse_sup"], state(status), A.WH_RECEIVE_SHORT).ok
+    assert check(ROLES["lead03"], state(status), A.WH_RECEIVE_SHORT).code == "forbidden"
+
+
+def test_wh_receive_short_refused_when_filed() -> None:
+    assert check(ROLES["warehouse"], state("filed"), A.WH_RECEIVE_SHORT).code == "filed"
+
+
+def test_wh_receive_short_needs_an_open_short() -> None:
+    d = check(ROLES["warehouse"], state("ready", open_shorts=0), A.WH_RECEIVE_SHORT)
+    assert d.code == "precondition" and d.why == MESSAGES["short_open"]
+    allowed, blocked = can(ROLES["warehouse"], state("draft", open_shorts=0))
+    assert allowed == [] and blocked == {A.WH_RECEIVE_SHORT: [MESSAGES["short_open"]]}
+
+
+def test_return_to_clerical_needs_wh_skipped() -> None:
+    assert check(ROLES["lead03"], state("ready"), A.RETURN_TO_CLERICAL, extra=FULL_EXTRA).ok
+    d = check(ROLES["lead03"], state("ready", wh_skipped=False), A.RETURN_TO_CLERICAL, extra=FULL_EXTRA)
+    assert d.code == "wrong_status"
+    assert "warehouse" in d.why
+
+
+def test_set_vclosed_needs_the_data_book_route() -> None:
+    assert check(ROLES["clerical"], state("complete"), A.SET_VCLOSED).ok
+    for route in ("shipping", None):
+        d = check(ROLES["clerical"], state("complete", routed_to=route), A.SET_VCLOSED)
+        assert d.code == "wrong_status"
+    assert check(ROLES["qc"], state("complete"), A.SET_VCLOSED).code == "forbidden"
+
+
+@pytest.mark.parametrize("status", STATUSES)
+def test_set_databook_at_is_clerical_or_qc(status: str) -> None:
+    s = state(status)
+    for role in ("clerical", "clerical_sup", "qc", "qc_sup"):
+        assert check(ROLES[role], s, A.SET_DATABOOK_AT).ok
+    for role in ("warehouse", "warehouse_sup", "sup03", "lead03", "pm", "welder"):
+        assert check(ROLES[role], s, A.SET_DATABOOK_AT).code == "forbidden"
+
+
+def test_rev_approve_needs_clerical_supervisor() -> None:
+    extra = {"reason": "Drafting confirmed rev C."}
+    assert check(ROLES["clerical_sup"], state("draft"), A.REV_APPROVE, extra=extra).ok
+    assert check(ROLES["clerical"], state("draft"), A.REV_APPROVE, extra=extra).code == "forbidden"
+
+
+def test_co_verify_needs_a_supervisor_of_the_named_department() -> None:
+    s = state("in_progress")
+    assert check(ROLES["sup03"], s, A.CO_VERIFY, extra={"dept": "03"}).ok
+    assert check(ROLES["sup03"], s, A.CO_VERIFY, extra={"dept": "06"}).code == "forbidden"
+    assert check(ROLES["lead03"], s, A.CO_VERIFY, extra={"dept": "03"}).code == "forbidden"
+    assert check(ROLES["warehouse_sup"], s, A.CO_VERIFY, extra={"dept": "WH"}).ok
+    assert check(ROLES["warehouse"], s, A.CO_VERIFY, extra={"dept": "WH"}).code == "forbidden"
+    assert check(ROLES["qc_sup"], s, A.CO_VERIFY, extra={"dept": "QC"}).ok
+    assert check(ROLES["clerical_sup"], s, A.CO_VERIFY, extra={"dept": "CL"}).code == "forbidden"
+    assert check(ROLES["sup03"], s, A.CO_VERIFY).code == "forbidden"
+
+
+@pytest.mark.parametrize("action", list(RULES))
+def test_pm_can_do_nothing(action: Action) -> None:
+    pm = ROLES["pm"]
+    for status in STATUSES:
+        assert not check(pm, state(status), action, extra=FULL_EXTRA).ok
+    for status in STATUSES:
+        assert can(pm, state(status)) == ([], {})
+
+
+SYSTEM = Principal(user_id=None, system="visual-feed")
+VIS = {A.VIS_UPDATE_DRAFT, A.VIS_OPEN_CO, A.VIS_AUTO_RELEASE, A.VIS_RETIRE_DRAFT}
+VIS_FROM = {
+    A.VIS_UPDATE_DRAFT: {"draft"},
+    A.VIS_RETIRE_DRAFT: {"draft"},
+    A.VIS_AUTO_RELEASE: {"draft"},
+    A.VIS_OPEN_CO: set(STATUSES) - {"draft", "filed"},
+}
+
+
+@pytest.mark.parametrize("action", list(RULES))
+def test_system_principal(action: Action) -> None:
+    assert {a for a, r in RULES.items() if r.system_only} == VIS
+    for status in STATUSES:
+        d = check(SYSTEM, state(status), action)
+        if action in VIS:
+            assert d.ok is (status in VIS_FROM[action]), (status, d.code)
+            if not d.ok:
+                assert d.code == "wrong_status"
+        elif not d.ok and d.code != "wrong_status":
+            assert d.code == "forbidden"
+        else:
+            assert not d.ok
+    assert can(SYSTEM, state("draft")) == ([], {})
+
+
+@pytest.mark.parametrize("action", sorted(VIS))
+def test_people_cannot_do_vis_actions(action: Action) -> None:
+    s = state(sorted(VIS_FROM[action])[0])
+    for p in ROLES.values():
+        d = check(p, s, action, it_reason=IT)
+        assert (d.ok, d.code) == (False, "system_only")
+
+
+def test_vis_auto_release_preconditions() -> None:
+    d = check(SYSTEM, state("draft", heat_map=None), A.VIS_AUTO_RELEASE)
+    assert (d.code, d.why) == ("precondition", MESSAGES["heat_map_answered"])
+
+
+# ---------- can and blocked ----------
+
+
+def test_can_and_blocked_on_a_bare_draft() -> None:
+    s = state("draft", heat_map=None, drawing_attached=False)
+    allowed, blocked = can(ROLES["clerical"], s)
+    reasons = [MESSAGES["heat_map_answered"], MESSAGES["drawing_attached"]]
+    assert blocked[A.SEND_TO_WAREHOUSE] == reasons
+    assert blocked[A.SEND_DRAFT_TO_QC] == reasons
+    assert blocked[A.RELEASE_TO_SHOP] == reasons
+    assert A.SEND_TO_WAREHOUSE not in allowed
+    assert {A.EDIT_HEADER, A.EDIT_BOM, A.EDIT_WOS, A.REMOVE, A.CO_ISSUE, A.DOC_ADD} <= set(allowed)
+    assert A.REV_APPROVE not in allowed  # clerical supervisor only
+    assert all(not RULES[a].system_only for a in allowed)
+    wh_allowed, wh_blocked = can(ROLES["warehouse"], s)
+    assert A.SEND_TO_WAREHOUSE not in wh_allowed
+    assert A.SEND_TO_WAREHOUSE not in wh_blocked
+    assert wh_allowed == [A.WH_RECEIVE_SHORT]
+
+
+def test_can_skips_input_only_preconditions() -> None:
+    assert {"welder_set", "target_dept_given", "route_given", "reason_given"} <= INPUT_ONLY
+    allowed, _ = can(ROLES["lead03"], state("ready"))
+    assert {A.ASSIGN, A.SEND_BACK_TO_WAREHOUSE, A.RETURN_TO_CLERICAL} <= set(allowed)
+    allowed, _ = can(ROLES["sup03"], state("review"))
+    assert {A.SEND_TO_QC, A.ROUTE_FROM_REVIEW, A.RETURN_TO_STATION} <= set(allowed)
+    allowed, _ = can(ROLES["qc"], state("qc"))
+    assert {A.QC_ACCEPT, A.QC_ACCEPT_AND_ROUTE, A.QC_REJECT} <= set(allowed)
+
+
+def test_blocked_by_co_hold_only_after_who_passes() -> None:
+    s = state("warehouse", co_hold=True, co_hold_why="Held for CO 1.", bom_ready=1)
+    _, blocked = can(ROLES["warehouse"], s)
+    assert blocked[A.WH_RELEASE] == ["Held for CO 1.", MESSAGES["bom_all_pulled_or_short"]]
+    _, other = can(ROLES["clerical"], s)
+    assert A.WH_RELEASE not in other
+
+
+def test_can_for_admin_assumes_an_it_reason() -> None:
+    from packetstation.policy import evaluate
+
+    allowed, _ = can(ROLES["admin"], state("draft"))
+    assert A.SEND_TO_WAREHOUSE in allowed and A.ADMIN_REOPEN not in allowed
+    assert evaluate(ROLES["admin"], state("draft"), A.SEND_TO_WAREHOUSE).admin_bypass
+    no_assume, _ = can(ROLES["admin"], state("draft"), assume_it_reason=False)
+    assert A.SEND_TO_WAREHOUSE not in no_assume
+    filed, _ = can(ROLES["admin"], state("filed"))
+    assert set(filed) == {A.ADMIN_REOPEN, A.SET_DATABOOK_AT}
+    assert not evaluate(ROLES["admin"], state("filed"), A.ADMIN_REOPEN).admin_bypass
+
+
+def test_can_for_the_welder() -> None:
+    allowed, blocked = can(ROLES["welder"], state("in_progress", checklist_complete=False))
+    assert A.SUBMIT in blocked and blocked[A.SUBMIT] == [MESSAGES["checklist_complete"]]
+    assert {A.EDIT_WELDS, A.CLOCK, A.EDIT_CHECKLIST} <= set(allowed)
+    assert A.PULL_BACK not in allowed
+
+
+# ---------- principal_from_roles ----------
+
+
+def test_principal_from_roles() -> None:
+    cs = principal_from_roles("u1", ["clerical_supervisor"])
+    assert cs.support == {"CL"} and cs.support_sup == {"CL"}
+    c = principal_from_roles("u1", ["clerical"])
+    assert c.support == {"CL"} and c.support_sup == frozenset()
+    w = principal_from_roles("u1", ["warehouse_supervisor", "qc"])
+    assert w.support == {"WH", "QC"} and w.support_sup == {"WH"}
+    shop = principal_from_roles("u1", ["shop:03:worker", "shop:06:supervisor", "shop:03:lead", "shop:10:worker"])
+    assert dict(shop.shop) == {"03": 2, "06": 3, "10": 1}
+    assert isinstance(shop.shop, MappingProxyType)
+    admin = principal_from_roles("u1", ["admin"])
+    assert admin.admin and not admin.support and not admin.shop and not admin.pm
+    pm = principal_from_roles("u1", ["pm"], session_kind="pin")
+    assert pm.pm and pm.session_kind == "pin" and not pm.support
+    limited = principal_from_roles("u1", ["shop:03:worker"], station_limits=[3, 4])
+    assert limited.station_limits == {3, 4}
+    assert principal_from_roles("u1", ["nonsense", "shop:03:boss"]) == Principal(user_id="u1")
+    assert principal_from_roles("u1", []).system is None
+
+
+def test_multi_role_user() -> None:
+    mia = principal_from_roles("u9", ["clerical", "shop:06:supervisor"])
+    assert check(mia, state("draft"), A.SEND_TO_WAREHOUSE).ok
+    assert check(mia, state("review", dept="06"), A.SEND_TO_QC, extra=FULL_EXTRA).ok
+    assert check(mia, state("review"), A.SEND_TO_QC, extra=FULL_EXTRA).code == "forbidden"