PacketStation/tests/server/test_itp_holds.py

"""Phase 6 ITP hold points: the pure functions in domain/itp.py (hold codes, blocks, department pages, open holds),
and the packet read model's `itp_hold` filtered by the packet's department page group. The hold is a warning only:
no precondition reads it, so `blocked` never mentions it."""

from __future__ import annotations

from collections.abc import Iterator
from typing import Any

import pytest
from fakedir import FakeDirectory
from pshelpers import needs_pg
from qchelpers import (
    ALL_HOLDS,
    BLOCKS,
    DEPT_PAGES,
    HOLDS,
    NCR,
    g,
    patch_line,
    reject_line,
    sign_line,
    start_itp,
)
from shophelpers import Shop, draft, fresh, make_unit, ok, open_shop, ready_packet, ready_skipped, run

from packetstation.settings import Settings

SHOP_CODES = ["01", "02", "03", "04", "05", "06", "07", "08", "09", "10", "11", "12", "13", "14"]


def itpd() -> Any:
    from packetstation.domain import itp

    return itp


def template() -> dict[str, Any]:
    from packetstation.domain.itp_template import ITP_TEMPLATE

    out: dict[str, Any] = ITP_TEMPLATE
    return out


def hold_keys(holds: Any) -> list[str]:
    return [str(g(h, "n")) for h in holds]


# ================================================================ pure


@pytest.mark.parametrize(
    ("code", "hold"),
    [
        ("H", True),
        ("W", True),
        ("R/W", True),
        ("R/H", True),
        ("AU/W", True),
        ("R/AU", False),
        ("NA", False),
        ("R", False),
        ("A", False),
        ("AU", False),
        ("O", False),
        ("I", False),
        ("WH", False),
        ("H/R", False),
        ("", False),
    ],
)
def test_is_hold_code(code: str, hold: bool) -> None:
    assert bool(itpd().is_hold_code(code)) is hold


def test_template_shape() -> None:
    t = template()
    assert t["ver"] == "ITP_Generation_Revised_2"
    assert "O = Observe" in t["legend"] and "NA = Not Applicable" in t["legend"]
    assert [str(s["n"]) for s in t["sections"]] == [str(n) for n in range(1, 10)]
    assert [s["page"] for s in t["sections"]] == ["WP"] * 3 + ["C"] * 3 + ["FA"] * 3
    assert [str(s["n"]) for s in t["sections"] if s.get("sig")] == ["3", "4", "5", "6", "7", "8", "9"]
    assert [str(s["n"]) for s in t["sections"] if s.get("final")] == ["9"]
    items = list(itpd().items(t))
    assert len(items) == 70
    codes = {str(i["n"]): (i["code"], i["cust"]) for s in t["sections"] for i in s["items"]}
    assert codes["3.4"] == ("A", "W") and codes["4.2"] == ("O", "H") and codes["9.7"] == ("R", "H")


def test_blocks() -> None:
    out = list(itpd().blocks(template()))
    assert [str(g(b, "key")) for b in out] == list(BLOCKS)

    # sections may be numbers or the section dicts themselves (the server gives the dicts)
    def sec_no(s: Any) -> str:
        return str(s["n"]) if isinstance(s, dict) else str(s)

    assert {str(g(b, "key")): [sec_no(s) for s in g(b, "sections")] for b in out} == BLOCKS
    assert [str(g(b, "key")) for b in out if g(b, "final")] == ["9"]


@pytest.mark.parametrize("code", SHOP_CODES)
def test_dept_pages_shop(code: str) -> None:
    assert list(itpd().dept_pages(code)) == DEPT_PAGES[code]


@pytest.mark.parametrize("code", ["CL", "WH", "QC", "PM", "ZZ", ""])
def test_dept_pages_support_or_unknown(code: str) -> None:
    assert itpd().dept_pages(code) is None


def test_open_holds_from_the_template() -> None:
    t = template()
    holds = itpd().open_holds(t, {})
    assert sorted(hold_keys(holds), key=_key) == sorted(ALL_HOLDS, key=_key)
    assert len(holds) == 17
    for page, keys in HOLDS.items():
        assert sorted(hold_keys(itpd().open_holds(t, {}, pages=[page])), key=_key) == sorted(keys, key=_key), page
    wp_c = itpd().open_holds(t, {}, pages=["WP", "C"])
    assert len(wp_c) == 12


def test_open_holds_drop_done_lines_and_add_rejected() -> None:
    t = template()
    lines = {
        "3.4": {"key": "3.4", "result": "approve", "na": False},
        "4.2": {"key": "4.2", "result": "na", "na": True},
        "1.1": {"key": "1.1", "result": "reject", "ncr": None, "na": False},
        "1.2": {"key": "1.2", "result": "reject", "ncr": NCR, "na": False},
        "1.3": {"key": "1.3", "result": None, "cust_code": "H", "na": False},
        "4.4": {"key": "4.4", "result": None, "cust_code": "R", "na": False},  # a line code R: 4.4's PESCO code is O
    }
    keys = hold_keys(itpd().open_holds(t, lines))
    assert "3.4" not in keys and "4.2" not in keys
    assert {"1.1", "1.2", "1.3"} <= set(keys)
    # 4.4: the template customer code is H; the line's own code R replaces it, and the PESCO code is O
    assert "4.4" not in keys
    assert len(keys) == 17 - 3 + 3
    wp = itpd().open_holds(t, lines, pages=["WP"])
    assert sorted(hold_keys(wp), key=_key) == ["1.1", "1.2", "1.3"]
    rej = {str(g(h, "n")): h for h in wp}
    assert g(rej["1.1"], "rejected") is True and g(rej["1.1"], "ncr") is None
    assert g(rej["1.2"], "ncr") == NCR and g(rej["1.3"], "rejected") is False


