PacketStation/tests/server/test_candidates.py

"""Candidates from the generator's state 1: packet keys, routing, go-live, tracked keys, revs and the BOM rules."""

from __future__ import annotations

from datetime import date, timedelta
from decimal import Decimal
from typing import Any

import pytest
from feedhelpers import gen

from packetstation.feed import candidates, orders
from packetstation.feed.materials import index
from packetstation.feed.types import Material, WoKey, parse_material, parse_orders, parse_row

TODAY = date.today()
WINDOW = (TODAY - timedelta(days=14), TODAY + timedelta(days=30))


def k(base: str, lot: str, split: str = "0", sub: str = "0") -> WoKey:
    return WoKey(base, lot, split, sub)


def state(n: int = 1) -> dict[str, Any]:
    return gen.feed_state(n, TODAY)


def material_legs(feed: dict[str, Any]) -> set[WoKey]:
    return {k(m["wo"], m["lot"], m["split"], m["sub"]) for m in feed["materials"] if m["st"] in ("F", "R")}


def build(
    n: int = 1,
    *,
    tracked: set[WoKey] | None = None,
    golive: date | None = None,
    departments: tuple[str, ...] = (),
    mats: bool = True,
) -> candidates.Built:
    feed = state(n)
    rows = [parse_row(r) for r in feed["rows"]]
    materials = [parse_material(m) for m in feed["materials"]] if mats else None
    return candidates.build(
        rows,
        materials,
        material_legs(feed) if mats else None,
        resource_map=gen.RESOURCE_MAP,
        window=WINDOW,
        golive=golive,
        departments=departments,
        tracked_keys=tracked or set(),
        today=TODAY,
    )


def test_feed_file_shape() -> None:
    feed = state(1)
    assert set(feed) >= {"meta", "rows", "materials", "orders_by_base", "orders_by_project"}
    row = feed["rows"][0]
    assert set(row) >= {
        "wo",
        "lot",
        "split",
        "sub",
        "seq",
        "res",
        "dwg",
        "rev",
        "ps",
        "desc",
        "qty",
        "tgt",
        "rel",
        "want",
        "st",
        "barcode",
        "part",
    }
    words = {r["st"] for r in feed["rows"]}
    assert words <= {"Firmed", "Released", "Unreleased", "Closed", "Cancelled"}
    assert {m["st"] for m in feed["materials"]} <= {"F", "R", "U", "C", "X"}
    for r in feed["rows"]:
        assert isinstance(r["seq"], int) and isinstance(r["lot"], str) and isinstance(r["split"], str)
        assert isinstance(r["qty"], int | float) and not isinstance(r["qty"], bool)
        assert r["rel"] is None or date.fromisoformat(r["rel"])
    for m in feed["materials"]:
        assert isinstance(m["piece"], int) and (m["part"] is None or isinstance(m["part"], str))
    assert any(r["lot"] == "7A" for r in feed["rows"])
    assert any(m["part"] is None for m in feed["materials"])


def test_generator_is_deterministic(tmp_path: Any) -> None:
    a, b = tmp_path / "a", tmp_path / "b"
    gen.generate(a, TODAY)
    gen.generate(b, TODAY)
    for name in ("feed.1.json", "feed.2.json", "drawings.1.json", "drawings.2.json", "state"):
        assert (a / name).read_bytes() == (b / name).read_bytes(), name
    assert (a / "state").read_text() == "1"
    assert (a / "L900-0101 r1.pdf").read_bytes().startswith(b"%PDF-")
    assert (a / "L900-0101 r1.pdf").read_bytes() != (a / "L900-0101 r2.pdf").read_bytes()


def test_packet_keys() -> None:
    built = build()
    assert set(built.candidates) == set(gen.STATE1_KEYS)
    c = built.candidates[gen.KEY_WELD_PART]
    assert (c.project, c.unit, c.lot, c.dept, c.op, c.drawing) == (
        "L900",
        "L900001",
        "001",
        "03",
        "WELD-PART",
        "L900-0101",
    )
    assert c.base == "L900-01" and c.in_window and c.title == "Inlet spool 6 in"
    assert built.candidates[gen.KEY_LAYOUT_A].dept == "01"
    nd = built.candidates[gen.KEY_NO_DWG]
    assert nd.drawing == "" and nd.key.endswith("|(no drawing)")


def test_two_operations_at_one_resource_are_two_legs() -> None:
    c = build().candidates[gen.KEY_TWO_OPS]
    assert [(lg.key, lg.seq) for lg in c.legs] == [(k("L900-02", "1"), 10), (k("L900-02", "1"), 20)]
    assert [lg.tgt for lg in c.legs] == [Decimal(4), Decimal(2)]


