PacketStation/tests/server/test_policy_warehouse.py

"""The phase 4 release gate in the policy: heats_recorded over PacketState values, WH_RELEASE with both rules,
and that can() and check() agree. No database."""

from __future__ import annotations

import dataclasses
import itertools
from typing import Any

import pytest

from packetstation.policy import RULES, Action, PacketState, Principal, can, check, principal_from_roles
from packetstation.policy.preconditions import INPUT_ONLY, MESSAGES, PRECONDITIONS, heats_recorded

A = Action
USER = "00000000-0000-0000-0000-0000000000c1"
IT = "Ticket 4411: release by hand"
MSG_PULL = "Pull every material line or mark it short first."
MSG_HEAT = "Scan or type the heat number of every pulled line first. Use N/A for a part with no traceability."


def P(*roles: str) -> Principal:
    return principal_from_roles(USER, list(roles))


WH = P("warehouse")
WH_SUP = P("warehouse_supervisor")
ADMIN = P("admin")
LEAD = P("shop:03:lead")
SUP = P("shop:03:supervisor")
QC = P("qc")
CL = P("clerical")


def state(status: str = "warehouse", **kw: Any) -> PacketState:
    base = PacketState(
        id=1,
        status=status,
        dept="03",
        station_id=None,
        welder_id=None,
        wh_skipped=False,
        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,
        bom_missing_heat=0,
    )
    return dataclasses.replace(base, **kw)


def test_packet_state_defaults_to_no_missing_heat() -> None:
    s = PacketState(
        id=1,
        status="warehouse",
        dept="03",
        station_id=None,
        welder_id=None,
        wh_skipped=False,
        routed_to=None,
        vclosed=False,
        removed=False,
        source_kind="hand",
        weld_dept=True,
        co_hold=False,
    )
    assert s.bom_missing_heat == 0


def test_heats_recorded_is_a_state_precondition() -> None:
    assert MESSAGES["heats_recorded"] == MSG_HEAT
    assert PRECONDITIONS["heats_recorded"] is heats_recorded
    assert "heats_recorded" not in INPUT_ONLY
    assert RULES[A.WH_RELEASE].preconditions == ("bom_all_pulled_or_short", "heats_recorded")


@pytest.mark.parametrize(("missing", "fails"), [(0, False), (1, True), (2, True), (40, True)])
def test_heats_recorded_over_state(missing: int, fails: bool) -> None:
    got = heats_recorded(state(bom_missing_heat=missing), {})
    assert got == (MSG_HEAT if fails else None)


def test_heats_recorded_ignores_input() -> None:
    assert heats_recorded(state(bom_missing_heat=1), {"heat": "H1", "na": True}) == MSG_HEAT
    assert heats_recorded(state(bom_missing_heat=0), {"anything": 1}) is None


@pytest.mark.parametrize("who", [WH, WH_SUP])
def test_release_blocked_by_a_missing_heat(who: Principal) -> None:
    d = check(who, state(bom_missing_heat=1), A.WH_RELEASE)
    assert not d.ok and d.code == "precondition" and d.why == MSG_HEAT
    assert check(who, state(bom_missing_heat=0), A.WH_RELEASE).ok


def test_release_rules_in_order() -> None:
    d = check(WH, state(bom_ready=2, bom_missing_heat=1), A.WH_RELEASE)
    assert not d.ok and d.why == MSG_PULL
    allowed, blocked = can(WH, state(bom_ready=2, bom_missing_heat=1))
    assert A.WH_RELEASE not in allowed
    assert blocked[A.WH_RELEASE] == [MSG_PULL, MSG_HEAT]
    allowed, blocked = can(WH, state(bom_ready=3, bom_missing_heat=1))
    assert A.WH_RELEASE not in allowed and blocked[A.WH_RELEASE] == [MSG_HEAT]
    allowed, blocked = can(WH, state())
    assert A.WH_RELEASE in allowed and A.WH_RELEASE not in blocked