def _key(n: str) -> tuple[int, ...]:
    return tuple(int(x) for x in n.split("."))


# ================================================================ through 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 _hold(shop: Shop, pid: int, user: str = "sup.wes") -> Any:
    return fresh(shop(user), pid)["itp_hold"]


@needs_pg
def test_itp_hold_by_department(shop: Shop) -> None:
    _, unit_id = make_unit(shop.settings)
    in_03 = ready_packet(shop, unit_id)
    assert _hold(shop, in_03["id"]) is None  # no ITP yet
    itp = start_itp(shop, unit_id)
    in_11 = ready_skipped(shop, unit_id, dept="11")
    in_09 = ready_skipped(shop, unit_id, dept="09")
    support = draft(shop, unit_id)
    run(shop.settings, "UPDATE packets SET dept_code = 'QC' WHERE id = %s", (support["id"],))

    h = _hold(shop, in_03["id"])
    assert h["no"] == itp["no"] and h["pages"] == ["WP"] and h["count"] == 1 and h["rejected"] is False
    assert hold_keys(h["items"]) == ["3.4"]
    item = h["items"][0]
    assert "W" in item["codes"] and item["t"] and item["rejected"] is False and item["ncr"] is None

    h = _hold(shop, in_11["id"])
    assert h["pages"] == ["FA"] and sorted(hold_keys(h["items"]), key=_key) == list(HOLDS["FA"])
    assert "9.6" in hold_keys(h["items"]) and "4.2" not in hold_keys(h["items"])
    h = _hold(shop, in_09["id"])
    assert h["pages"] == ["WP", "C"] and h["count"] == 12
    h = _hold(shop, support["id"], "clerk.clara")
    assert h["pages"] is None and h["count"] == 17 and len(h["items"]) == 17

    # the summary on the packet
    s = fresh(shop("sup.wes"), in_03["id"])["itp"]
    assert s["id"] == itp["id"] and s["no"] == itp["no"] and s["status"] == "open"
    assert s["done"] == 0 and s["total"] == 70 and s["holds"] == 17

    # a warning only: nothing in blocked names the ITP
    for pid in (in_03["id"], in_11["id"]):
        for user in ("sup.wes", "lead.lou", "qc.quinn", "clerk.clara", "wh.wade"):
            blocked = fresh(shop(user), pid)["blocked"]
            assert not any("ITP" in why for reasons in blocked.values() for why in reasons), (user, blocked)


@needs_pg
def test_itp_hold_follows_the_lines(shop: Shop) -> None:
    _, unit_id = make_unit(shop.settings)
    in_03 = ready_packet(shop, unit_id)
    itp = start_itp(shop, unit_id)
    q = shop("qc.quinn")
    # approving the only WP hold clears the 03 packet's hold
    itp = ok(sign_line(q, itp, "3.4"))
    assert _hold(shop, in_03["id"]) is None
    assert fresh(shop("sup.wes"), in_03["id"])["itp"]["holds"] == 16
    # a rejected WP line comes back as a rejected point, with its NCR when given
    itp = ok(reject_line(q, itp, "1.1", note="MTR heat does not match"))
    h = _hold(shop, in_03["id"])
    assert h["count"] == 1 and h["rejected"] is True
    assert h["items"][0]["n"] == "1.1" and h["items"][0]["rejected"] is True and h["items"][0]["ncr"] is None
    itp = ok(reject_line(q, itp, "1.2", ncr=NCR, note="WPS rev"))
    h = _hold(shop, in_03["id"])
    assert h["count"] == 2 and next(i for i in h["items"] if i["n"] == "1.2")["ncr"] == NCR
    # a customer hold code on a WP line adds a point; N/A takes a point away
    itp = ok(patch_line(q, itp, "2.1", cust_code="H"))
    assert _hold(shop, in_03["id"])["count"] == 3
    itp = ok(patch_line(q, itp, "2.1", na=True))
    assert _hold(shop, in_03["id"])["count"] == 2
    # the ITP read lists every open point, unfiltered
    assert itp["progress"]["holds"] == len(itp["holds"])
    assert {"1.1", "1.2", "4.2", "9.7"} <= set(hold_keys(itp["holds"]))
    # once the ITP is complete the hold is gone
    run(shop.settings, "UPDATE itps SET status = 'complete' WHERE id = %s", (itp["id"],))
    assert _hold(shop, in_03["id"]) is None
    run(shop.settings, "UPDATE itps SET status = 'open' WHERE id = %s", (itp["id"],))
    assert _hold(shop, in_03["id"])["count"] == 2


@needs_pg
def test_itp_hold_count_on_the_qc_home(shop: Shop) -> None:
    from qchelpers import at_qc

    _, unit_id = make_unit(shop.settings)
    start_itp(shop, unit_id)
    pkt = at_qc(shop, unit_id)
    home = ok(shop("qc.quinn").get("/api/qc/home"))
    card = next(c for c in home["columns"]["ready"] if c["id"] == pkt["id"])
    # dept 03 sees the WP page: 3.4
    assert card["itp_hold_count"] == 1