dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
tests/test_trace.py
"""MFG-16 and SC-MFG-03: recall a green lot forward to every roasted and bagged lot and where each went, and a bag back to its beans and the
supplier's receipt - derived from the orders, never typed, one statement per level whatever the number of lots."""
from decimal import Decimal
from conftest import (
ACCOUNTANT,
D16,
ENGINEER,
OPERATOR,
PLANNER,
QUALITY,
api,
issue,
line,
must,
new_order,
on_day,
output,
refused,
release,
roast_bom,
)
from erp_platform.testing import count_statements
G, R1, R2, B1 = "G-2611-ETH-07", "R-20261116-01", "R-20261117-02", "B-20261117-01"
def recall_world(nb):
"""G-2611-ETH-07 went into two roasts (R1, R2); R1 was packed into a bag lot B1 (100 bags)."""
roast_bom()
nb.available[("GREEN-ETH", "ROASTERY")] = Decimal("500")
for due, lot in (("2026-11-16", R1), ("2026-11-17", R2)):
oid = new_order(due=due)
must(release(oid))
must(issue(oid))
must(output(oid, lot=lot, chaff=None))
must(api("POST", "/boms/BAG-ETH-250:production/versions", ENGINEER, {
"version": 1, "valid_from": "2026-11-16", "output_quantity": "1", "output_unit_code": "C62", "company_code": "UK01",
"lines": [line("ROAST-ETH", "0.250"), line("POUCH-250", "1", "C62"), line("LABEL-ETH", "1", "C62")]}))
must(api("POST", "/boms/BAG-ETH-250:production/versions/1/release", ENGINEER))
nb.costs["BAG-ETH-250"] = Decimal("3.0725")
nb.available.update({("ROAST-ETH", "ROASTERY"): Decimal("100"), ("POUCH-250", "ROASTERY"): Decimal("100"), ("LABEL-ETH", "ROASTERY"): Decimal("100")})
nb.onhand.update({("ROAST-ETH", "ROASTERY/RAW"): Decimal("100"), ("POUCH-250", "ROASTERY/RAW"): Decimal("100"), ("LABEL-ETH", "ROASTERY/RAW"): Decimal("100")})
bag = must(api("POST", "/production-orders", PLANNER, {"company_code": "UK01", "product_id": "BAG-ETH-250", "warehouse_id": "ROASTERY",
"planned_quantity": "100", "due_on": "2026-11-17"}))["production_order_id"]
must(release(bag))
with on_day(D16):
must(api("POST", f"/production-orders/{bag}/issue", OPERATOR, {"location_id": "ROASTERY/RAW", "lines": [
{"product_id": "ROAST-ETH", "quantity": "25.000", "lot_code": R1}, {"product_id": "POUCH-250", "quantity": "100", "unit_code": "C62"},
{"product_id": "LABEL-ETH", "quantity": "100", "unit_code": "C62"}]}))
must(api("POST", f"/production-orders/{bag}/output", OPERATOR, {"location_id": "ROASTERY/FG", "quantity": "100", "lot_code": B1}))
nb.traces[("GREEN-ETH", G)] = {"product_id": "GREEN-ETH", "code": G, "kind": "lot", "on_hand": [{"location_id": "ROASTERY/RAW", "quantity_base": "300"}],
"history": [{"document_id": "PO-77/1", "bp_id": "SUPPLIER-ETH", "to_location_id": "ROASTERY/RAW", "quantity_base": "500"}],
"delivered_to": []}
nb.traces[("ROAST-ETH", R1)] = {"product_id": "ROAST-ETH", "code": R1, "kind": "lot", "history": [], "on_hand": [],
"delivered_to": [{"bp_id": "SHOP-4", "document_id": "TR-9", "quantity_base": "10"}]}
nb.traces[("ROAST-ETH", R2)] = {"product_id": "ROAST-ETH", "code": R2, "kind": "lot", "history": [], "on_hand": [{"location_id": "ROASTERY/FG", "quantity_base": "82.5"}],
"delivered_to": []}
nb.traces[("BAG-ETH-250", B1)] = {"product_id": "BAG-ETH-250", "code": B1, "kind": "lot", "history": [], "on_hand": [],
"delivered_to": [{"bp_id": "CUST-1", "document_id": "DEL-1", "quantity_base": "40"}, {"bp_id": "CUST-2", "document_id": "DEL-2", "quantity_base": "60"}]}
return bag
def trace(product, lot, direction="forward", who=None):
return api("GET", f"/trace/{product}/{lot}", who or QUALITY, {"direction": direction})
def test_mfg_16_a_green_lot_traces_forward_to_every_roast_and_bag_and_where_each_went(nb):
recall_world(nb)
got = must(trace("GREEN-ETH", G))
assert [(x["lot_code"], x["depth"]) for x in got["lots"]] == [(G, 0), (R1, 1), (R2, 1), (B1, 2)]
by = {x["lot_code"]: x["inventory"] for x in got["lots"]}
assert by[R1]["delivered_to"][0]["bp_id"] == "SHOP-4" and by[R2]["on_hand"][0]["quantity_base"] == "82.5"
assert [d["bp_id"] for d in by[B1]["delivered_to"]] == ["CUST-1", "CUST-2"]
assert got["unaccounted"] == [] # nothing unaccounted
assert {(e["input"]["lot_code"], e["output"]["lot_code"]) for e in got["edges"]} == {(G, R1), (G, R2), (R1, B1)}
assert {e["order"] for e in got["edges"] if e["output"]["lot_code"] == B1} and all(e["input_quantity"] for e in got["edges"])
def test_mfg_16_a_bag_traces_back_to_its_roast_its_beans_and_the_suppliers_receipt(nb):
recall_world(nb)
got = must(trace("BAG-ETH-250", B1, "back"))
assert [(x["lot_code"], x["depth"]) for x in got["lots"]] == [(B1, 0), (R1, 1), (G, 2)]
green = next(x for x in got["lots"] if x["lot_code"] == G)
assert green["inventory"]["history"][0]["bp_id"] == "SUPPLIER-ETH" and green["inventory"]["history"][0]["document_id"] == "PO-77/1"
def test_mfg_16_a_lot_inventory_cannot_answer_is_listed_not_hidden(nb):
recall_world(nb)
del nb.traces[("ROAST-ETH", R2)]
got = must(trace("GREEN-ETH", G))
assert got["unaccounted"] == [{"product_id": "ROAST-ETH", "lot_code": R2}]
nb.status["inventory"] = 503
assert api("GET", f"/trace/GREEN-ETH/{G}", QUALITY, {})[0] in (502, 503)
def test_mfg_16_an_unknown_lot_is_a_404_and_a_lot_only_inventory_knows_traces_as_itself(nb):
recall_world(nb)
refused(trace("GREEN-ETH", "NOPE"), 404)
nb.traces[("GREEN-ETH", "G-EXTRA")] = {"product_id": "GREEN-ETH", "code": "G-EXTRA", "history": [], "on_hand": [], "delivered_to": []}
got = must(trace("GREEN-ETH", "G-EXTRA"))
assert [x["lot_code"] for x in got["lots"]] == ["G-EXTRA"] and got["edges"] == []
refused(api("GET", f"/trace/GREEN-ETH/{G}", QUALITY, {"direction": "sideways"}), 422)
def test_mfg_16_only_quality_traces(nb):
recall_world(nb)
for who in (OPERATOR, PLANNER, ACCOUNTANT):
refused(trace("GREEN-ETH", G, who=who), 403)
def test_mfg_16_a_trace_walks_its_lot_only_and_its_statements_do_not_grow_with_unrelated_lots(nb, engine):
recall_world(nb)
from brick_manufacturing.infrastructure.tables import Repository
with count_statements(engine) as few:
must(trace("GREEN-ETH", G))
with engine.begin() as conn:
Repository(conn).facts.genealogy([{"output_product_id": "ROAST-ETH", "output_lot_code": f"X-{i}", "input_product_id": "GREEN-ETH",
"input_lot_code": f"Y-{i}", "production_order_id": f"UK01-MO-2026-9{i:04d}", "input_quantity": Decimal(1),
"output_quantity": Decimal(1)} for i in range(50)])
with count_statements(engine) as many:
got = must(trace("GREEN-ETH", G))
assert many.statements == few.statements, (few.statements, many.statements)
assert [x["lot_code"] for x in got["lots"]] == [G, R1, R2, B1] # none of the unrelated lots is in it