spencerm/PacketStation
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
591fe900761be313f9be6954bae94f8d755f09e7
PacketStation/tests/server/test_policy_matrix.py
T

631 lines
26 KiB
Python
Raw Normal View History

Phase 2: core domain, policy engine, packet API, Clerical screen
2026-10-16 09:31:45 -06:00
"""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"
Reference in New Issue Copy Permalink