def test_admin_cannot_bypass_the_heat_rule() -> None:
    d = check(ADMIN, state(bom_missing_heat=1), A.WH_RELEASE, it_reason=IT)
    assert not d.ok and d.code == "precondition" and d.why == MSG_HEAT
    d = check(ADMIN, state(bom_missing_heat=0), A.WH_RELEASE, it_reason=IT)
    assert d.ok and d.admin_bypass
    d = check(ADMIN, state(bom_missing_heat=0), A.WH_RELEASE)
    assert not d.ok and d.code == "it_reason_required"


@pytest.mark.parametrize("who", [CL, QC, LEAD, SUP])
def test_others_are_refused_before_the_gate(who: Principal) -> None:
    d = check(who, state(bom_missing_heat=1), A.WH_RELEASE)
    assert not d.ok and d.code == "forbidden"
    allowed, blocked = can(who, state(bom_missing_heat=1))
    assert A.WH_RELEASE not in allowed and A.WH_RELEASE not in blocked


def test_the_gate_only_touches_wh_release() -> None:
    """A missing heat blocks nothing else the warehouse can do on a warehouse packet."""
    ok_allowed, ok_blocked = can(WH, state())
    bad_allowed, bad_blocked = can(WH, state(bom_missing_heat=3))
    assert set(ok_allowed) - set(bad_allowed) == {A.WH_RELEASE}
    assert set(bad_blocked) - set(ok_blocked) == {A.WH_RELEASE}


PRINCIPALS = {"wh": WH, "wh_sup": WH_SUP, "admin": ADMIN, "lead": LEAD, "sup": SUP, "qc": QC, "cl": CL}


@pytest.mark.parametrize(
    ("name", "status", "total", "ready", "missing", "co_hold"),
    [
        (n, st, t, r, m, h)
        for n, st, (t, r), m, h in itertools.product(
            PRINCIPALS, ("warehouse", "ready", "draft"), ((3, 3), (3, 2), (0, 0)), (0, 1), (False, True)
        )
    ],
)
def test_can_and_check_agree(name: str, status: str, total: int, ready: int, missing: int, co_hold: bool) -> None:
    who = PRINCIPALS[name]
    s = state(status, bom_total=total, bom_ready=ready, bom_missing_heat=missing, co_hold=co_hold)
    allowed, blocked = can(who, s)
    d = check(who, s, A.WH_RELEASE, it_reason=IT if who.admin else None)
    assert (A.WH_RELEASE in allowed) == d.ok
    who_passes = who in (WH, WH_SUP, ADMIN) and status == "warehouse"
    if not d.ok and who_passes:
        assert d.code in ("precondition", "co_hold")
        assert A.WH_RELEASE in blocked and d.why == blocked[A.WH_RELEASE][0]
    if d.ok:
        assert A.WH_RELEASE not in blocked
        assert total <= ready and missing == 0 and not co_hold


def test_receive_by_custody_in_the_policy() -> None:
    """The warehouse receives anywhere but filed (WH_RECEIVE_SHORT); the holder through EDIT_BOM."""
    for status in ("warehouse", "ready", "assigned", "in_progress", "review", "qc", "complete"):
        assert check(WH, state(status, open_shorts=1), A.WH_RECEIVE_SHORT).ok, status
    assert check(WH, state("filed", open_shorts=1), A.WH_RECEIVE_SHORT).code == "filed"
    assert check(WH, state("ready", open_shorts=0), A.WH_RECEIVE_SHORT).why == MESSAGES["short_open"]
    ready = state("ready", open_shorts=1)
    assert check(LEAD, ready, A.EDIT_BOM).ok and check(SUP, ready, A.EDIT_BOM).ok
    assert check(WH, ready, A.EDIT_BOM).code == "forbidden"
    for who in (QC, CL):
        assert check(who, ready, A.EDIT_BOM).code == "forbidden"
        assert check(who, ready, A.WH_RECEIVE_SHORT).code == "forbidden"