dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
tests/test_planning.py
"""MFG-08, MFG-09, MFG-17, MFG-18 (the plan), MFG-23, MFG-27: MRP-lite proposes, the planner firms and converts, replenishment reads
what the plan will take. The numbers are the spec's: 1,200 bags - 300 on hand = 900; 945 KGM gross - 60 + 40 safety = 925, rounded UP to
multiples of 84 week by week = 1,008."""
from decimal import Decimal
from conftest import (
ACCOUNTANT,
ENGINEER,
OPERATOR,
PLANNER,
SYSTEM,
api,
as_role,
line,
must,
on_day,
refused,
roast_bom,
set_day,
)
from erp_platform import seam
from test_engineering import roast_routing, work_centres
D16 = "2026-11-16T09:00:00+00:00"
WH = "ROASTERY"
def world(nb, bag_lines=((("2026-11-24"), 700), (("2026-11-27"), 500))):
"""The MFG-08 'given': BAG-ETH-250 on BAG BOM, ROAST-ETH on its roast BOM + routing, stock, safety stock, a 84 KGM lot, forecasts."""
work_centres()
roast_bom()
roast_routing()
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)
nb.sites = [{"product_id": "BAG-ETH-250", "site_id": WH, "procurement_type": "make", "safety_stock": "0"},
{"product_id": "ROAST-ETH", "site_id": WH, "procurement_type": "make", "safety_stock": "40"},
{"product_id": "GREEN-ETH", "site_id": WH, "procurement_type": "buy", "safety_stock": "0"}]
nb.available.update({("BAG-ETH-250", WH): Decimal("300"), ("ROAST-ETH", WH): Decimal("60")})
nb.round_multiple["ROAST-ETH"] = Decimal("84")
nb.demand[("BAG-ETH-250", WH)] = [{"needed_on": d, "open_base": Decimal(q)} for d, q in bag_lines]
must(api("POST", f"/forecasts/{WH}", PLANNER, {"forecasts": [{"product_id": "ROAST-ETH", "week_start": w, "quantity": "180.000"}
for w in ("2026-11-16", "2026-11-23", "2026-11-30", "2026-12-07")]}))
set_day(D16)
def run(who=None, **kw):
return api("POST", "/mrp/runs", who or PLANNER, {"warehouse_id": WH, **kw})
def proposals(got):
return sorted((p["product_id"], p["due_on"], p["quantity"]) for p in got["planned_orders"])
def test_mfg_08_the_plan_nets_demand_stock_and_safety_stock_and_rounds_up_to_a_lot(nb):
world(nb)
got = must(run(), 201)
assert got["run_no"] == 1 and got["horizon_weeks"] == 4
assert proposals(got) == [("BAG-ETH-250", "2026-11-23", "900.0000"), # 1,200 - 300; one day before the first need
("ROAST-ETH", "2026-11-16", "168.0000"), ("ROAST-ETH", "2026-11-22", "420.0000"),
("ROAST-ETH", "2026-11-29", "168.0000"), ("ROAST-ETH", "2026-12-06", "252.0000")]
roast = [p for p in got["planned_orders"] if p["product_id"] == "ROAST-ETH"]
assert sum(Decimal(p["quantity"]) for p in roast) == Decimal("1008") # 12 x 84: 11 would be 924, one short
assert all(Decimal(p["quantity"]) % 84 == 0 for p in roast)
first = roast[0]
assert first["status"] == "proposed" and first["cause"]["on_hand"] == "60" and first["cause"]["safety_stock"] == "40"
assert first["planned_order_ref"] == "ROASTERY:ROAST-ETH:2026-11-16:run 1"
def test_mfg_08_each_roast_order_publishes_green_beans_at_100_over_84_of_its_output(nb):
world(nb)
must(run(), 201)
ask = {"product_id": "GREEN-ETH", "warehouse_id": WH, "from": "2026-11-16", "to": "2026-12-14"}
got = must(api("GET", "/dependent-requirements", SYSTEM, ask))
assert [(r["required_on"], r["quantity"], r["source"]) for r in got["requirements"]] == [
("2026-11-16", "200.0000", "planned"), ("2026-11-22", "500.0000", "planned"), ("2026-11-29", "200.0000", "planned"),
("2026-12-06", "300.0000", "planned")] # 1,200 KGM in four weeks (MFG-27)
bag_need = must(api("GET", "/dependent-requirements", SYSTEM, {**ask, "product_id": "ROAST-ETH"}))["requirements"]
assert [(r["required_on"], r["quantity"]) for r in bag_need] == [("2026-11-23", "225.0000")] # 900 x 0.250
def test_mfg_17_the_same_inputs_give_the_same_orders_and_a_rerun_supersedes_unfirmed_ones(nb):
world(nb)
first = must(run(), 201)
second = must(run(), 201)
assert second["run_no"] == 2 and proposals(second) == proposals(first)
status = {p["planned_order_ref"]: p["status"] for p in must(api("GET", "/planned-orders", PLANNER, {"warehouse_id": WH}))["planned_orders"]}
assert {v for k, v in status.items() if ":run 1" in k} == {"superseded"} and {v for k, v in status.items() if ":run 2" in k} == {"proposed"}
# superseded orders no longer publish requirements: the plan's GREEN-ETH is still 1,200, not 2,400
got = must(api("GET", "/dependent-requirements", SYSTEM, {"product_id": "GREEN-ETH", "warehouse_id": WH, "from": "2026-11-16", "to": "2026-12-14"}))
assert sum(Decimal(r["quantity"]) for r in got["requirements"]) == Decimal("1200")
def test_mfg_17_a_firmed_order_is_kept_and_counts_as_supply(nb):
world(nb)
first = must(run(), 201)
ref = next(p["planned_order_ref"] for p in first["planned_orders"] if p["product_id"] == "ROAST-ETH" and p["due_on"] == "2026-11-16")
must(api("POST", f"/planned-orders/{ref}/firm", PLANNER))
second = must(run(), 201)
kept = {p["planned_order_ref"]: p["status"] for p in must(api("GET", "/planned-orders", PLANNER, {"warehouse_id": WH}))["planned_orders"]}
assert kept[ref] == "firmed"
assert [p for p in proposals(second) if p[0] == "ROAST-ETH"] == [("ROAST-ETH", "2026-11-22", "420.0000"), ("ROAST-ETH", "2026-11-29", "168.0000"),
("ROAST-ETH", "2026-12-06", "252.0000")] # the firmed 168 is not planned twice
def test_mfg_17_a_second_starter_loses_with_a_run_in_progress(nb, monkeypatch):
world(nb)
must(run(), 201)
from brick_manufacturing.infrastructure import tables
with monkeypatch.context() as m:
m.setattr(tables.ClaimRepository, "counts", lambda self, scope, keys: {k: 0 for k in keys}) # both read the same state
refused(run(), 409, "A run is in progress")
assert len(must(api("GET", "/planned-orders", PLANNER, {"warehouse_id": WH, "status": "proposed"}))["planned_orders"]) == 5 # nothing half-written
def test_mfg_17_only_a_planner_runs_it_and_the_worker_uses_its_machine_grant(nb):
world(nb)
refused(run(OPERATOR), 403)
refused(run(as_role("cost_accountant")), 403)
refused(api("POST", "/mrp/derive", PLANNER, {"warehouse_id": WH}), 403) # a machine permission, not a person's
assert must(api("POST", "/mrp/derive", SYSTEM, {"warehouse_id": WH}), 201)["run_no"] == 1
refused(run(weeks=0), 422, "1 to 26")
def test_mfg_17_a_product_that_is_bought_or_has_no_site_row_is_not_planned(nb):
world(nb)
nb.sites = [r for r in nb.sites if r["product_id"] != "ROAST-ETH"]
got = must(run(), 201)
assert proposals(got) == [("BAG-ETH-250", "2026-11-23", "900.0000")] # ROAST-ETH has no 'make' site: its need is not planned here
def test_mfg_17_a_neighbour_that_is_down_is_answered_never_a_500(nb):
world(nb)
nb.status["product"] = 503
got = api("POST", "/mrp/runs", PLANNER, {"warehouse_id": WH})
assert got[0] in (502, 503), got
nb.status["product"] = None
assert must(run(), 201)["run_no"] == 1 # the failed run took no run_no
# ── MFG-23 ─────────────────────────────────────────────────────────────────────────────────────
def test_mfg_23_open_sales_lines_are_read_from_sales_order_page_by_page(nb):
world(nb, bag_lines=[("2026-11-24", 1)] * 501)
got = must(run(), 201)
assert [p for p in proposals(got) if p[0] == "BAG-ETH-250"] == [("BAG-ETH-250", "2026-11-23", "201.0000")] # 501 - 300 on hand
def test_mfg_23_sales_order_is_optional(nb):
world(nb)
seam.unregister_local("sales-order")
got = must(run(), 201)
assert [p for p in proposals(got) if p[0] == "BAG-ETH-250"] == [] # no demand from it, and no failure
assert [p[2] for p in proposals(got) if p[0] == "ROAST-ETH"]
# ── MFG-09 ─────────────────────────────────────────────────────────────────────────────────────
def test_mfg_09_a_planned_order_is_firmed_then_converted_and_never_proposed_again(nb):
world(nb)
first = must(run(), 201)
ref = "ROASTERY:ROAST-ETH:2026-11-16:run 1"
refused(api("POST", f"/planned-orders/{ref}/convert", PLANNER), 409, "firm it first", gate="planned_firmed")
must(api("POST", f"/planned-orders/{ref}/firm", PLANNER))
refused(api("POST", f"/planned-orders/{ref}/firm", PLANNER), 409, "already firmed", gate="state")
got = must(api("POST", f"/planned-orders/{ref}/convert", PLANNER), 201)
o = got["production_order"]
assert (o["production_order_id"], o["status"], o["planned_quantity"], o["due_on"], o["bom_version"], o["planned_order_ref"]) == (
"UK01-MO-2026-00001", "created", "168.0000", "2026-11-16", 1, ref)
assert got["planned_order"]["status"] == "converted" and got["planned_order"]["production_order_id"] == "UK01-MO-2026-00001"
refused(api("POST", f"/planned-orders/{ref}/convert", PLANNER), 409, "UK01-MO-2026-00001", gate="already_converted")
again = must(run(), 201) # the order is supply now: not proposed twice
assert ("ROAST-ETH", "2026-11-16", "168.0000") not in proposals(again)
assert ref in {p["planned_order_ref"] for p in first["planned_orders"]}
nb.available[("GREEN-ETH", WH)] = Decimal("500")
released = must(api("POST", "/production-orders/UK01-MO-2026-00001/release", PLANNER))
assert released["routing_version"] == 1 and released["status"] == "released"
def test_mfg_09_a_superseded_proposal_cannot_be_firmed_or_converted_and_access_is_the_planners(nb):
world(nb)
must(run(), 201)
must(run(), 201)
old = "ROASTERY:ROAST-ETH:2026-11-16:run 1"
refused(api("POST", f"/planned-orders/{old}/firm", PLANNER), 409, "a later run replaced it", gate="state")
refused(api("POST", f"/planned-orders/{old}/convert", PLANNER), 409, "is superseded")
refused(api("POST", "/planned-orders/NOPE/firm", PLANNER), 404)
new = "ROASTERY:ROAST-ETH:2026-11-16:run 2"
refused(api("POST", f"/planned-orders/{new}/firm", OPERATOR), 403)
refused(api("POST", f"/planned-orders/{new}/firm", ACCOUNTANT), 403)
def test_mfg_09_an_order_can_also_be_created_by_hand_with_the_same_fields(nb):
world(nb)
with on_day(D16):
got = must(api("POST", "/production-orders", PLANNER, {"company_code": "UK01", "product_id": "ROAST-ETH", "warehouse_id": WH,
"planned_quantity": "84.000", "due_on": "2026-11-16"}))
assert got["status"] == "created" and got["planned_order_ref"] is None and got["bom_version"] == 1
# ── MFG-18: the plan goes stale ────────────────────────────────────────────────────────────────
def test_mfg_18_the_last_run_shows_stale_with_the_cause(nb):
world(nb)
must(run(), 201)
assert must(api("GET", "/planned-orders", PLANNER, {"warehouse_id": WH}))["run"]["stale"] is False
must(api("POST", f"/forecasts/{WH}", PLANNER, {"forecasts": [{"product_id": "ROAST-ETH", "week_start": "2026-11-18", "quantity": "10"}]}))
got = must(api("GET", "/planned-orders", PLANNER, {"warehouse_id": WH}))["run"]
assert got["stale"] is True and "forecast changed" in got["stale_cause"]
must(run(), 201)
assert must(api("GET", "/planned-orders", PLANNER, {"warehouse_id": WH}))["run"]["stale"] is False
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", "/planned-orders", PLANNER, {"warehouse_id": WH}))["run"]
assert got["stale"] is True and "ROAST-ETH:production version 2 released" in got["stale_cause"]
# ── MFG-27: released orders publish what is left to issue ──────────────────────────────────────
def test_mfg_27_a_released_order_publishes_the_components_not_yet_issued(nb):
world(nb)
must(run(), 201)
ref = "ROASTERY:ROAST-ETH:2026-11-16:run 1"
must(api("POST", f"/planned-orders/{ref}/firm", PLANNER))
oid = must(api("POST", f"/planned-orders/{ref}/convert", PLANNER), 201)["production_order"]["production_order_id"]
nb.available[("GREEN-ETH", WH)] = Decimal("500")
must(api("POST", f"/production-orders/{oid}/release", PLANNER))
ask = {"product_id": "GREEN-ETH", "warehouse_id": WH, "from": "2026-11-16", "to": "2026-11-16"}
got = must(api("GET", "/dependent-requirements", SYSTEM, ask))["requirements"]
assert [(r["source"], r["quantity"], r["production_order_id"]) for r in got] == [("released", "200.0000", oid)] # the planned one is now the order
must(api("POST", f"/production-orders/{oid}/issue", OPERATOR, {"location_id": "ROASTERY/RAW",
"lines": [{"product_id": "GREEN-ETH", "quantity": "150.000", "lot_code": "G-1"}]}))
got = must(api("GET", "/dependent-requirements", SYSTEM, ask))["requirements"]
assert [(r["source"], r["quantity"]) for r in got] == [("released", "50.0000")]
def test_mfg_27_it_is_a_machine_route_for_inventory_and_validates_its_dates(nb):
world(nb)
ask = {"product_id": "GREEN-ETH", "warehouse_id": WH, "from": "2026-11-16", "to": "2026-12-14"}
for who in (PLANNER, OPERATOR, ACCOUNTANT, ENGINEER):
refused(api("GET", "/dependent-requirements", who, ask), 403)
refused(api("GET", "/dependent-requirements", SYSTEM, {**ask, "from": "2026-12-14", "to": "2026-11-16"}), 422, gate="bounds")
refused(api("GET", "/dependent-requirements", SYSTEM, {**ask, "from": "yesterday"}), 422)
assert must(api("GET", "/dependent-requirements", SYSTEM, ask))["requirements"] == []
_ = roast_bom