dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
tests/test_stories.py
"""One test per story (module.yaml stories, Given/When/Then), named for it: slice A. The business day is 2026-11-10 unless a test moves
it (a BOM is released for a LATER day, so an order due 2026-11-16 is released and worked on that day).
Slice B stories (MFG-01, 03, 06, 07, 08, 09 convert, 12, 14, 16, 17, 18, 23, 27, 32) are listed in README.md and have no test here."""
from decimal import Decimal
from conftest import (
ENGINEER,
OWNER,
PLANNER,
ROLES,
SYSTEM,
api,
close,
complete,
get_order,
issue,
line,
must,
new_order,
on_day,
output,
person,
refused,
release,
released_order,
roast_bom,
)
from brick_manufacturing.infrastructure.tables import Repository
PLANNER_AT_THE_ROASTER = person("pat@roastery.example", ROLES["operator"] + ["mfg:order:release"]) # reports AND may release
# ── MFG-02: a roast as a BOM with its planned yield ─────────────────────────────────────────────────────
def test_mfg_02_a_roast_is_a_bom_with_a_planned_yield():
out = roast_bom()
v = out["versions"][0]
assert out["bom_id"] == "ROAST-ETH:production" and v["status"] == "released" and v["valid_from"] == "2026-11-16"
assert Decimal(v["planned_yield_pct"]) == Decimal("84.0000") # 84.000 / 100.000, shown, never typed
assert [ln["kind"] for ln in v["lines"]] == ["component", "by_product"]
assert out["valid_today"] is None # valid from the 16th, not today
def test_mfg_02_a_released_version_is_never_edited_the_change_is_version_2():
roast_bom()
body = {"version": 1, "valid_from": "2026-12-01", "output_quantity": "85.000", "output_unit_code": "KGM",
"lines": [line("GREEN-ETH", "100.000")]}
refused(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER, body), words="never edited", gate="version_frozen")
must(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER, {**body, "version": 2}))
out = must(api("POST", "/boms/ROAST-ETH:production/versions/2/release", ENGINEER))
assert [v["status"] for v in out["versions"]] == ["released", "released"]
with on_day("2026-11-16T09:00:00+00:00"):
assert must(api("GET", "/boms/ROAST-ETH:production", ENGINEER))["valid_today"] == 1
with on_day("2026-12-01T09:00:00+00:00"):
assert must(api("GET", "/boms/ROAST-ETH:production", ENGINEER))["valid_today"] == 2 # two versions never overlap
def test_mfg_02_valid_from_must_be_after_today_and_after_the_last_released_version():
body = {"version": 1, "valid_from": "2026-11-10", "output_quantity": "84.000", "output_unit_code": "KGM",
"lines": [line("GREEN-ETH", "100.000")], "company_code": "UK01"}
must(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER, body))
refused(api("POST", "/boms/ROAST-ETH:production/versions/1/release", ENGINEER), words="not after today", gate="valid_from")
must(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER, {**body, "valid_from": "2026-11-16"})) # a draft is rewritten
must(api("POST", "/boms/ROAST-ETH:production/versions/1/release", ENGINEER))
must(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER, {**body, "version": 2, "valid_from": "2026-11-16"}))
refused(api("POST", "/boms/ROAST-ETH:production/versions/2/release", ENGINEER), words="overlap", gate="overlap")
def test_mfg_02_a_component_that_leads_back_to_the_output_is_a_cycle():
roast_bom()
refused(api("POST", "/boms/GREEN-ETH:production/versions", ENGINEER,
{"version": 1, "valid_from": "2026-11-16", "output_quantity": "1.000", "output_unit_code": "KGM", "company_code": "UK01",
"lines": [line("ROAST-ETH", "1.000")]}), words="contain itself", gate="cycle") # through another BOM
refused(api("POST", "/boms/CHAFF:production/versions", ENGINEER,
{"version": 1, "valid_from": "2026-11-16", "output_quantity": "1.000", "output_unit_code": "KGM", "company_code": "UK01",
"lines": [line("CHAFF", "1.000")]}), words="itself", gate="cycle") # directly
def test_mfg_02_quantities_convert_exactly_or_are_refused(nb):
body = {"version": 1, "valid_from": "2026-11-16", "output_quantity": "1.000", "output_unit_code": "KGM", "company_code": "UK01",
"lines": [line("THIRD", "1.0000", "PCT")]}
refused(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER, body), 422, "does not convert exactly", "bounds")
body["lines"] = [line("GREEN-ETH", "250", "GRM")]
got = must(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER, body))
assert got["versions"][0]["lines"][0]["quantity_base"] == "0.2500"
# ── MFG-04, MFG-05, MFG-30: a second-level BOM, a recipe of usage sale, and the explode pos calls ───────
def latte(**over):
lines = [line("ROAST-HOUSE", "0.018"), line("MILK-WHOLE", "0.250", "LTR"), line("CUP-12OZ", "1", "C62"), line("LID-12OZ", "1", "C62"),
line("MILK-OAT", "0.250", "LTR", kind="modifier", replaces_product_id="MILK-WHOLE", modifier_id="MOD-OAT")]
body = {"version": 1, "valid_from": "2026-12-01", "output_quantity": "1", "output_unit_code": "C62", "lines": lines,
"company_code": "UK01", **over}
must(api("POST", "/boms/LATTE-REG:sale/versions", ENGINEER, body))
must(api("POST", "/boms/LATTE-REG:sale/versions/1/release", ENGINEER))
def test_mfg_05_a_drinks_recipe_is_a_bom_of_usage_sale_and_pos_explodes_it_with_a_modifier():
latte()
q = {"usage": "sale", "quantity": "1", "modifiers": "MOD-OAT", "on": "2026-12-02"}
got = must(api("GET", "/boms/LATTE-REG/explode", SYSTEM, q))
assert got["bom_id"] == "LATTE-REG:sale" and got["version"] == 1 and got["valid_from"] == "2026-12-01"
assert [(c["product_id"], c["quantity"], c["unit_code"]) for c in got["components"]] == [
("ROAST-HOUSE", "0.0180", "KGM"), ("MILK-OAT", "0.2500", "LTR"), ("CUP-12OZ", "1.0000", "C62"), ("LID-12OZ", "1.0000", "C62")]
assert must(api("GET", "/boms/LATTE-REG/explode", SYSTEM, q)) == got # the same request answers the same
plain = must(api("GET", "/boms/LATTE-REG/explode", SYSTEM, {**q, "modifiers": ""}))
assert plain["components"][1]["product_id"] == "MILK-WHOLE"
refused(api("GET", "/boms/LATTE-REG/explode", SYSTEM, {**q, "on": "2026-11-30"}), 404, "no released version")
refused(api("GET", "/boms/LATTE-REG/explode", SYSTEM, {**q, "modifiers": "MOD-NOPE"}), 422, "no modifier", "bounds")
def test_mfg_05_a_usage_sale_bom_is_never_produced():
latte()
refused(api("POST", "/production-orders", PLANNER, {"company_code": "UK01", "product_id": "LATTE-REG", "warehouse_id": "ROASTERY",
"planned_quantity": "1", "due_on": "2026-12-02"}), words="no production bill")
refused(api("POST", "/boms/LATTE-REG:sale/versions", ENGINEER,
{"version": 2, "valid_from": "2026-12-05", "output_quantity": "1", "output_unit_code": "C62",
"lines": [line("CUP-12OZ", "1", "C62"), line("CHAFF", "1", kind="by_product")]}), words="no by-products", gate="usage")
def test_mfg_04_the_bag_is_a_second_level_under_the_roast_and_the_graph_is_walked():
roast_bom()
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")]}))
# the roast can no longer contain the bag, however indirectly
refused(api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER,
{"version": 2, "valid_from": "2026-12-01", "output_quantity": "84.000", "output_unit_code": "KGM",
"lines": [line("BAG-ETH-250", "1", "C62")]}), words="contain itself", gate="cycle")
# ── MFG-09 (by hand), MFG-10: create and release ────────────────────────────────────────────────────────
def test_mfg_09_an_order_is_created_by_hand_from_the_yearly_series_against_the_valid_bom():
roast_bom()
first = new_order()
second = new_order()
assert (first, second) == ("UK01-MO-2026-00001", "UK01-MO-2026-00002")
got = get_order(first)
assert got["status"] == "created" and got["bom_id"] == "ROAST-ETH:production" and got["bom_version"] == 1
refused(api("POST", "/production-orders", PLANNER, {"company_code": "UK01", "product_id": "ROAST-ETH", "warehouse_id": "ROASTERY",
"planned_quantity": "84.000", "due_on": "2026-11-12"}), words="No released version", gate="bom_valid")
def test_mfg_10_release_freezes_a_copy_of_the_bom_and_a_later_change_never_reaches_it(nb):
roast_bom()
oid = new_order()
must(release(oid))
got = get_order(oid)
assert got["status"] == "released" and got["released_on"] == "2026-11-16"
assert got["planned_components"][0]["planned_base"] == "100.0000"
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 get_order(oid)["bom_version"] == 1 and get_order(oid)["planned_components"][0]["planned_base"] == "100.0000"
refused(release(oid), words="only a created order", gate="state") # a released order never returns to created
def test_mfg_10_a_shortage_refuses_the_release_unless_the_owner_allows_it(nb):
roast_bom()
nb.available[("GREEN-ETH", "ROASTERY")] = Decimal("60.000")
oid = new_order()
refused(release(oid), words="GREEN-ETH needs 100.0000 KGM, 60.000 available", gate="components_available")
must(api("POST", "/settings/release_with_shortage", OWNER, {"value": "true"}))
got = must(release(oid))
assert got["status"] == "released" and "GREEN-ETH needs" in got["shortage_warning"]
def test_mfg_10_release_needs_a_bom_valid_on_the_release_date():
roast_bom()
oid = new_order()
refused(api("POST", f"/production-orders/{oid}/release", PLANNER), words="No released version", gate="bom_valid") # today is the 10th
# ── MFG-11: issue by lot ────────────────────────────────────────────────────────────────────────────────
def test_mfg_11_issues_go_to_inventory_from_pending_rows_with_their_own_move_refs_within_the_tolerance(nb):
oid = released_order()
first = must(issue(oid, "100.000"))
assert first["sent"]["sent"] == 1 and first["status"] == "in_progress"
assert "value" not in first["movements"][0] # the shop-floor view never returns cost fields
mv = get_order(oid)["movements"][0]
assert mv["move_ref"] == f"mfg:{oid}:issue:1:1" and mv["status"] == "sent" and mv["value"] == "620.00" # inventory's value, 100 x 6.2000
more = must(issue(oid, "3.000"))
assert [m["move_ref"] for m in more["movements"]] == [f"mfg:{oid}:issue:1:1", f"mfg:{oid}:issue:1:2"]
assert more["movements"][1]["over_issue"] is True and more["movements"][0]["over_issue"] is False # 103 of 105 allowed
refused(issue(oid, "4.000"), words="tolerance of 5%", gate="over_issue_tolerance") # 107 > 105
assert len([c for c in nb.calls if c[0] == "issue"]) == 2 # the refusal sent nothing
assert get_order(oid)["planned_components"][0]["issued_base"] == "103.0000"
def test_mfg_11_zero_negative_and_huge_quantities_are_422_never_500():
oid = released_order()
for bad in ("0", "-1.000", "1e20", "100.00001", "abc", "99999999999999999999"):
status, body = issue(oid, bad)
assert status == 422, (bad, status, body)
def test_mfg_11_an_issue_inventory_refuses_is_recorded_rejected_and_counts_for_nothing(nb):
oid = released_order()
nb.onhand[("GREEN-ETH", "ROASTERY/RAW")] = Decimal("10.000")
refused(issue(oid, "100.000"), words="only 10", gate="on_hand")
got = get_order(oid)
assert got["movements"][0]["status"] == "rejected" and got["planned_components"][0]["issued_base"] == "0.0000"
nb.onhand[("GREEN-ETH", "ROASTERY/RAW")] = Decimal("500.000")
again = must(issue(oid, "100.000"))
assert again["movements"][1]["move_ref"] == f"mfg:{oid}:issue:1:2" and again["movements"][1]["status"] == "sent"
def test_mfg_11_only_a_component_of_the_order_is_issued_and_only_while_it_is_open():
oid = released_order()
refused(issue(oid, "1.000", product="CHAFF"), words="not a component", gate="component")
oid2 = new_order()
refused(issue(oid2, "1.000"), 404) # created, not released: an operator does not see it
# ── MFG-13: output ──────────────────────────────────────────────────────────────────────────────────────
def test_mfg_13_output_is_received_at_standard_with_the_chaff_at_nrv_and_names_its_input_lot(nb, engine):
oid = released_order()
must(issue(oid, "100.000"))
must(output(oid))
got = get_order(oid)
assert got["yield_pct"] == "82.50" and got["planned_yield_pct"] == "84.0000" # 82.50% against 84.00%
kinds = {m["kind"]: m for m in got["movements"] if m["kind"] != "issue"}
assert kinds["output"]["move_ref"] == f"mfg:{oid}:output:1" and kinds["output"]["value"] == "774.67" # 82.500 x 9.3899, rounded once
assert kinds["byproduct"]["move_ref"] == f"mfg:{oid}:byproduct:2:1" and kinds["byproduct"]["value"] == "0.00"
receipt = [b for k, b in nb.calls if k == "receipt" and b["source_kind"] == "production_receipt"][0]
assert receipt["lot_code"] == "R-20261116-01" and receipt["unit_cost"] == "9.3899"
assert receipt["input_lots"] == [{"product_id": "GREEN-ETH", "lot_code": "G-2611-ETH-07", "quantity": "100.0000"}]
with engine.begin() as conn: # the genealogy names the lot: derived from the order's issues, never typed
rows = Repository(conn).facts.lots_of(oid)
assert [(r["output_lot_code"], r["input_lot_code"], r["input_quantity"], r["output_quantity"]) for r in rows] == [
("R-20261116-01", "G-2611-ETH-07", Decimal("100.0000"), Decimal("82.5000"))]
def test_mfg_13_output_above_110_percent_of_plan_is_a_typo_unless_a_planner_confirms():
oid = released_order()
must(issue(oid, "100.000"))
refused(output(oid, "95.000"), words="looks like a typo", gate="output_plausibility")
refused(output(oid, "95.000", confirm_above_plan=True), words="Only a planner", gate="output_plausibility") # an operator cannot confirm
must(output(oid, "95.000", confirm_above_plan=True, who=PLANNER_AT_THE_ROASTER))
def test_mfg_13_output_needs_a_cost_from_inventory(nb):
oid = released_order()
nb.costs.pop("ROAST-ETH")
refused(output(oid), words="holds no cost", gate="standard_price")
# ── MFG-15: close ───────────────────────────────────────────────────────────────────────────────────────
def worked_order(nb=None, out="82.500", chaff="1.100"):
oid = released_order()
must(issue(oid, "100.000"))
must(output(oid, out, chaff=chaff))
must(complete(oid))
return oid
def test_mfg_15_close_splits_the_wip_balance_exactly_and_posts_one_journal_to_gl(nb):
oid = worked_order()
got = must(close(oid))
s = got["settlements"][0]
assert (s["wip_debit"], s["wip_credit"], s["standard_material"]) == ("620.00", "774.67", "608.93")
assert s["material_usage_variance"] == "11.07" # 620.00 - 608.93, incl. the 1.500 KGM below plan
# slice A has no operation confirmations, so no activity is absorbed: the conversion standard shows as activity variance
assert Decimal(s["material_usage_variance"]) + Decimal(s["activity_variance"]) == Decimal(s["wip_debit"]) - Decimal(s["wip_credit"])
assert got["wip_balance"] == "-154.67" and s["valuation_method"] == "standard"
journal = [b for k, b in nb.calls if k == "journal"][0]
assert journal["source_module"] == "manufacturing" and journal["event"] == "settle" and journal["source_document"] == oid
assert {x["purpose"] for x in journal["lines"]} == {"material_usage_variance", "activity_variance", "wip"}
assert sum(Decimal(x.get("debit", "0")) for x in journal["lines"]) == sum(Decimal(x.get("credit", "0")) for x in journal["lines"])
assert got["status"] == "closed" and got["sent"]["sent"] == 1
refused(close(oid), words="only a completed order", gate="state") # a closed order never reopens, and the retry posts nothing
assert len([c for c in nb.calls if c[0] == "journal"]) == 1
def test_mfg_15_the_roasting_loss_planned_in_the_yield_is_not_a_variance(nb):
oid = worked_order(out="84.000", chaff="1.200") # exactly the planned yield: 100 in, 84 out
s = must(close(oid))["settlements"][0]
assert s["material_usage_variance"] == "0.00" and s["standard_material"] == "620.00"
def test_mfg_15_an_average_cost_company_gets_a_valuation_adjustment_not_a_journal(nb):
must(api("POST", "/settings/valuation_method", OWNER, {"value": "average"}))
oid = worked_order()
got = must(close(oid))
assert [c[0] for c in nb.calls if c[0] in ("journal", "adjustment")] == ["adjustment"]
adj = nb.adjustments[f"mfg:{oid}:settle:1"]
assert adj["source_kind"] == "production_settlement" and adj["amount"] == "-154.67" and adj["on"] == "2026-11-18"
assert got["settlements"][0]["journal_id"] is None
def test_mfg_15_close_waits_while_a_movement_has_not_been_taken_by_inventory(nb):
oid = released_order()
nb.status["inventory"] = 503
assert must(issue(oid, "100.000"))["movements"][0]["status"] == "pending" # recorded, waiting; the operator is not blocked
must(complete(oid))
refused(close(oid), words=f"mfg:{oid}:issue:1:1", gate="outbox_empty")
nb.status["inventory"] = None
assert must(api("POST", "/outbox/dispatch", SYSTEM, {"production_order_id": oid}))["sent"] == 1
assert get_order(oid)["movements"][0]["value"] == "620.00"
assert must(api("POST", "/outbox/dispatch", SYSTEM, {"production_order_id": oid}))["sent"] == 0 # nothing is sent twice
# ── MFG-31: the controls are the owner's ────────────────────────────────────────────────────────────────
def test_mfg_31_a_planner_cannot_widen_the_controls_and_the_owner_changes_them_from_a_later_day():
got = api("POST", "/settings/release_with_shortage", PLANNER, {"value": "true"})
assert got[0] == 403 and "mfg:settings:controls" in str(got[1])
must(api("POST", "/settings/over_issue_tolerance_pct", OWNER, {"value": "8", "valid_from": "2026-12-01"}))
assert must(api("GET", "/settings", OWNER))["settings"]["over_issue_tolerance_pct"] == "5"
with on_day("2026-12-01T09:00:00+00:00"):
assert must(api("GET", "/settings", OWNER))["settings"]["over_issue_tolerance_pct"] == "8"
refused(api("POST", "/settings/over_issue_tolerance_pct", OWNER, {"value": "9"}), words="a day after today", gate="valid_from")
refused(api("POST", "/settings/over_issue_tolerance_pct", OWNER, {"value": "-1", "valid_from": "2026-12-05"}), 422, "not a valid", "bounds")