PacketStation/tests/server/test_policy_shop.py

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