spencermandClaude Fable 5.1
39a4eb5213
The Warehouse home with the numbers, the Open shorts table, the find bar and the five-column task board (waiting, being pulled, on hold, pulled, released in the last 3 days) with buttons, hold reasons and drag. Every role sees the warehouse stage on a packet as the WH pill and the line under it. The pull table on the Materials tab: Pulled with auto-start, issued, location, heat with Enter to the next line, shorts with receipts, N/A, MTR per line with a viewer. Send back to Clerical with a reason. Release to supervisor stays blocked until every line is pulled or short and every pulled line has a heat. Server: the heats_recorded precondition on WH_RELEASE, receive by custody, server-validated hold and short reasons, the old packet history texts, GET /api/warehouse/board, cards with the warehouse stage and pull counts. Uploads typed by content per kind (drawing, mtr, document, bom with XLSX and CSV), limits per kind, optional ClamAV behind the clamav compose profile, a storage quota alert on the new notification_events table (migration 0005) shown on the Admin Storage tab. Tests: every warehouse action as every role, the release gate, the short-receive exception, the heat pattern, upload sniffing and limits, the ClamAV path with a fake scanner, the quota alert; frontend board and pull table tests; Playwright scenarios for the board and the pull flow. Verified on docker-1 in the CI stack: 3557 server tests, 106 frontend tests, 17 Playwright scenarios. Docs updated; D5 closed. Co-Authored-By: Claude Fable 5.1 <[email protected]>
632 lines
26 KiB
Python
632 lines
26 KiB
Python
"""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.WH_RELEASE, "warehouse", "warehouse", {"bom_missing_heat": 1}, {}, "heats_recorded"),
|
|
(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"
|