def test_fold_and_ignore() -> None:
    c = build().candidates[gen.KEY_WELD_PART]
    folded = {(f.seq, f.op) for f in c.folded}
    assert folded == {(30, "MFG"), (50, "MAT_PIPE")}
    assert all(lg.op not in ("MFG", "FORKLIFT", "MAT_PIPE") for cand in build().candidates.values() for lg in cand.legs)
    # the LAYOUT packet of the same work order and lot folds the same operations
    assert {f.seq for f in build().candidates[gen.KEY_LAYOUT_A].folded} == {30, 50}
    assert candidates.route("MAT_STEEL", {}) == "fold" and candidates.route("FORKLIFT", gen.RESOURCE_MAP) == "ignore"
    assert candidates.route("NEWRES", gen.RESOURCE_MAP) is None


def test_partspecs_distinct_in_first_seen_order() -> None:
    c = build().candidates[gen.KEY_WELD_PART]
    assert c.partspecs == ["Cut to length per sheet 1", "Weld per WPS-11\nBack purge"]


def test_unmapped_counts() -> None:
    assert build().unmapped == {"NEWRES": 2}


def test_lettered_lot_makes_no_candidate() -> None:
    built = build()
    assert built.skipped_lots == {"L900-14/7A": 1}
    assert candidates.unit_number("L900", "7A") is None
    assert candidates.unit_number("L900", "1") == "L900001"
    assert candidates.unit_number("L900", "012") == "L900012"


def test_golive_ignores_old_releases() -> None:
    golive = TODAY + timedelta(days=gen.GOLIVE_OFFSET)
    built = build(golive=golive)
    assert gen.KEY_GOLIVE not in built.candidates
    assert [(r.wo, r.lot, r.seq) for r in built.ignored] == [("L900-13", "2", 10)]
    # tracked keys come through whatever their release date
    built = build(golive=golive, tracked={k("L900-13", "2")})
    assert gen.KEY_GOLIVE in built.candidates and built.ignored == []


def test_departments_filter() -> None:
    built = build(departments=("06",))
    assert built.candidates and {c.dept for c in built.candidates.values()} == {"06"}


def test_tracked_keys_outside_the_window_and_status() -> None:
    assert gen.KEY_RELEASED not in build().candidates  # Released: not in the window set
    tracked = {k("L900-06", "2"), k("L900-01", "1", "1"), k("L900-11", "1"), k("L900-12", "1")}
    built = build(tracked=tracked)
    rel = built.candidates[gen.KEY_RELEASED]
    assert not rel.in_window and rel.statuses == {k("L900-06", "2"): "R"}
    wp = built.candidates[gen.KEY_WELD_PART]
    assert wp.statuses == {k("L900-01", "1"): "F", k("L900-01", "1", "1"): "U"}
    assert {(lg.key.split, lg.seq) for lg in wp.legs} == {("0", 20), ("1", 20)}
    assert built.candidates["L900001|WELD-CODE|L900-1101"].statuses == {k("L900-11", "1"): "C"}
    assert built.candidates["L900001|WELD-CODE|L900-1201"].statuses == {k("L900-12", "1"): "X"}


def test_rel_min_want_max() -> None:
    c = build().candidates[gen.KEY_REV_A]  # legs L900-04 (rel -4, want 22) and L900-08-1 (rel -2, want 24)
    assert c.rel == TODAY - timedelta(days=4) and c.want == TODAY + timedelta(days=24)


def test_rev_highest_numeric() -> None:
    c = build().candidates[gen.KEY_REV_A]
    assert c.rev == "A"  # the non-numeric rev holds against an empty rev
    built = build(2, tracked={k("L900-01", "1", "1")})
    assert built.candidates[gen.KEY_WELD_PART].rev == "2"


def _row(seq: int, rev: str) -> Any:
    return parse_row(
        {
            "wo": "L900-20",
            "lot": "1",
            "split": "0",
            "sub": "0",
            "seq": seq,
            "res": "WELD-PART",
            "dwg": "D-1",
            "rev": rev,
            "qty": 1,
            "tgt": 1,
            "rel": TODAY.isoformat(),
            "st": "Firmed",
        }
    )


@pytest.mark.parametrize(
    "revs,expect",
    [(["1", "3", "2"], "3"), (["", "2"], "2"), (["2", "A"], "2"), (["A", "2"], "A"), (["10", "9"], "10")],
)
def test_rev_rule(revs: list[str], expect: str) -> None:
    rows = [_row(10 * (i + 1), r) for i, r in enumerate(revs)]
    built = candidates.build(
        rows,
        None,
        None,
        resource_map=gen.RESOURCE_MAP,
        window=WINDOW,
        golive=None,
        departments=(),
        tracked_keys=set(),
        today=TODAY,
    )
    assert built.candidates["L900001|WELD-PART|D-1"].rev == expect


def _bom(key: str, n: int = 1, tracked: set[WoKey] | None = None) -> dict[str, Any]:
    c = build(n, tracked=tracked).candidates[key]
    assert c.bom is not None
    return {b.item: b for b in c.bom}


