"""The phase 4 release gate in the policy: heats_recorded over PacketState values, WH_RELEASE with both rules, and that can() and check() agree. No database.""" from __future__ import annotations import dataclasses import itertools from typing import Any import pytest from packetstation.policy import RULES, Action, PacketState, Principal, can, check, principal_from_roles from packetstation.policy.preconditions import INPUT_ONLY, MESSAGES, PRECONDITIONS, heats_recorded A = Action USER = "00000000-0000-0000-0000-0000000000c1" IT = "Ticket 4411: release by hand" MSG_PULL = "Pull every material line or mark it short first." MSG_HEAT = "Scan or type the heat number of every pulled line first. Use N/A for a part with no traceability." def P(*roles: str) -> Principal: return principal_from_roles(USER, list(roles)) WH = P("warehouse") WH_SUP = P("warehouse_supervisor") ADMIN = P("admin") LEAD = P("shop:03:lead") SUP = P("shop:03:supervisor") QC = P("qc") CL = P("clerical") def state(status: str = "warehouse", **kw: Any) -> PacketState: base = PacketState( id=1, status=status, dept="03", station_id=None, welder_id=None, wh_skipped=False, routed_to=None, vclosed=False, removed=False, source_kind="hand", weld_dept=True, co_hold=False, drawing_attached=True, heat_map=True, bom_total=3, bom_ready=3, bom_missing_heat=0, ) return dataclasses.replace(base, **kw) def test_packet_state_defaults_to_no_missing_heat() -> None: s = PacketState( id=1, status="warehouse", dept="03", station_id=None, welder_id=None, wh_skipped=False, routed_to=None, vclosed=False, removed=False, source_kind="hand", weld_dept=True, co_hold=False, ) assert s.bom_missing_heat == 0 def test_heats_recorded_is_a_state_precondition() -> None: assert MESSAGES["heats_recorded"] == MSG_HEAT assert PRECONDITIONS["heats_recorded"] is heats_recorded assert "heats_recorded" not in INPUT_ONLY assert RULES[A.WH_RELEASE].preconditions == ("bom_all_pulled_or_short", "heats_recorded") @pytest.mark.parametrize(("missing", "fails"), [(0, False), (1, True), (2, True), (40, True)]) def test_heats_recorded_over_state(missing: int, fails: bool) -> None: got = heats_recorded(state(bom_missing_heat=missing), {}) assert got == (MSG_HEAT if fails else None) def test_heats_recorded_ignores_input() -> None: assert heats_recorded(state(bom_missing_heat=1), {"heat": "H1", "na": True}) == MSG_HEAT assert heats_recorded(state(bom_missing_heat=0), {"anything": 1}) is None @pytest.mark.parametrize("who", [WH, WH_SUP]) def test_release_blocked_by_a_missing_heat(who: Principal) -> None: d = check(who, state(bom_missing_heat=1), A.WH_RELEASE) assert not d.ok and d.code == "precondition" and d.why == MSG_HEAT assert check(who, state(bom_missing_heat=0), A.WH_RELEASE).ok def test_release_rules_in_order() -> None: d = check(WH, state(bom_ready=2, bom_missing_heat=1), A.WH_RELEASE) assert not d.ok and d.why == MSG_PULL allowed, blocked = can(WH, state(bom_ready=2, bom_missing_heat=1)) assert A.WH_RELEASE not in allowed assert blocked[A.WH_RELEASE] == [MSG_PULL, MSG_HEAT] allowed, blocked = can(WH, state(bom_ready=3, bom_missing_heat=1)) assert A.WH_RELEASE not in allowed and blocked[A.WH_RELEASE] == [MSG_HEAT] allowed, blocked = can(WH, state()) assert A.WH_RELEASE in allowed and A.WH_RELEASE not in blocked def test_admin_cannot_bypass_the_heat_rule() -> None: d = check(ADMIN, state(bom_missing_heat=1), A.WH_RELEASE, it_reason=IT) assert not d.ok and d.code == "precondition" and d.why == MSG_HEAT d = check(ADMIN, state(bom_missing_heat=0), A.WH_RELEASE, it_reason=IT) assert d.ok and d.admin_bypass d = check(ADMIN, state(bom_missing_heat=0), A.WH_RELEASE) assert not d.ok and d.code == "it_reason_required" @pytest.mark.parametrize("who", [CL, QC, LEAD, SUP]) def test_others_are_refused_before_the_gate(who: Principal) -> None: d = check(who, state(bom_missing_heat=1), A.WH_RELEASE) assert not d.ok and d.code == "forbidden" allowed, blocked = can(who, state(bom_missing_heat=1)) assert A.WH_RELEASE not in allowed and A.WH_RELEASE not in blocked def test_the_gate_only_touches_wh_release() -> None: """A missing heat blocks nothing else the warehouse can do on a warehouse packet.""" ok_allowed, ok_blocked = can(WH, state()) bad_allowed, bad_blocked = can(WH, state(bom_missing_heat=3)) assert set(ok_allowed) - set(bad_allowed) == {A.WH_RELEASE} assert set(bad_blocked) - set(ok_blocked) == {A.WH_RELEASE} PRINCIPALS = {"wh": WH, "wh_sup": WH_SUP, "admin": ADMIN, "lead": LEAD, "sup": SUP, "qc": QC, "cl": CL} @pytest.mark.parametrize( ("name", "status", "total", "ready", "missing", "co_hold"), [ (n, st, t, r, m, h) for n, st, (t, r), m, h in itertools.product( PRINCIPALS, ("warehouse", "ready", "draft"), ((3, 3), (3, 2), (0, 0)), (0, 1), (False, True) ) ], ) def test_can_and_check_agree(name: str, status: str, total: int, ready: int, missing: int, co_hold: bool) -> None: who = PRINCIPALS[name] s = state(status, bom_total=total, bom_ready=ready, bom_missing_heat=missing, co_hold=co_hold) allowed, blocked = can(who, s) d = check(who, s, A.WH_RELEASE, it_reason=IT if who.admin else None) assert (A.WH_RELEASE in allowed) == d.ok who_passes = who in (WH, WH_SUP, ADMIN) and status == "warehouse" if not d.ok and who_passes: assert d.code in ("precondition", "co_hold") assert A.WH_RELEASE in blocked and d.why == blocked[A.WH_RELEASE][0] if d.ok: assert A.WH_RELEASE not in blocked assert total <= ready and missing == 0 and not co_hold def test_receive_by_custody_in_the_policy() -> None: """The warehouse receives anywhere but filed (WH_RECEIVE_SHORT); the holder through EDIT_BOM.""" for status in ("warehouse", "ready", "assigned", "in_progress", "review", "qc", "complete"): assert check(WH, state(status, open_shorts=1), A.WH_RECEIVE_SHORT).ok, status assert check(WH, state("filed", open_shorts=1), A.WH_RECEIVE_SHORT).code == "filed" assert check(WH, state("ready", open_shorts=0), A.WH_RECEIVE_SHORT).why == MESSAGES["short_open"] ready = state("ready", open_shorts=1) assert check(LEAD, ready, A.EDIT_BOM).ok and check(SUP, ready, A.EDIT_BOM).ok assert check(WH, ready, A.EDIT_BOM).code == "forbidden" for who in (QC, CL): assert check(who, ready, A.EDIT_BOM).code == "forbidden" assert check(who, ready, A.WH_RECEIVE_SHORT).code == "forbidden"