177 lines
6.7 KiB
Python
177 lines
6.7 KiB
Python
 
|
"""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"
|