"""Phase 6 Q.C. policy: the packet Q.C. actions (QC_ACCEPT, QC_ACCEPT_AND_ROUTE, QC_REJECT, EDIT_QC, DOC_ADD, DOC_REMOVE) as every test user from every status the helpers reach, the scoped actions (QC_DOC_ADD and the ITP actions) through `check_scoped`, and the qc_missing_empty gate one reason at a time. The expectations come from an oracle written from the phase 6 contract (who, from-status, CO hold, filed lock, preconditions), not from the engine. Part one runs the pure policy; part two runs the API.""" from __future__ import annotations import dataclasses import itertools from collections.abc import Iterator from typing import Any import pytest from fakedir import FakeDirectory from pkthelpers import step from pshelpers import needs_pg from qchelpers import ( QC_KEYS, accepted_packet, add_packet_doc, at_qc, filed_packet, finish_inspection, qc_check, qc_patch, qc_weld, take_out, upload_doc, ) from shophelpers import ( IT, SIGN, USERS, Shop, draft, fresh, in_progress, in_review, issue_co, ok, open_shop, ready_packet, refused, to_station, to_warehouse, ) from packetstation.policy import SYSTEM_FEED, Action, PacketState, Principal, can, check, principal_from_roles from packetstation.settings import Settings ANDY = "00000000-0000-0000-0000-00000000a001" REASON = "Weld 2 undercut too deep" NCR = "NCR-2026-0021" PACKET_ACTIONS = ("qc_accept", "qc_accept_and_route", "qc_reject", "edit_qc", "doc_add", "doc_remove") TRANSITIONS = ("qc_accept", "qc_accept_and_route", "qc_reject") FROM: dict[str, frozenset[str]] = { "qc_accept": frozenset({"qc"}), "qc_accept_and_route": frozenset({"qc"}), "qc_reject": frozenset({"qc"}), "edit_qc": frozenset({"qc"}), "doc_add": frozenset(), # any status; the filed lock applies "doc_remove": frozenset(), } WHO = { "qc_accept": "QC", "qc_accept_and_route": "QC", "qc_reject": "QC", "edit_qc": "QC", "doc_add": "CL_OR_QC", "doc_remove": "CL_OR_QC", } CO_HELD = frozenset({"qc_accept", "qc_accept_and_route"}) GATED = frozenset({"qc_accept", "qc_accept_and_route"}) # qc_missing_empty SIGNED = { "qc_accept": "Q.C. accept", "qc_accept_and_route": "Q.C. accept and route", "qc_reject": "Q.C. reject", } ROLES = { "admin.ann": ["admin"], "clerk.clara": ["clerical"], "clerk.leo": ["clerical_supervisor"], "wh.wade": ["warehouse"], "qc.quinn": ["qc"], "pm.paula": ["pm"], "weld.andy": ["shop:03:worker"], "lead.lou": ["shop:03:lead"], "sup.wes": ["shop:03:supervisor"], "multi.mia": ["clerical", "shop:06:supervisor"], "both.bo": ["shop:03:worker", "shop:03:supervisor"], "dup.dave": ["shop:03:worker"], } QC_PEOPLE = frozenset({"qc.quinn"}) CL_OR_QC_PEOPLE = frozenset({"clerk.clara", "clerk.leo", "multi.mia", "qc.quinn"}) @dataclasses.dataclass(frozen=True) class Case: name: str status: str qc_done: bool = False # the open inspection has nothing missing CASES = { "draft": Case("draft", "draft"), "warehouse": Case("warehouse", "warehouse"), "ready": Case("ready", "ready"), "assigned": Case("assigned", "assigned"), "in_progress": Case("in_progress", "in_progress"), "review": Case("review", "review"), "qc": Case("qc", "qc"), "qc_done": Case("qc_done", "qc", qc_done=True), "complete": Case("complete", "complete"), "filed": Case("filed", "filed"), } def who_ok(user: str, action: str) -> bool: if WHO[action] == "QC": return user in QC_PEOPLE return user in CL_OR_QC_PEOPLE def expected(user: str, case: Case, action: str, *, it_reason: bool = False, co_hold: bool = False) -> tuple[int, str]: """The answer with every request input given (route, dept, reason, signature). The order is the contract's: status, CO hold, admin bypass, filed lock, who, preconditions.""" if FROM[action] and case.status not in FROM[action]: return 403, "wrong_status" if co_hold and action in CO_HELD: return 403, "co_hold" allowed = who_ok(user, action) if not allowed and user == "admin.ann": if not it_reason: return 400, "it_reason_required" allowed = True if case.status == "filed": return 403, "filed" if not allowed: return 403, "forbidden" if action in GATED and not case.qc_done: return 403, "precondition" return 200, "" def in_can(user: str, case: Case, action: str, **kw: Any) -> bool: return expected(user, case, action, it_reason=True, **kw)[0] == 200 def in_blocked(user: str, case: Case, action: str, **kw: Any) -> bool: """`blocked`: status, who and the filed lock pass, but qc_missing_empty or the CO hold fails.""" code = expected(user, case, action, it_reason=True, **kw)[1] if code == "co_hold": return who_ok(user, action) or user == "admin.ann" return code == "precondition" # ================================================================ part one: the pure policy MISSING = ("10 inspection items", "2 welds without a visual result", "inspector name") def P(user: str) -> Principal: uid = ANDY if user == "weld.andy" else f"00000000-0000-0000-0000-{abs(hash(user)) % 10**12:012d}" return principal_from_roles(uid, ROLES[user]) def state(case: Case, **kw: Any) -> PacketState: active = case.status in ("assigned", "in_progress", "review") base = PacketState( id=1, status=case.status, dept="03", station_id=101 if active else None, welder_id=ANDY if active else None, wh_skipped=False, routed_to="databook" if case.status in ("complete", "filed") else None, vclosed=case.status == "filed", 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, qc_missing=() if case.qc_done or case.status != "qc" else MISSING, has_qc_sent=case.status == "qc", ) return dataclasses.replace(base, **kw) EXTRA: dict[str, dict[str, Any]] = { "qc_accept": {"routed_to": "databook"}, "qc_accept_and_route": {"dept": "09"}, "qc_reject": {"reason": REASON}, } @pytest.mark.parametrize(("user", "case", "action"), list(itertools.product(ROLES, CASES, PACKET_ACTIONS))) def test_policy_matrix(user: str, case: str, action: str) -> None: c = CASES[case] s = state(c) d = check(P(user), s, Action(action), extra=EXTRA.get(action, {})) want = expected(user, c, action) assert ("" if d.ok else d.code) == want[1], (user, case, action, d) if d.ok and action in SIGNED: assert d.signature == SIGNED[action] allowed, blocked = can(P(user), s) assert (Action(action) in allowed) == in_can(user, c, action), (user, case, action, allowed) assert (Action(action) in blocked) == in_blocked(user, c, action), (user, case, action, blocked) if Action(action) in blocked and action in GATED: assert blocked[Action(action)] == ["Q.C. still needs: " + ", ".join(MISSING) + "."] @pytest.mark.parametrize("user", list(ROLES)) @pytest.mark.parametrize("case", ["qc", "qc_done"]) def test_policy_co_hold_on_accept_not_reject(user: str, case: str) -> None: c = CASES[case] why = "Change order CO-1 must be verified first by: 03 Weld Parts." s = state(c, co_hold=True, co_hold_why=why) allowed, blocked = can(P(user), s) for action in PACKET_ACTIONS: d = check(P(user), s, Action(action), extra=EXTRA.get(action, {})) want = expected(user, c, action, co_hold=True) assert ("" if d.ok else d.code) == want[1], (user, case, action, d) if want[1] == "co_hold": assert d.why == why assert (Action(action) in allowed) == in_can(user, c, action, co_hold=True), (user, case, action) assert (Action(action) in blocked) == in_blocked(user, c, action, co_hold=True), (user, case, action) if Action(action) in blocked and action in CO_HELD: assert blocked[Action(action)][0] == why if case == "qc": assert blocked[Action(action)][1] == "Q.C. still needs: " + ", ".join(MISSING) + "." def test_policy_reject_needs_a_reason() -> None: s = state(CASES["qc"]) for extra in ({}, {"reason": ""}, {"reason": "bad"}): d = check(P("qc.quinn"), s, Action.QC_REJECT, extra=extra) assert d.code == "reason_required", (extra, d) assert check(P("qc.quinn"), s, Action.QC_REJECT, extra={"reason": NCR}).ok # the accept routes are inputs assert check(P("qc.quinn"), state(CASES["qc_done"]), Action.QC_ACCEPT, extra={}).code == "precondition" d = check(P("qc.quinn"), state(CASES["qc_done"]), Action.QC_ACCEPT_AND_ROUTE, extra={"dept": "03"}) assert d.code == "precondition" # ---------------------------------------------------------------- scoped actions (no packet) SCOPED: dict[str, str] = { "qc_doc_add": "CL_OR_QC", "itp_start": "QC", "itp_line_sign": "QC", "itp_line_unsign": "QC", "itp_edit_line": "QC", "itp_block_sign": "QC", "itp_block_unsign": "QC", "itp_sign_off": "QC", "itp_reopen": "QC_SUP", } SCOPED_SIGNED = {"itp_block_sign": "Sign an ITP block", "itp_sign_off": "Sign off the ITP"} SCOPED_ROLES = {**ROLES, "qc.sup": ["qc_supervisor"]} def scoped_who(user: str, who: str) -> bool: if who == "QC": return user in ("qc.quinn", "qc.sup") if who == "QC_SUP": return user == "qc.sup" return user in CL_OR_QC_PEOPLE or user == "qc.sup" def scoped_expected(user: str, action: str, *, it_reason: bool, reason: bool = True) -> tuple[bool, str, bool]: """(ok, code, admin_bypass) for check_scoped.""" allowed = scoped_who(user, SCOPED[action]) bypass = False if not allowed and user == "admin.ann": if not it_reason: return False, "it_reason_required", False allowed, bypass = True, True if not allowed: return False, "forbidden", False if action == "itp_reopen" and not reason: return False, "reason_required", bypass return True, "", bypass def _check_scoped() -> Any: from packetstation.policy.engine import check_scoped return check_scoped def SP(user: str) -> Principal: return principal_from_roles(f"00000000-0000-0000-0000-{abs(hash(user)) % 10**12:012d}", SCOPED_ROLES[user]) @pytest.mark.parametrize(("user", "action"), list(itertools.product(SCOPED_ROLES, SCOPED))) @pytest.mark.parametrize("it", [False, True]) def test_check_scoped(user: str, action: str, it: bool) -> None: check_scoped = _check_scoped() extra = {"reason": "Customer asked to re-inspect 4.2"} if action == "itp_reopen" else {} d = check_scoped(SP(user), Action(action), extra=extra, it_reason=IT if it else None) want = scoped_expected(user, action, it_reason=it) assert (d.ok, "" if d.ok else d.code) == want[:2], (user, action, it, d) if d.ok: assert d.admin_bypass == want[2], (user, action, d) if action in SCOPED_SIGNED: assert d.signature == SCOPED_SIGNED[action] def test_check_scoped_special_cases() -> None: check_scoped = _check_scoped() # the system principal never passes a people action for action in SCOPED: d = check_scoped(SYSTEM_FEED, Action(action), extra={"reason": "Customer asked again"}) assert d.code == "forbidden", (action, d) # admin: an IT reason shorter than 8 characters is no IT reason d = check_scoped(SP("admin.ann"), Action("itp_start"), it_reason="short") assert d.code == "it_reason_required" d = check_scoped(SP("admin.ann"), Action("itp_start"), it_reason=IT) assert d.ok and d.admin_bypass # reopen: the reason is required for the Q.C. supervisor and for the admin d = check_scoped(SP("qc.sup"), Action("itp_reopen"), extra={}) assert d.code == "reason_required" d = check_scoped(SP("admin.ann"), Action("itp_reopen"), extra={"reason": "no"}, it_reason=IT) assert d.code == "reason_required" d = check_scoped(SP("qc.quinn"), Action("itp_reopen"), extra={"reason": "Customer asked again"}) assert d.code == "forbidden" # PM reads only for action in SCOPED: assert check_scoped(SP("pm.paula"), Action(action), extra={"reason": "Customer asked"}).code == "forbidden" def test_scoped_actions_exist_with_labels() -> None: for name in ("qc_doc_add", "itp_start", "itp_line_sign", "itp_line_unsign", "itp_block_unsign"): a = Action(name) assert a.value == name from packetstation.policy.actions import EDIT_WHAT, RULES for name in SCOPED: rule = RULES[Action(name)] assert rule.to_status is None and not rule.system_only, name assert name in {str(a) for a in EDIT_WHAT}, name assert EDIT_WHAT[Action("qc_doc_add")] == "the Q.C. folder" assert EDIT_WHAT[Action("itp_start")] == "the ITP" # ================================================================ part two: the API @pytest.fixture def shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]: yield from open_shop(pg_settings, fake_dir) def body_for(action: str) -> dict[str, Any]: return { "qc_accept": {"routed_to": "databook", "sign": SIGN}, "qc_accept_and_route": {"dept": "09", "sign": SIGN}, "qc_reject": {"reason": REASON, "sign": SIGN}, }[action] def build(shop: Shop, case: str) -> dict[str, Any]: """A packet in the case's status with one packet document (by clerk.clara), made through the API.""" if case == "filed": pkt = accepted_packet(shop) ok(add_packet_doc(shop("clerk.clara"), pkt["id"])) c = shop("clerk.clara") pkt = ok(c.post(f"/api/packets/{pkt['id']}/vclosed", json={"closed": True})) return ok(step(c, pkt, "file", sign=SIGN)) if case == "draft": pkt = draft(shop) elif case == "warehouse": pkt = to_warehouse(shop, draft(shop)) elif case == "ready": pkt = ready_packet(shop) elif case == "assigned": pkt = to_station(shop, ready_packet(shop)) elif case == "in_progress": pkt = in_progress(shop) elif case == "review": pkt = in_review(shop) elif case == "qc": pkt = at_qc(shop) elif case == "qc_done": pkt = finish_inspection(shop, at_qc(shop)) elif case == "complete": pkt = accepted_packet(shop) else: raise AssertionError(case) return ok(add_packet_doc(shop("clerk.clara"), pkt["id"])) def live_doc(pkt: dict[str, Any]) -> int: return int(next(d for d in pkt["docs"] if not d.get("removed"))["id"]) @needs_pg @pytest.mark.parametrize("case", list(CASES)) def test_api_matrix(shop: Shop, case: str) -> None: """Every refused Q.C. step answers the oracle's code and changes nothing; `can` and `blocked` agree for every user. Allowed steps are not sent here (test_qc_api and test_qc_docs send them).""" c = CASES[case] pkt = build(shop, case) assert pkt["status"] == c.status # Building a filed packet takes minutes through the tunnel; sessions idle that long lock. Sign in afresh. shop.clients.clear() doc_id = live_doc(pkt) asset = upload_doc(shop("clerk.clara"))["id"] for user in USERS: mine = fresh(shop(user), pkt["id"]) for action in PACKET_ACTIONS: assert (action in mine["can"]) == in_can(user, c, action), (user, case, action, mine["can"]) assert (action in mine["blocked"]) == in_blocked(user, c, action), (user, case, action, mine["blocked"]) want = expected(user, c, action) if c.status == "filed": # The write pipeline answers `filed` before it runs the policy (domain/pipeline.packet_write), so # through the API every write on a filed packet is 403 filed, whatever the engine order says. want = (403, "filed") if want[0] == 200: continue if action in TRANSITIONS: r = step(shop(user), pkt, action, **body_for(action)) elif action == "doc_add": r = add_packet_doc(shop(user), pkt["id"], asset_id=asset) elif action == "doc_remove": r = take_out(shop(user), pkt["id"], doc_id) else: r = qc_patch(shop(user), pkt["id"], {"notes": "Looks fine"}) if c.status != "qc" and r.status_code == 409: # no open inspection: the contract does not fix whether this or wrong_status answers first assert r.json()["code"] == "no_inspection", r.text continue assert (r.status_code, r.json().get("code")) == want, (user, case, action, r.text) if want[1] == "precondition" and action in GATED: assert r.json()["detail"].startswith("Q.C. still needs: ") assert mine["blocked"][action] == [r.json()["detail"]] after = fresh(shop("admin.ann"), pkt["id"]) assert (after["status"], after["version"]) == (pkt["status"], pkt["version"]) @needs_pg def test_api_admin_with_it_reason(shop: Shop) -> None: """The admin passes who with an IT reason but never the Q.C. gate or the filed lock.""" pkt = at_qc(shop) ann = shop("admin.ann") r = refused(step(ann, pkt, "qc_accept", routed_to="databook", sign=SIGN, it_reason=IT), 403, "precondition") assert r["detail"].startswith("Q.C. still needs: ") out = ok(qc_patch(ann, pkt["id"], {"notes": "Fixed by IT"}, it_reason=IT)) assert out["qc"]["notes"] == "Fixed by IT" out = ok(add_packet_doc(ann, pkt["id"], it_reason=IT)) refused(add_packet_doc(ann, pkt["id"]), 400, "it_reason_required") out = ok(step(ann, out, "qc_reject", reason=REASON, sign=SIGN, it_reason=IT)) assert out["status"] == "review" @needs_pg def test_api_signature_required(shop: Shop) -> None: pkt = finish_inspection(shop, at_qc(shop)) q = shop("qc.quinn") for action, body in ( ("qc_accept", {"routed_to": "databook"}), ("qc_accept_and_route", {"dept": "09"}), ("qc_reject", {"reason": REASON}), ): r = refused(step(q, pkt, action, **body), 400, "signature_required") assert r.get("meaning") == SIGNED[action], r assert fresh(q, pkt["id"])["status"] == "qc" def _missing(pkt: dict[str, Any]) -> list[str]: out: list[str] = pkt["qc"]["missing"] return out def _gate(shop: Shop, pkt: dict[str, Any], parts: list[str]) -> None: """The read model, the blocked list and a refused accept all say the same thing.""" msg = "Q.C. still needs: " + ", ".join(parts) + "." mine = fresh(shop("qc.quinn"), pkt["id"]) assert _missing(mine) == parts assert "qc_accept" not in mine["can"] assert mine["blocked"]["qc_accept"] == [msg] assert mine["blocked"]["qc_accept_and_route"] == [msg] r = refused(step(shop("qc.quinn"), mine, "qc_accept", routed_to="databook", sign=SIGN), 403, "precondition") assert r["detail"] == msg @needs_pg def test_api_qc_missing_one_by_one(shop: Shop) -> None: pkt = at_qc(shop) q = shop("qc.quinn") pid = pkt["id"] welds = sorted(int(w["weld_no"]) for w in pkt["welds"]) assert len(welds) == 2 _gate(shop, pkt, ["10 inspection items", "2 welds without a visual result", "inspector name"]) for k in QC_KEYS[:7]: ok(qc_check(q, pid, k)) _gate(shop, pkt, ["3 inspection items", "2 welds without a visual result", "inspector name"]) for k in QC_KEYS[7:9]: ok(qc_check(q, pid, k)) _gate(shop, pkt, ["1 inspection item", "2 welds without a visual result", "inspector name"]) ok(qc_check(q, pid, QC_KEYS[9])) _gate(shop, pkt, ["2 welds without a visual result", "inspector name"]) ok(qc_weld(q, pid, welds[0], result="accept")) _gate(shop, pkt, ["1 weld without a visual result", "inspector name"]) ok(qc_weld(q, pid, welds[1], result="reject", note="undercut")) _gate(shop, pkt, ["a rejected weld (reject the packet instead)", "inspector name"]) ok(qc_weld(q, pid, welds[1], result="accept")) _gate(shop, pkt, ["inspector name"]) ok(qc_patch(q, pid, {"nde_method": "RT"})) _gate(shop, pkt, ["NDE result", "inspector name"]) ok(qc_patch(q, pid, {"nde_result": "Rejectable"})) _gate(shop, pkt, ["NDE result", "inspector name"]) ok(qc_patch(q, pid, {"nde_result": "Acceptable"})) _gate(shop, pkt, ["inspector name"]) out = ok(qc_patch(q, pid, {"inspector": "Quinn QC"})) assert _missing(out) == [] assert "qc_accept" in out["can"] and "qc_accept_and_route" in out["can"] assert "qc_accept" not in out["blocked"] # ticking one key off again brings the gate back ok(qc_check(q, pid, "ph_qty", done=False)) _gate(shop, pkt, ["1 inspection item"]) @needs_pg def test_api_co_hold_blocks_accept_not_reject(shop: Shop) -> None: pkt = finish_inspection(shop, at_qc(shop)) pkt = issue_co(shop, pkt, ["03"]) mine = fresh(shop("qc.quinn"), pkt["id"]) held = "Change order CO-1 must be verified first by: 03 Weld Parts." assert mine["blocked"]["qc_accept"] == [held] assert "qc_reject" in mine["can"] refused(step(shop("qc.quinn"), mine, "qc_accept", routed_to="databook", sign=SIGN), 403, "co_hold", held) refused(step(shop("qc.quinn"), mine, "qc_accept_and_route", dept="09", sign=SIGN), 403, "co_hold", held) out = ok(step(shop("qc.quinn"), mine, "qc_reject", reason="Weld 2 undercut too deep", ncr=NCR, sign=SIGN)) assert out["status"] == "review" @needs_pg def test_api_filed_lock_on_documents(shop: Shop) -> None: pkt = filed_packet(shop) for user in ("qc.quinn", "clerk.clara"): refused(add_packet_doc(shop(user), pkt["id"]), 403, "filed") refused(add_packet_doc(shop("admin.ann"), pkt["id"], it_reason=IT), 403, "filed")