"""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"