dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
tests/test_engineering.py
"""MFG-01, MFG-03, MFG-06, MFG-07, MFG-18: work centres and their rates, routings, the cost roll-up and its second-person release, and
what goes stale when an input changes - each from its Given/When/Then in the signed spec."""
from decimal import Decimal
from conftest import (
ACCOUNTANT,
ACCOUNTANT2,
ENGINEER,
OPERATOR,
PLANNER,
api,
line,
must,
new_order,
on_day,
person,
refused,
roast_bom,
)
WC = "/work-centres"
def work_centres():
must(api("POST", f"{WC}/ROASTER-1", ENGINEER, {"company_code": "UK01", "warehouse_id": "ROASTERY", "cost_centre": "ROASTERY",
"batch_capacity": "15.000", "capacity_unit_code": "KGM", "hours_per_day": "8",
"machine_rate": "45.0000", "labour_rate": "22.5000"}), 201)
must(api("POST", f"{WC}/PACK-1", ENGINEER, {"company_code": "UK01", "warehouse_id": "ROASTERY", "cost_centre": "PACKING",
"batch_capacity": "100.000", "capacity_unit_code": "C62", "hours_per_day": "8",
"labour_rate": "22.5000"}), 201)
def roast_routing(version=1, start="2026-11-16"):
return must(api("POST", "/routings/ROAST-ETH/versions", ENGINEER, {
"routing_version": version, "valid_from": start, "operations": [
{"op_no": 10, "description": "roast", "work_centre_id": "ROASTER-1", "setup_minutes": "10", "run_minutes": "20",
"per_quantity": "15.000", "basis": "input", "unit_code": "KGM"},
{"op_no": 20, "description": "cool and degas", "wait_hours": "12"}]}), 201)
def costed_roast():
work_centres()
roast_bom()
roast_routing()
def rollup(who=None, eff="2026-12-01"):
return api("POST", "/cost-rollups", who or ACCOUNTANT, {"company_code": "UK01", "effective_on": eff})
# ── MFG-01 ─────────────────────────────────────────────────────────────────────────────────────
def test_mfg_01_a_work_centre_is_written_down_with_its_capacity_and_what_an_hour_costs():
work_centres()
wc = must(api("GET", f"{WC}/ROASTER-1", ENGINEER))
assert wc["batch_capacity"] == "15.0000" and wc["cost_centre"] == "ROASTERY"
assert {(r["rate_kind"], r["rate_amount"]) for r in wc["rates"]} == {("machine", "45.0000"), ("labour", "22.5000")}
assert must(api("GET", WC, ENGINEER))["work_centres"][1]["work_centre_id"] == "ROASTER-1"
def test_mfg_01_the_code_is_fixed_and_the_cost_centre_must_be_open_in_gl():
work_centres()
body = {"company_code": "UK01", "warehouse_id": "ROASTERY", "cost_centre": "ROASTERY", "batch_capacity": "1", "capacity_unit_code": "KGM",
"hours_per_day": "8"}
refused(api("POST", f"{WC}/ROASTER-1", ENGINEER, body), 409, "exists", gate="work_centre_exists")
refused(api("POST", f"{WC}/X-1", ENGINEER, {**body, "cost_centre": "NOPE"}), 409, "no cost centre", gate="cost_centre")
refused(api("POST", f"{WC}/X-2", ENGINEER, {**body, "cost_centre": "OLD"}), 409, "closed", gate="cost_centre")
refused(api("POST", f"{WC}/X-3", ENGINEER, {**body, "hours_per_day": "25"}), 422, "24", gate="bounds")
refused(api("POST", f"{WC}/X-4", ENGINEER, {**body, "machine_rate": "-1"}), 422, "zero or more")
def test_mfg_01_a_rate_change_is_a_later_dated_row_never_an_edit():
work_centres()
refused(api("POST", f"{WC}/ROASTER-1/rates", ENGINEER, {"rate_kind": "machine", "rate_amount": "50"}), 409, "from a day after today",
gate="valid_from")
must(api("POST", f"{WC}/ROASTER-1/rates", ENGINEER, {"rate_kind": "machine", "rate_amount": "50.0000", "valid_from": "2026-12-01"}), 201)
refused(api("POST", f"{WC}/ROASTER-1/rates", ENGINEER, {"rate_kind": "machine", "rate_amount": "51", "valid_from": "2026-12-01"}), 409)
wc = must(api("GET", f"{WC}/ROASTER-1", ENGINEER))
assert [r["rate_amount"] for r in wc["rates"] if r["rate_kind"] == "machine"] == ["45.0000", "50.0000"]
refused(api("POST", f"{WC}/NOPE/rates", ENGINEER, {"rate_kind": "machine", "rate_amount": "1"}), 404)
refused(api("POST", f"{WC}/ROASTER-1/rates", ENGINEER, {"rate_kind": "fuel", "rate_amount": "1"}), 422)
def test_mfg_01_only_the_engineer_writes_and_an_operator_never_sees_a_rate():
work_centres()
body = {"company_code": "UK01", "warehouse_id": "ROASTERY", "cost_centre": "ROASTERY", "batch_capacity": "1", "capacity_unit_code": "KGM",
"hours_per_day": "8"}
refused(api("POST", f"{WC}/Z-1", PLANNER, body), 403)
refused(api("POST", f"{WC}/ROASTER-1/rates", OPERATOR, {"rate_kind": "machine", "rate_amount": "1", "valid_from": "2026-12-01"}), 403)
assert "rates" not in must(api("GET", f"{WC}/ROASTER-1", OPERATOR))
# ── MFG-03 ─────────────────────────────────────────────────────────────────────────────────────
def test_mfg_03_a_routing_plans_batches_minutes_and_cost_and_waiting_costs_nothing():
costed_roast()
plan = must(api("GET", "/routings/ROAST-ETH/plan", ENGINEER, {"planned_quantity": "84.000", "on": "2026-11-16"}))
op10, op20 = plan["operations"]
assert (op10["batches"], op10["minutes"], op10["cost"]) == (7, "150", "168.75") # ceil(100/15) = 7; 10 + 7 x 20
assert (op20["minutes"], op20["wait_hours"], op20.get("cost")) == ("0", "12", "0.00")
assert plan["total_cost"] == "168.75" and plan["lead_days"] == 1 # 14.5 hours -> one day
assert "cost" not in must(api("GET", "/routings/ROAST-ETH/plan", OPERATOR, {"planned_quantity": "84.000", "on": "2026-11-16"}))["operations"][0]
def test_mfg_03_a_routing_is_versioned_and_dated_like_a_bom():
costed_roast()
v = must(api("GET", "/routings/ROAST-ETH", OPERATOR))
assert v["valid_today"] is None and [x["routing_version"] for x in v["versions"]] == [1]
ops = [{"op_no": 10, "work_centre_id": "ROASTER-1", "run_minutes": "30"}]
refused(api("POST", "/routings/ROAST-ETH/versions", ENGINEER, {"routing_version": 2, "valid_from": "2026-11-16", "operations": ops}), 409,
"never overlap", gate="valid_from")
refused(api("POST", "/routings/ROAST-ETH/versions", ENGINEER, {"routing_version": 1, "valid_from": "2026-12-01", "operations": ops}), 409)
refused(api("POST", "/routings/ROAST-ETH/versions", ENGINEER, {"routing_version": 2, "valid_from": "2026-11-10", "operations": ops}), 409,
"starts after today", gate="valid_from")
refused(api("POST", "/routings/ROAST-ETH/versions", ENGINEER,
{"routing_version": 2, "valid_from": "2026-12-01", "operations": [{"op_no": 10, "work_centre_id": "GHOST", "run_minutes": "1"}]}),
409, "does not exist", gate="work_centre")
refused(api("POST", "/routings/ROAST-ETH/versions", ENGINEER,
{"routing_version": 2, "valid_from": "2026-12-01", "operations": [{"op_no": 10, "work_centre_id": "ROASTER-1"}]}), 409, "takes no time")
refused(api("POST", "/routings/ROAST-ETH/versions", ENGINEER,
{"routing_version": 2, "valid_from": "2026-12-01", "operations": [{"op_no": 10, "work_centre_id": "ROASTER-1", "run_minutes": "1",
"basis": "input"}]}), 422, "unit_code", gate="bounds")
must(api("POST", "/routings/ROAST-ETH/versions", ENGINEER, {"routing_version": 2, "valid_from": "2026-12-01", "operations": ops}), 201)
assert must(api("GET", "/routings/ROAST-ETH", OPERATOR))["versions"][1]["valid_from"] == "2026-12-01"
refused(api("POST", "/routings/ROAST-ETH/versions", OPERATOR, {"routing_version": 3, "valid_from": "2027-01-01", "operations": ops}), 403)
refused(api("GET", "/routings/NOPE", OPERATOR), 404)
def test_mfg_03_the_order_freezes_the_routing_at_release_and_needs_one_valid_on_the_day():
work_centres()
roast_bom(valid_from="2026-11-12")
roast_routing() # starts 2026-11-16, four days after the BOM
oid = new_order(due="2026-11-16")
with on_day("2026-11-16T09:00:00+00:00"):
got = must(api("POST", f"/production-orders/{oid}/release", PLANNER))
assert got["routing_version"] == 1
refused(api("POST", "/production-orders", PLANNER, {"company_code": "UK01", "product_id": "ROAST-ETH", "warehouse_id": "ROASTERY",
"planned_quantity": "84", "due_on": "2026-11-12"}), 409, "No routing", gate="routing_valid")
# ── MFG-06 ─────────────────────────────────────────────────────────────────────────────────────
def test_mfg_06_the_roast_rolls_up_to_9_3899_split_material_and_conversion():
costed_roast()
r = must(rollup(), 201)
assert r["rollup_id"] == "UK01-CR-2026-12" and r["status"] == "computed"
roast = next(x for x in r["lines"] if x["product_id"] == "ROAST-ETH")
# (100 x 6.2000 + 168.75 - 0.00) / 84 = 788.75 / 84, divided once at the end
assert (roast["total"], roast["material"], roast["conversion"], roast["by_product_credit"]) == ("9.3899", "7.3810", "2.0089", "0.0000")
assert roast["bom_id"] == "ROAST-ETH:production" and roast["bom_version"] == 1
assert r["used"] == {"boms": ["ROAST-ETH:production"], "routings": ["ROAST-ETH"], "work_centres": ["ROASTER-1"]}
assert must(api("GET", "/cost-rollups", ACCOUNTANT))["cost_rollups"] == ["UK01-CR-2026-12"]
assert must(rollup(), 201)["rollup_id"] == "UK01-CR-2026-12-2"
def test_mfg_06_a_component_with_no_price_stops_the_run_with_the_product_named(nb):
costed_roast()
nb.costs.pop("GREEN-ETH")
refused(rollup(), 409, "GREEN-ETH", gate="standard_price")
assert must(api("GET", "/cost-rollups", ACCOUNTANT))["cost_rollups"] == []
def test_mfg_06_the_bag_costs_the_roast_at_its_rolled_up_standard_plus_components_and_packing(nb):
costed_roast()
nb.costs.update({"POUCH-250": Decimal("0.3000"), "LABEL-ETH": Decimal("0.0500")})
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))
must(api("POST", "/routings/BAG-ETH-250/versions", ENGINEER, {"routing_version": 1, "valid_from": "2026-11-16", "operations": [
{"op_no": 10, "description": "pack", "work_centre_id": "PACK-1", "run_minutes": "1", "per_quantity": "1", "basis": "output"}]}), 201)
r = must(rollup(), 201)
bag = next(x for x in r["lines"] if x["product_id"] == "BAG-ETH-250")
# 0.250 x 9.3899 + 0.3000 + 0.0500 + 1 minute of PACK-1 labour (22.5 / 60 = 0.375) = 3.072475
assert (bag["total"], bag["material"], bag["conversion"]) == ("3.0725", "2.6975", "0.3750")
assert r["used"]["boms"] == ["BAG-ETH-250:production", "ROAST-ETH:production"]
def test_mfg_06_only_a_cost_accountant_runs_and_reads_a_rollup():
costed_roast()
refused(rollup(PLANNER), 403)
refused(rollup(OPERATOR), 403)
refused(api("GET", "/cost-rollups", OPERATOR), 403)
refused(api("POST", "/cost-rollups", ACCOUNTANT, {"company_code": "UK01", "effective_on": "not-a-date"}), 422)
refused(api("POST", "/cost-rollups", ACCOUNTANT, {"company_code": "DE01", "effective_on": "2026-12-01"}), 409, "No production BOM", gate="bom_valid")
# ── MFG-07 ─────────────────────────────────────────────────────────────────────────────────────
def test_mfg_07_a_second_person_releases_each_standard_to_inventory(nb):
costed_roast()
rid = must(rollup(), 201)["rollup_id"]
refused(api("POST", f"/cost-rollups/{rid}/approve", ACCOUNTANT), 409, "ran this roll-up yourself", gate="second_person")
# the same person under either identity is still the same person
refused(api("POST", f"/cost-rollups/{rid}/approve", person(ACCOUNTANT_EMAIL, ["mfg:cost:approve"], sub="a-new-sub")), 409, gate="second_person")
refused(api("POST", f"/cost-rollups/{rid}/approve", person("changed@roastery.example", ["mfg:cost:approve"], sub=ACCOUNTANT_EMAIL)), 409,
gate="second_person")
refused(api("POST", f"/cost-rollups/{rid}/approve", PLANNER), 403)
assert nb.standard_prices == {}
got = must(api("POST", f"/cost-rollups/{rid}/approve", ACCOUNTANT2))
assert got["status"] == "released" and got["approved_by"] == "second.accountant@roastery.example"
assert got["sent"] == {"sent": 1, "rejected": [], "pending": 0}
sent = nb.standard_prices[("UK01", "ROAST-ETH", "2026-12-01", f"mfg:{rid}")]
assert (sent["standard_price"], sent["released_by"]) == ("9.3899", "second.accountant@roastery.example")
refused(api("POST", f"/cost-rollups/{rid}/approve", ACCOUNTANT2), 409, "released", gate="state")
def test_mfg_07_a_released_rollup_is_never_changed_and_a_retry_is_inventorys_replay(nb):
costed_roast()
rid = must(rollup(), 201)["rollup_id"]
must(api("POST", f"/cost-rollups/{rid}/approve", ACCOUNTANT2))
must(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER, {
"version": 2, "valid_from": "2026-12-01", "output_quantity": "85.000", "output_unit_code": "KGM", "lines": [line("GREEN-ETH", "100.000")]}))
must(api("POST", "/boms/ROAST-ETH:production/versions/2/release", ENGINEER))
assert must(api("GET", f"/cost-rollups/{rid}", ACCOUNTANT))["status"] == "released" # staleness never reaches a released roll-up
assert len(nb.standard_prices) == 1
def test_mfg_07_inventory_refusing_a_price_is_shown_not_hidden(nb):
costed_roast()
rid = must(rollup(), 201)["rollup_id"]
nb.refuse_standard = {"gate": "standard_price_taken", "reason": "a price from that day exists", "fix": "pick a later day"}
got = must(api("POST", f"/cost-rollups/{rid}/approve", ACCOUNTANT2))
assert got["sent"]["rejected"][0]["gate"] == "standard_price_taken" and got["outbox"][0]["status"] == "rejected"
def test_mfg_07_a_standard_price_starts_after_today():
costed_roast()
rid = must(rollup(), 201)["rollup_id"]
with on_day("2026-12-01T09:00:00+00:00"): # the effective day has come: inventory would refuse a price that starts today
refused(api("POST", f"/cost-rollups/{rid}/approve", ACCOUNTANT2), 409, "starts after today", gate="starts_after_today")
# ── MFG-18 ─────────────────────────────────────────────────────────────────────────────────────
def test_mfg_18_a_new_bom_version_makes_the_open_rollup_stale_and_a_new_run_uses_it():
costed_roast()
rid = must(rollup(), 201)["rollup_id"]
early = new_order(due="2026-12-05") # unreleased, due after v2 starts
before = new_order(due="2026-11-16") # unreleased, due before it
must(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER, {
"version": 2, "valid_from": "2026-12-01", "output_quantity": "85.000", "output_unit_code": "KGM", "lines": [line("GREEN-ETH", "100.000")]}))
must(api("POST", "/boms/ROAST-ETH:production/versions/2/release", ENGINEER))
got = must(api("GET", f"/cost-rollups/{rid}", ACCOUNTANT))
assert got["status"] == "stale" and "version 2 released" in got["stale_cause"]
refused(api("POST", f"/cost-rollups/{rid}/approve", ACCOUNTANT2), 409, "stale", gate="stale")
assert must(api("GET", f"/production-orders/{early}", ACCOUNTANT))["stale_note"] == "BOM v2 now valid"
assert must(api("GET", f"/production-orders/{before}", ACCOUNTANT))["stale_note"] is None
nr = must(rollup(), 201)
roast = next(x for x in nr["lines"] if x["product_id"] == "ROAST-ETH")
assert nr["rollup_id"] == "UK01-CR-2026-12-2" and roast["bom_version"] == 2 and roast["total"] == "9.2794" # 788.75 / 85
def test_mfg_18_a_rate_or_routing_change_marks_the_rollup_stale_too():
costed_roast()
rid = must(rollup(), 201)["rollup_id"]
must(api("POST", f"{WC}/ROASTER-1/rates", ENGINEER, {"rate_kind": "machine", "rate_amount": "50", "valid_from": "2026-12-01"}), 201)
assert must(api("GET", f"/cost-rollups/{rid}", ACCOUNTANT))["stale_cause"] == "machine rate of ROASTER-1 changes from 2026-12-01"
rid2 = must(rollup(), 201)["rollup_id"]
roast_routing(2, "2026-12-15")
assert "routing of ROAST-ETH version 2" in must(api("GET", f"/cost-rollups/{rid2}", ACCOUNTANT))["stale_cause"]
def test_mfg_18_a_released_order_keeps_its_frozen_bom():
costed_roast()
oid = new_order()
with on_day("2026-11-16T09:00:00+00:00"):
must(api("POST", f"/production-orders/{oid}/release", PLANNER))
must(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER, {
"version": 2, "valid_from": "2026-12-01", "output_quantity": "85.000", "output_unit_code": "KGM", "lines": [line("GREEN-ETH", "100.000")]}))
must(api("POST", "/boms/ROAST-ETH:production/versions/2/release", ENGINEER))
got = must(api("GET", f"/production-orders/{oid}", ACCOUNTANT))
assert got["bom_version"] == 1 and got["stale_note"] is None
ACCOUNTANT_EMAIL = "cost_accountant@roastery.example"