def test_bom_split_items() -> None:
    bom = _bom(gen.KEY_LAYOUT_A)
    assert set(bom) == {"11", "12"}  # the piece 10 header has no part: not a line
    assert bom["11"].grp == "10" and bom["12"].grp == "10"
    assert bom["12"].cons is True and bom["11"].cons is False
    assert bom["11"].vkey == "L900-01|1|0|10|11"


def test_bom_main_loc_wip_sub_and_dwg_filter() -> None:
    bom = _bom(gen.KEY_WELD_PART)
    assert set(bom) == {"1", "2", "4"}  # piece 3 is on another drawing: dropped; piece 4 has a blank drawing
    assert bom["1"].loc == "A-01" and bom["1"].whs == "MAIN"  # MAIN is left out of loc
    assert bom["4"].loc == "SHOP R1"
    assert bom["2"].sub is True and bom["2"].na is True
    assert bom["1"].sub is False and bom["1"].issued == Decimal("12.5") and bom["1"].qty == Decimal("12.5")
    assert all(b.grp is None for b in bom.values())


def test_bom_duplicate_items() -> None:
    bom = _bom(gen.KEY_REV_A)
    assert set(bom) == {"1", "1-1", "2"}
    assert bom["1"].vkey.startswith("L900-04|") and bom["1-1"].vkey.startswith("L900-08-1|")
    assert bom["2"].istat == 0


def test_bom_two_operations() -> None:
    bom = _bom(gen.KEY_TWO_OPS)
    assert set(bom) == {"1", "2"} and bom["2"].istat == 3


def test_released_leg_counts_as_known() -> None:
    bom = _bom(gen.KEY_RELEASED, tracked={k("L900-06", "2")})
    assert set(bom) == {"1"}


def test_unreleased_leg_alone_is_unknown() -> None:
    # A2 (Unreleased) is not a material leg; on the WELD-PART packet the firmed leg still makes it known
    built = build(tracked={k("L900-01", "1", "1")})
    assert built.candidates[gen.KEY_WELD_PART].bom is not None


def test_mat_known_none() -> None:
    built = build()
    assert built.candidates[gen.KEY_NO_DWG].bom is None  # C has no material lines: not a material leg
    assert all(c.bom is None for c in build(mats=False).candidates.values())


def test_best_status_per_piece() -> None:
    key = k("L900-30", "1")

    def m(st: str, qty: int) -> Material:
        return parse_material(
            {
                "wo": "L900-30",
                "lot": "1",
                "split": "0",
                "sub": "0",
                "seq": 10,
                "piece": 1,
                "part": "P-1",
                "qty": qty,
                "issued": 0,
                "st": st,
                "dwg": "D-9",
            }
        )

    leg = candidates.Leg(
        key=key, seq=10, op="WELD-PART", qty=Decimal(1), tgt=Decimal(1), dwg="D-9", rev="1", partspecs="", st="F"
    )
    from packetstation.feed.materials import bom_for

    for lines, qty in (
        ([m("U", 3), m("F", 2), m("R", 1)], 1),
        ([m("U", 3), m("F", 2)], 2),
        ([m("C", 4), m("U", 3)], 3),
    ):
        bom = bom_for([leg], index(lines))
        assert [(b.item, b.qty) for b in bom] == [("1", Decimal(qty))]


def test_orders_unit_rule_on_fixture_orders() -> None:
    feed = state(1)
    by_base, by_project = parse_orders(feed["orders_by_base"]), parse_orders(feed["orders_by_project"])
    assert by_base is not None and by_project is not None

    def pick(unit: str, base: str, project: str = "L900", pin: str | None = None) -> tuple[str, str] | None:
        ch = orders.choose(
            unit_number=unit, base=base, by_base=by_base, by_project=by_project, project=project, pin=pin
        )
        return None if ch is None else (ch.order.cust_order_id, ch.via)

    assert pick("L900001", "L900-01") == (gen.ORDER_BAYOU, "unit_top_job")
    assert pick("L900002", "L900-04") == (gen.ORDER_ACME, "base")
    assert pick("L900002", "L900-07") == (gen.ORDER_ACME, "unit_top_job")
    assert pick("L900003", "L900-99") == (gen.ORDER_ACME, "project_open")
    assert pick("L901001", "L901-01", "L901") == (gen.ORDER_L901, "unit_top_job")
    assert pick("L901002", "L901-02", "L901") == (gen.ORDER_L901, "project_newest")
    assert pick("L900001", "L900-01", pin=gen.ORDER_ACME) == (gen.ORDER_ACME, "pin")
    assert pick("L900001", "L900-01", pin="C99-0000") == (gen.ORDER_BAYOU, "unit_top_job")  # not the project's
    assert pick("L902001", "L902-01", "L902") is None
    d = orders.derive(by_project["L900"][1])
    assert d["customer"] == "Bayou Chemical" and d["po"] == "PO-2002"
    assert d["ship_date"] == TODAY + timedelta(days=70)  # no promise date: the desired ship date