216 lines
7.6 KiB
Python
216 lines
7.6 KiB
Python
"""The candidate fingerprint: which changes are major, minor or nothing, and what never counts."""
|
|||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import dataclasses
|
||
|
|
from collections.abc import Callable
|
||
|
|
from datetime import date
|
||
|
|
from decimal import Decimal
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from packetstation.feed import fingerprint as fp
|
||
|
|
from packetstation.feed.candidates import BomLine, Candidate, Folded, Leg
|
||
|
|
from packetstation.feed.types import WoKey
|
||
|
|
|
||
|
|
K1 = WoKey("L900-01", "1", "0", "0")
|
||
|
|
K2 = WoKey("L900-02", "1", "0", "0")
|
||
|
|
|
||
|
|
|
||
|
|
def leg(key: WoKey = K1, seq: int = 20, qty: Any = "2", tgt: Any = "3", **over: Any) -> Leg:
|
||
|
|
values: dict[str, Any] = {
|
||
|
|
"key": key,
|
||
|
|
"seq": seq,
|
||
|
|
"op": "WELD-PART",
|
||
|
|
"qty": Decimal(str(qty)),
|
||
|
|
"tgt": Decimal(str(tgt)),
|
||
|
|
"dwg": "L900-0101",
|
||
|
|
"rev": "1",
|
||
|
|
"partspecs": "",
|
||
|
|
"st": "F",
|
||
|
|
}
|
||
|
|
values.update(over)
|
||
|
|
return Leg(**values)
|
||
|
|
|
||
|
|
|
||
|
|
def line(item: str = "1", part: str | None = "PIPE-6", qty: Any = "2", **over: Any) -> BomLine:
|
||
|
|
values: dict[str, Any] = {
|
||
|
|
"item": item,
|
||
|
|
"qty": Decimal(str(qty)),
|
||
|
|
"part_id": part or "",
|
||
|
|
"desc": "Pipe",
|
||
|
|
"loc": "R1",
|
||
|
|
"whs": "SHOP",
|
||
|
|
"comm": "PIPE",
|
||
|
|
"cons": False,
|
||
|
|
"sub": False,
|
||
|
|
"na": False,
|
||
|
|
"grp": None,
|
||
|
|
"vkey": f"L900-01|1|0|20|{item}",
|
||
|
|
"issued": Decimal(0),
|
||
|
|
"istat": 2,
|
||
|
|
}
|
||
|
|
values.update(over)
|
||
|
|
return BomLine(**values)
|
||
|
|
|
||
|
|
|
||
|
|
def cand(**over: Any) -> Candidate:
|
||
|
|
values: dict[str, Any] = {
|
||
|
|
"key": "L900001|WELD-PART|L900-0101",
|
||
|
|
"project": "L900",
|
||
|
|
"unit": "L900001",
|
||
|
|
"lot": "1",
|
||
|
|
"dept": "03",
|
||
|
|
"op": "WELD-PART",
|
||
|
|
"drawing": "L900-0101",
|
||
|
|
"rev": "1",
|
||
|
|
"title": "Inlet spool",
|
||
|
|
"legs": [leg()],
|
||
|
|
"folded": [],
|
||
|
|
"partspecs": [],
|
||
|
|
"rel": date(2026, 10, 10),
|
||
|
|
"want": date(2026, 11, 5),
|
||
|
|
"bom": [line("1"), line("2", "ELB-6", "1")],
|
||
|
|
"statuses": {K1: "F"},
|
||
|
|
"base": "L900-01",
|
||
|
|
"in_window": True,
|
||
|
|
}
|
||
|
|
values.update(over)
|
||
|
|
return Candidate(**values)
|
||
|
|
|
||
|
|
|
||
|
|
def classify(before: Candidate, after: Candidate) -> fp.Change:
|
||
|
|
return fp.classify(fp.compute(before).parts, fp.compute(after).parts)
|
||
|
|
|
||
|
|
|
||
|
|
MAJOR_EDITS: dict[str, tuple[str, Callable[[], dict[str, Any]]]] = {
|
||
|
|
"leg added": ("legs", lambda: {"legs": [leg(), leg(K2, 10)]}),
|
||
|
|
"leg removed": ("legs", lambda: {"legs": []}),
|
||
|
|
"leg seq": ("legs", lambda: {"legs": [leg(seq=30)]}),
|
||
|
|
"qty": ("qty", lambda: {"legs": [leg(qty="3")]}),
|
||
|
|
"rev": ("rev", lambda: {"rev": "2"}),
|
||
|
|
"drawing": ("drawing", lambda: {"drawing": "L900-0102"}),
|
||
|
|
"bom item": ("bom", lambda: {"bom": [line("1"), line("3", "ELB-6", "1")]}),
|
||
|
|
"bom part": ("bom", lambda: {"bom": [line("1"), line("2", "ELB-6-SR", "1")]}),
|
||
|
|
"bom qty": ("bom", lambda: {"bom": [line("1"), line("2", "ELB-6", "2")]}),
|
||
|
|
"bom line added": ("bom", lambda: {"bom": [line("1"), line("2", "ELB-6", "1"), line("3", "TEE-6", "1")]}),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("name", list(MAJOR_EDITS))
|
||
|
|
def test_major_change(name: str) -> None:
|
||
|
|
part, edit = MAJOR_EDITS[name]
|
||
|
|
ch = classify(cand(), cand(**edit()))
|
||
|
|
assert ch.changed and not ch.is_minor
|
||
|
|
assert part in ch.major
|
||
|
|
|
||
|
|
|
||
|
|
MINOR_EDITS: dict[str, tuple[str, Callable[[], dict[str, Any]]]] = {
|
||
|
|
"tgt": ("tgt", lambda: {"legs": [leg(tgt="3.5")]}),
|
||
|
|
"want": ("want", lambda: {"want": date(2026, 11, 12)}),
|
||
|
|
"want cleared": ("want", lambda: {"want": None}),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("name", list(MINOR_EDITS))
|
||
|
|
def test_minor_change(name: str) -> None:
|
||
|
|
part, edit = MINOR_EDITS[name]
|
||
|
|
ch = classify(cand(), cand(**edit()))
|
||
|
|
assert ch.changed and ch.is_minor
|
||
|
|
assert ch.major == frozenset() and ch.minor == frozenset({part})
|
||
|
|
|
||
|
|
|
||
|
|
def test_minor_and_major_together_is_major() -> None:
|
||
|
|
ch = classify(cand(), cand(rev="2", want=date(2026, 12, 1)))
|
||
|
|
assert ch.major == {"rev"} and ch.minor == {"want"} and not ch.is_minor
|
||
|
|
|
||
|
|
|
||
|
|
NEVER_EDITS: dict[str, Callable[[], dict[str, Any]]] = {
|
||
|
|
"rel": lambda: {"rel": date(2026, 10, 1)},
|
||
|
|
"folded": lambda: {"folded": [Folded(K1, 30, "MFG", Decimal("0.5"))]},
|
||
|
|
"partspecs": lambda: {"partspecs": ["Weld per WPS-11"]},
|
||
|
|
"title": lambda: {"title": "Another title"},
|
||
|
|
"statuses": lambda: {"statuses": {K1: "R"}},
|
||
|
|
"issued": lambda: {"bom": [line("1", issued=Decimal(2)), line("2", "ELB-6", "1")]},
|
||
|
|
"whs": lambda: {"bom": [line("1", whs="MAIN"), line("2", "ELB-6", "1")]},
|
||
|
|
"loc": lambda: {"bom": [line("1", loc="A-01"), line("2", "ELB-6", "1")]},
|
||
|
|
"dept and op": lambda: {"dept": "06", "op": "WELD-CODE"},
|
||
|
|
"leg status": lambda: {"legs": [leg(st="C")]},
|
||
|
|
"leg partspecs": lambda: {"legs": [leg(partspecs="Back purge")]},
|
||
|
|
"in_window": lambda: {"in_window": False},
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("name", list(NEVER_EDITS))
|
||
|
|
def test_never_in_the_fingerprint(name: str) -> None:
|
||
|
|
before, after = cand(), cand(**NEVER_EDITS[name]())
|
||
|
|
assert not classify(before, after).changed
|
||
|
|
assert fp.compute(before).full == fp.compute(after).full
|
||
|
|
|
||
|
|
|
||
|
|
INVARIANT: dict[str, tuple[Callable[[], dict[str, Any]], Callable[[], dict[str, Any]]]] = {
|
||
|
|
"2 vs 2.0": (lambda: {"legs": [leg(qty="2")]}, lambda: {"legs": [leg(qty="2.0")]}),
|
||
|
|
"trailing zeros": (lambda: {"legs": [leg(tgt="1.5")]}, lambda: {"legs": [leg(tgt="1.500")]}),
|
||
|
|
"empty vs None rev": (lambda: {"rev": ""}, lambda: {"rev": None}),
|
||
|
|
"list order": (
|
||
|
|
lambda: {"legs": [leg(), leg(K2, 10)], "bom": [line("1"), line("2", "ELB-6", "1")]},
|
||
|
|
lambda: {"legs": [leg(K2, 10), leg()], "bom": [line("2", "ELB-6", "1"), line("1")]},
|
||
|
|
),
|
||
|
|
"drawing case and spaces": (lambda: {"drawing": "m00543"}, lambda: {"drawing": " M00 543 "}),
|
||
|
|
"part id case": (
|
||
|
|
lambda: {"bom": [line("1", "m00543")]},
|
||
|
|
lambda: {"bom": [line("1", "M00543")]},
|
||
|
|
),
|
||
|
|
"CRLF": (lambda: {"rev": "1\r\n"}, lambda: {"rev": "1\n"}),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("name", list(INVARIANT))
|
||
|
|
def test_invariance(name: str) -> None:
|
||
|
|
a, b = INVARIANT[name]
|
||
|
|
assert fp.compute(cand(**a())).full == fp.compute(cand(**b())).full
|
||
|
|
|
||
|
|
|
||
|
|
def test_lot_7a_stays_text() -> None:
|
||
|
|
k = WoKey("L900-14", "7A", "0", "0")
|
||
|
|
c = cand(legs=[leg(k, 10)], statuses={k: "F"})
|
||
|
|
values = fp.part_values(c)
|
||
|
|
assert values["legs"] == [["L900-14", "7A", "0", "0", 10]]
|
||
|
|
# A lot with leading zeros is not a number either
|
||
|
|
k0 = WoKey("L900-14", "007", "0", "0")
|
||
|
|
other = fp.part_values(cand(legs=[leg(k0, 10)]))
|
||
|
|
assert other["legs"][0][1] == "007"
|
||
|
|
assert fp.compute(c).full != fp.compute(cand(legs=[leg(k0, 10)])).full
|
||
|
|
|
||
|
|
|
||
|
|
def test_canonical_is_stable_json() -> None:
|
||
|
|
assert fp.canonical({"b": Decimal("2.0"), "a": ["x\r\n", None, ""]}) == b'{"a":["x",null,null],"b":"2"}'
|
||
|
|
assert fp.canonical({"d": date(2026, 10, 16)}) == b'{"d":"2026-10-16"}'
|
||
|
|
assert fp.canonical("é") == fp.canonical("é") # NFC
|
||
|
|
|
||
|
|
|
||
|
|
def test_parts_and_full() -> None:
|
||
|
|
f = fp.compute(cand())
|
||
|
|
assert set(f.parts) == set(fp.MAJOR) | set(fp.MINOR)
|
||
|
|
assert all(len(v) == 64 for v in f.parts.values()) and len(f.full) == 64
|
||
|
|
assert fp.compute(cand()) == f # deterministic
|
||
|
|
|
||
|
|
|
||
|
|
def test_no_before_is_all_changed() -> None:
|
||
|
|
ch = fp.classify(None, fp.compute(cand()).parts)
|
||
|
|
assert ch.major == frozenset(fp.MAJOR) and ch.minor == frozenset(fp.MINOR)
|
||
|
|
|
||
|
|
|
||
|
|
def test_unknown_materials_are_not_a_bom_change() -> None:
|
||
|
|
before = fp.compute(cand())
|
||
|
|
after = fp.compute(cand(bom=None))
|
||
|
|
assert "bom" not in classify(cand(), cand(bom=None)).major
|
||
|
|
merged = fp.merge(before.parts, after)
|
||
|
|
assert merged.parts["bom"] == before.parts["bom"] and merged.full == before.full
|
||
|
|
|
||
|
|
|
||
|
|
def test_dataclass_copy_is_unchanged() -> None:
|
||
|
|
c = cand()
|
||
|
|
assert fp.compute(dataclasses.replace(c)).full == fp.compute(c).full
|