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

575 lines
22 KiB
Python
Raw Permalink Normal View History

Phase 5: Supervisor and Stall screens, weld map, heat table, clock and CO verification
2026-10-16 22:46:56 -06:00
"""Phase 5 shop policy: every shop step as every test user from every status the helpers reach, with the expected
code (200, 403 forbidden, 403 wrong_status, 403 precondition, 400 it_reason_required, 400 reason_required, 400
signature_required); the assigned-welder and station-limit rules; lead versus supervisor; RETURN_TO_CLERICAL only
when wh_skipped; the change-order hold set; and the read model's `can` and `blocked` agreeing with the writes.
The expectations come from an oracle written from the contract (who, from-status, preconditions), not from the
engine, so the engine and the oracle check each other. 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 app_conn, needs_pg
from psycopg.types.json import Jsonb
from shophelpers import (
IT,
MSG_CHECKLIST,
SIGN,
USERS,
Shop,
draft,
finish_stage,
fresh,
in_progress,
in_review,
issue_co,
ok,
open_shop,
ready_packet,
ready_skipped,
refused,
run,
set_limits,
start,
submit,
to_station,
to_warehouse,
verify,
)
from packetstation.policy import Action, PacketState, Principal, can, check, principal_from_roles
from packetstation.settings import Settings
A = Action
ANDY = "00000000-0000-0000-0000-00000000a001"
STALL1, STALL2 = 101, 102
SHOP_ACTIONS = (
"assign",
"unassign",
"pull_back",
"start",
"submit",
"return_to_station",
"send_to_qc",
"route_from_review",
"send_back_to_warehouse",
"return_to_clerical",
)
FROM: dict[str, frozenset[str]] = {
"assign": frozenset({"ready"}),
"unassign": frozenset({"assigned"}),
"pull_back": frozenset({"in_progress"}),
"start": frozenset({"assigned"}),
"submit": frozenset({"assigned", "in_progress"}),
"return_to_station": frozenset({"review"}),
"send_to_qc": frozenset({"review"}),
"route_from_review": frozenset({"review"}),
"send_back_to_warehouse": frozenset({"ready"}),
"return_to_clerical": frozenset({"ready"}),
}
WHO = {
"assign": "lead",
"unassign": "lead",
"pull_back": "lead",
"start": "works",
"submit": "works",
"return_to_station": "sup",
"send_to_qc": "sup",
"route_from_review": "sup",
"send_back_to_warehouse": "lead",
"return_to_clerical": "lead",
}
CO_HELD = frozenset({"submit", "send_to_qc", "route_from_review"})
REASON = "Weld 2 needs a repair pass"
# Shop levels in 03 per test user (multi.mia leads 06 only; both.bo is worker and supervisor, so supervisor).
LEVEL_03 = {"weld.andy": 1, "dup.dave": 1, "lead.lou": 2, "sup.wes": 3, "both.bo": 3}
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"],
}
@dataclasses.dataclass(frozen=True)
class Case:
"""A status the helpers reach, and what matters about it for the shop steps."""
name: str
status: str
skipped: bool = False # wh_skipped
finished: bool = False # checklist, a weld and the filler done
CASES = {
"warehouse": Case("warehouse", "warehouse"),
"ready": Case("ready", "ready"),
"ready_skipped": Case("ready_skipped", "ready", skipped=True),
"assigned": Case("assigned", "assigned"),
"in_progress": Case("in_progress", "in_progress"),
"in_progress_done": Case("in_progress_done", "in_progress", finished=True),
"review": Case("review", "review", finished=True),
"qc": Case("qc", "qc", finished=True),
}
def who_ok(user: str, action: str, *, welder: str | None = "weld.andy", limited_out: bool = False) -> bool:
level = LEVEL_03.get(user, 0)
kind = WHO[action]
if kind == "lead":
return level >= 2
if kind == "sup":
return level >= 3
return level >= 2 or (level == 1 and user == welder and not limited_out)
def expected(
user: str,
case: Case,
action: str,
*,
it_reason: bool = False,
co_hold: bool = False,
welder: str | None = "weld.andy",
limited_out: bool = False,
) -> tuple[int, str]:
"""The answer to the step with every request input given (station, welder, reason, book, spool, location,
dept, signature). The order is the engine's: status, CO hold, admin bypass, who, preconditions."""
if case.status not in FROM[action]:
return 403, "wrong_status"
if action == "return_to_clerical" and not case.skipped:
return 403, "wrong_status"
if co_hold and action in CO_HELD:
return 403, "co_hold"
allowed = who_ok(user, action, welder=welder, limited_out=limited_out)
if not allowed and user == "admin.ann":
if not it_reason:
return 400, "it_reason_required"
allowed = True
if not allowed:
return 403, "forbidden"
if action == "submit" and not case.finished:
return 403, "precondition"
return 200, ""
def in_can(user: str, case: Case, action: str, **kw: Any) -> bool:
"""`can` assumes the admin's IT reason and skips request inputs."""
return expected(user, case, action, it_reason=True, **kw)[0] == 200
def in_blocked(user: str, case: Case, action: str, **kw: Any) -> bool:
"""`blocked`: who and custody pass (or the admin bypass) but a state precondition or the CO hold fails."""
code = expected(user, case, action, it_reason=True, **kw)[1]
if code == "co_hold":
# The engine defers the CO hold in read mode and lists it only once who passes.
return who_ok(user, action, welder=kw.get("welder", "weld.andy")) or user == "admin.ann"
return code == "precondition"
# ================================================================ part one: the pure policy
def P(user: str, limits: tuple[int, ...] = ()) -> 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], station_limits=limits)
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=STALL1 if active else None,
welder_id=ANDY if active else None,
wh_skipped=case.skipped,
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,
welds=1 if case.finished else 0,
checklist_complete=case.finished,
filler_set=case.finished,
)
return dataclasses.replace(base, **kw)
EXTRA = {
"assign": {"station_id": STALL1, "welder_id": ANDY},
"return_to_station": {"reason": REASON},
"send_back_to_warehouse": {"reason": REASON},
"return_to_clerical": {"reason": REASON},
"send_to_qc": {"book": True, "spool": True, "location": "Bay 2"},
"route_from_review": {"dept": "09"},
}
@pytest.mark.parametrize(("user", "case", "action"), list(itertools.product(ROLES, CASES, SHOP_ACTIONS)))
def test_policy_matrix(user: str, case: str, action: str) -> None:
c = CASES[case]
s = state(c)
d = check(P(user), s, A(action), extra=EXTRA.get(action, {}))
want = expected(user, c, action)
code = "" if d.ok else d.code
assert code == want[1], (user, case, action, d)
allowed, blocked = can(P(user), s)
assert (A(action) in allowed) == in_can(user, c, action), (user, case, action, allowed)
assert (A(action) in blocked) == in_blocked(user, c, action), (user, case, action, blocked)
@pytest.mark.parametrize("user", list(ROLES))
@pytest.mark.parametrize("case", ["ready", "assigned", "in_progress_done", "review"])
def test_policy_co_hold_set(user: str, case: str) -> None:
"""The CO hold blocks SUBMIT, SEND_TO_QC and ROUTE_FROM_REVIEW; never ASSIGN, START or the others."""
c = CASES[case]
why = "Change order CO-2 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 SHOP_ACTIONS:
d = check(P(user), s, A(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
if in_blocked(user, c, action, co_hold=True):
assert blocked[A(action)][0] == why
assert (A(action) in allowed) == in_can(user, c, action, co_hold=True), (user, case, action)
@pytest.mark.parametrize("action", ["start", "submit"])
def test_policy_assigned_welder_rules(action: str) -> None:
c = CASES["in_progress_done"] if action == "submit" else CASES["assigned"]
s = state(c)
assert check(P("weld.andy"), s, A(action)).ok
# another worker in the department is not the assigned welder
assert check(P("dup.dave"), s, A(action)).code == "forbidden"
# the assigned welder limited to another station
assert check(P("weld.andy", (STALL2,)), s, A(action)).code == "forbidden"
assert check(P("weld.andy", (STALL1, STALL2)), s, A(action)).ok
# no welder recorded: a worker cannot, a lead can
nobody = dataclasses.replace(s, welder_id=None)
assert check(P("weld.andy"), nobody, A(action)).code == "forbidden"
assert check(P("lead.lou"), nobody, A(action)).ok
# lead and above pass without being the welder
for user in ("lead.lou", "sup.wes", "both.bo"):
assert check(P(user), s, A(action)).ok, user
def test_policy_lead_versus_supervisor() -> None:
s = state(CASES["review"])
for action in ("send_to_qc", "route_from_review", "return_to_station"):
d = check(P("lead.lou"), s, A(action), extra=EXTRA[action])
assert d.code == "forbidden", action
assert check(P("sup.wes"), s, A(action), extra=EXTRA[action]).ok, action
allowed, blocked = can(P("lead.lou"), s)
assert not {A.SEND_TO_QC, A.ROUTE_FROM_REVIEW, A.RETURN_TO_STATION} & set(allowed)
assert not {A.SEND_TO_QC, A.ROUTE_FROM_REVIEW, A.RETURN_TO_STATION} & set(blocked)
def test_policy_return_to_clerical_only_when_skipped() -> None:
body = EXTRA["return_to_clerical"]
d = check(P("lead.lou"), state(CASES["ready"]), A.RETURN_TO_CLERICAL, extra=body)
assert d.code == "wrong_status"
assert d.why == "This packet came through the warehouse. Send it back to the warehouse instead."
assert check(P("lead.lou"), state(CASES["ready_skipped"]), A.RETURN_TO_CLERICAL, extra=body).ok
d = check(P("lead.lou"), state(CASES["ready_skipped"]), A.RETURN_TO_CLERICAL, extra={})
assert d.code == "reason_required"
# ================================================================ 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(shop: Shop, action: str) -> dict[str, Any]:
"""Every request input the step can need."""
return {
"assign": {"station_id": shop.stall(), "welder_id": shop.uid("weld.andy")},
"submit": {"sign": SIGN},
"return_to_station": {"reason": REASON},
"send_to_qc": {"book": True, "spool": True, "location": "Bay 2", "sign": SIGN},
"route_from_review": {"dept": "09", "sign": SIGN},
"send_back_to_warehouse": {"reason": REASON},
"return_to_clerical": {"reason": REASON},
}.get(action, {})
def build(shop: Shop, case: str) -> dict[str, Any]:
"""A packet in the case's status, made through the API."""
if case == "warehouse":
return to_warehouse(shop, draft(shop))
if case == "ready":
return ready_packet(shop)
if case == "ready_skipped":
return ready_skipped(shop)
if case == "assigned":
return to_station(shop, ready_packet(shop))
if case == "in_progress":
return in_progress(shop)
if case == "in_progress_done":
return finish_stage(shop, in_progress(shop))
if case == "review":
return in_review(shop)
if case == "qc":
pkt = in_review(shop)
return ok(step(shop("sup.wes"), pkt, "send_to_qc", book=True, spool=True, location="Bay 2", sign=SIGN))
raise AssertionError(case)
@needs_pg
@pytest.mark.parametrize("case", list(CASES))
def test_api_matrix(shop: Shop, case: str) -> None:
"""Every refused step answers the oracle's code and changes nothing; `can` and `blocked` agree for every user.
Steps the oracle allows are not sent here (they would move the packet); test_api_success sends each one."""
c = CASES[case]
pkt = build(shop, case)
assert pkt["status"] == c.status and pkt["wh_skipped"] == c.skipped
for user in USERS:
mine = fresh(shop(user), pkt["id"])
for action in SHOP_ACTIONS:
want = expected(user, c, action)
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"])
if want[0] == 200:
continue
r = step(shop(user), pkt, action, **body_for(shop, action))
assert (r.status_code, r.json().get("code")) == want, (user, case, action, r.text)
if want[1] == "precondition" and action == "submit":
assert r.json()["detail"] == MSG_CHECKLIST
assert mine["blocked"]["submit"][0] == MSG_CHECKLIST
after = fresh(shop("admin.ann"), pkt["id"])
assert (after["status"], after["version"]) == (pkt["status"], pkt["version"])
SUCCESS = [
(case, action, user)
for case, action in (
("ready", "assign"),
("ready", "send_back_to_warehouse"),
("ready_skipped", "return_to_clerical"),
("assigned", "unassign"),
("assigned", "start"),
("assigned", "submit"),
("in_progress", "pull_back"),
("in_progress_done", "submit"),
("review", "return_to_station"),
("review", "send_to_qc"),
("review", "route_from_review"),
)
for user in USERS
if expected(user, CASES[case], action, it_reason=True)[0] == 200 and not (case == "assigned" and action == "submit")
]
TO_STATUS = {
"assign": "assigned",
"send_back_to_warehouse": "warehouse",
"return_to_clerical": "draft",
"unassign": "ready",
"start": "in_progress",
"submit": "review",
"pull_back": "ready",
"return_to_station": "assigned",
"send_to_qc": "qc",
"route_from_review": "ready",
}
@needs_pg
@pytest.mark.parametrize(("case", "action", "user"), SUCCESS)
def test_api_success(shop: Shop, case: str, action: str, user: str) -> None:
"""Each allowed user does each step; the admin only with an IT reason."""
pkt = build(shop, case)
assert action in fresh(shop(user), pkt["id"])["can"]
body = body_for(shop, action)
if user == "admin.ann":
refused(step(shop(user), pkt, action, **body), 400, "it_reason_required")
body["it_reason"] = IT
out = ok(step(shop(user), pkt, action, **body))
assert out["status"] == TO_STATUS[action]
if action == "route_from_review":
assert out["dept"]["code"] == "09"
@needs_pg
@pytest.mark.parametrize(
("case", "action"),
[
("ready", "send_back_to_warehouse"),
("ready_skipped", "return_to_clerical"),
("review", "return_to_station"),
],
)
def test_api_reason_required(shop: Shop, case: str, action: str) -> None:
pkt = build(shop, case)
for reason in (None, "", "short"):
body = {} if reason is None else {"reason": reason}
refused(step(shop("sup.wes"), pkt, action, **body), 400, "reason_required")
assert fresh(shop("sup.wes"), pkt["id"])["status"] == pkt["status"]
@needs_pg
@pytest.mark.parametrize(
("case", "action", "user"),
[
("in_progress_done", "submit", "weld.andy"),
("review", "send_to_qc", "sup.wes"),
("review", "route_from_review", "sup.wes"),
],
)
def test_api_signature_required(shop: Shop, case: str, action: str, user: str) -> None:
pkt = build(shop, case)
body = body_for(shop, action)
body.pop("sign")
r = refused(step(shop(user), pkt, action, **body), 400, "signature_required")
assert r.get("meaning")
assert fresh(shop(user), pkt["id"])["status"] == pkt["status"]
@needs_pg
def test_api_assigned_welder_and_station_limits(shop: Shop) -> None:
pkt = to_station(shop, ready_packet(shop))
# dup.dave works in 03 but is not the welder
refused(step(shop("dup.dave"), pkt, "start"), 403, "forbidden")
assert "start" not in fresh(shop("dup.dave"), pkt["id"])["can"]
# weld.andy limited to Stall 2 while the packet is at Stall 1
set_limits(shop.settings, "weld.andy", [shop.stall("03", "Stall 2")])
refused(step(shop("weld.andy"), pkt, "start"), 403, "forbidden")
mine = fresh(shop("weld.andy"), pkt["id"])
assert "start" not in mine["can"] and "start" not in mine["blocked"]
# the lead starts it (lead and above pass)
pkt = start(shop, pkt, by="lead.lou")
refused(step(shop("weld.andy"), pkt, "submit", sign=SIGN), 403, "forbidden")
# limits that include Stall 1: weld.andy works on it again
set_limits(shop.settings, "weld.andy", [shop.stall("03", "Stall 1"), shop.stall("03", "Stall 2")])
pkt = finish_stage(shop, pkt)
pkt = submit(shop, pkt, by="weld.andy")
# the lead may submit too: a second packet
other = finish_stage(shop, start(shop, to_station(shop, ready_packet(shop)), by="weld.andy"), by="lead.lou")
submit(shop, other, by="lead.lou")
@needs_pg
def test_api_lead_review_read_only(shop: Shop) -> None:
pkt = in_review(shop)
lou = fresh(shop("lead.lou"), pkt["id"])
for action in ("send_to_qc", "route_from_review", "return_to_station"):
assert action not in lou["can"] and action not in lou["blocked"], action
refused(step(shop("lead.lou"), pkt, action, **body_for(shop, action)), 403, "forbidden")
wes = fresh(shop("sup.wes"), pkt["id"])
assert {"send_to_qc", "route_from_review", "return_to_station"} <= set(wes["can"])
@needs_pg
def test_api_return_to_clerical_only_when_skipped(shop: Shop) -> None:
pkt = ready_packet(shop)
r = refused(step(shop("lead.lou"), pkt, "return_to_clerical", reason=REASON), 403, "wrong_status")
assert r["detail"] == "This packet came through the warehouse. Send it back to the warehouse instead."
skipped = ready_skipped(shop)
out = ok(step(shop("lead.lou"), skipped, "return_to_clerical", reason=REASON))
assert out["status"] == "draft" and out["return_note"] == REASON
def held_text(no: int, depts: str = "03 Weld Parts") -> str:
return f"Change order CO-{no} must be verified first by: {depts}."
@needs_pg
def test_api_co_hold_set(shop: Shop) -> None:
"""A CO requiring 03 holds SUBMIT, SEND_TO_QC and ROUTE_FROM_REVIEW, never ASSIGN or START."""
pkt = ready_packet(shop)
pkt = issue_co(shop, pkt, ["03"])
assert pkt["open_cos"] == [1]
wes = fresh(shop("sup.wes"), pkt["id"])
assert "assign" in wes["can"] and "assign" not in wes["blocked"]
pkt = to_station(shop, pkt)
andy = fresh(shop("weld.andy"), pkt["id"])
assert "start" in andy["can"]
pkt = finish_stage(shop, start(shop, pkt))
andy = fresh(shop("weld.andy"), pkt["id"])
assert "submit" not in andy["can"] and andy["blocked"]["submit"] == [held_text(1)]
refused(step(shop("weld.andy"), pkt, "submit", sign=SIGN), 403, "co_hold", held_text(1))
refused(step(shop("admin.ann"), pkt, "submit", sign=SIGN, it_reason=IT), 403, "co_hold", held_text(1))
ok(verify(shop, pkt, 1, "03", "sup.wes"))
pkt = submit(shop, fresh(shop("weld.andy"), pkt["id"]))
pkt = issue_co(shop, pkt, ["03"])
wes = fresh(shop("sup.wes"), pkt["id"])
for action in ("send_to_qc", "route_from_review"):
assert action not in wes["can"] and wes["blocked"][action] == [held_text(2)], action
refused(step(shop("sup.wes"), pkt, action, **body_for(shop, action)), 403, "co_hold", held_text(2))
assert "return_to_station" in wes["can"]
# a pending VISUAL change order with hold: both sentences, joined by a space
with_pending = _pending_feed_co(shop, pkt["id"], hold=True)
pending = (
f"CO-{with_pending}: Clerical is reviewing a change from VISUAL. Held until Clerical approves or rejects it."
)
wes = fresh(shop("sup.wes"), pkt["id"])
assert wes["blocked"]["send_to_qc"] == [f"{held_text(2)} {pending}"] or wes["blocked"]["send_to_qc"] == [
f"{pending} {held_text(2)}"
]
ok(verify(shop, pkt, 2, "03", "sup.wes"))
wes = fresh(shop("sup.wes"), pkt["id"])
assert wes["blocked"]["send_to_qc"] == [pending]
run(
shop.settings,
"UPDATE change_orders SET hold = false WHERE packet_id = %s AND co_no = %s",
(pkt["id"], with_pending),
)
wes = fresh(shop("sup.wes"), pkt["id"])
assert "send_to_qc" in wes["can"] and "send_to_qc" not in wes["blocked"]
ok(step(shop("sup.wes"), wes, "send_to_qc", book=True, spool=True, location="Bay 2", sign=SIGN))
def _pending_feed_co(shop: Shop, pid: int, *, hold: bool) -> int:
"""A pending VISUAL change order by SQL, as the feed opens one. Returns its number."""
with app_conn(shop.settings) as conn:
no = conn.execute(
"UPDATE packets SET next_co_no = next_co_no + 1 WHERE id = %s RETURNING next_co_no - 1", (pid,)
).fetchone()[0]
conn.execute(
"INSERT INTO change_orders (packet_id, co_no, origin, review, minor, hold, held_at, ecn, reason, vis)"
" VALUES (%s, %s, 'visual', 'pending', false, %s, CASE WHEN %s THEN now() END, 'VISUAL',"
" 'VISUAL changed the packet: qty 2 -> 3', %s)",
(pid, no, hold, hold, Jsonb({"dwg_change": False, "changes": ["qty 2 -> 3"]})),
)
return int(no)
@needs_pg
def test_api_stale_expect_after_pull_back(shop: Shop) -> None:
pkt = finish_stage(shop, in_progress(shop))
ok(step(shop("sup.wes"), pkt, "pull_back"))
r = refused(
step(shop("weld.andy"), pkt, "submit", sign=SIGN), 409, "stale_step", "Someone else already took this step."
)
assert r["current"]["status"] == "ready"
Reference in New Issue Copy Permalink