dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
tests/test_floor.py
"""MFG-12 (time posts activity cost), MFG-14 (scrap), reverse_entry (an operator's own wrong entry, the same day), and the two scenarios the spec
names (SC-MFG-01 roast a batch from plan to close, SC-MFG-02 a burnt batch)."""
import json
from decimal import Decimal
from conftest import (
ACCOUNTANT,
OPERATOR,
OWNER,
PLANNER,
SYSTEM,
api,
as_role,
close,
complete,
get_order,
issue,
must,
new_order,
on_day,
output,
person,
refused,
release,
roast_bom,
)
from erp_platform.seam import derived_id
from test_engineering import roast_routing, work_centres
D16 = "2026-11-16T09:00:00+00:00"
def routed_order(chaff="1.200", qty="84.000"):
"""A released ROAST-ETH order with its BOM v1 and routing v1 frozen (op 10 on ROASTER-1, op 20 waiting)."""
work_centres()
roast_bom(chaff=chaff)
roast_routing()
oid = new_order(qty)
must(release(oid))
return oid
def confirm(oid, op=10, machine="160", labour="160", final=True, who=None, day=D16):
with on_day(day):
return api("POST", f"/production-orders/{oid}/confirmations", who or OPERATOR,
{"op_no": op, "machine_minutes": machine, "labour_minutes": labour, "final": final})
def reverse(oid, ref, who=None, day=D16, reason="wrong bin"):
with on_day(day):
return api("POST", f"/production-orders/{oid}/reverse", who or OPERATOR, {"entry_ref": ref, "reason": reason})
# ── MFG-12 ─────────────────────────────────────────────────────────────────────────────────────
def test_mfg_12_the_roaster_time_costs_180_and_posts_to_gl_from_a_pending_row(nb):
oid = routed_order()
got = must(confirm(oid), 201)
assert got["confirmation_no"] == 1 and got["sent"] == {"sent": 1, "rejected": [], "pending": 0}
assert "cost" not in got["confirmations"][0] # the shop-floor view carries no cost
cost = must(api("GET", f"/production-orders/{oid}", ACCOUNTANT))["confirmations"][0]
assert (cost["cost"], cost["machine_minutes"], cost["is_final"], cost["work_centre_id"]) == ("180.00", "160.0000", True, "ROASTER-1")
jid = str(derived_id(f"{oid}:activity:10:1", "activity"))
j = nb.journals[jid]
assert (j["source_module"], j["event"], j["company_code"], j["journal_date"]) == ("manufacturing", "activity", "UK01", "2026-11-16")
assert j["lines"] == [{"purpose": "wip", "debit": "180.00"}, {"purpose": "activity_absorption", "credit": "180.00", "cost_centre": "ROASTERY"}]
second = must(confirm(oid, machine="10", labour="5", final=False), 201)
assert second["confirmation_no"] == 2
assert str(derived_id(f"{oid}:activity:10:2", "activity")) in nb.journals and len(nb.journals) == 2 # its own journal
def test_mfg_12_machine_and_labour_minutes_are_costed_on_their_own_rates(nb):
oid = routed_order()
must(confirm(oid, machine="60", labour="30"), 201) # 60/60 x 45.0000 + 30/60 x 22.5000 = 56.25
assert must(api("GET", f"/production-orders/{oid}", ACCOUNTANT))["confirmations"][0]["cost"] == "56.25"
def test_mfg_12_what_cannot_be_confirmed_is_refused_with_the_reason(nb):
oid = routed_order()
refused(confirm(oid, op=20), 409, "not a step on a work centre", gate="operation") # waiting time
refused(confirm(oid, op=99), 409, gate="operation")
refused(confirm(oid, machine="0", labour="0"), 422, "some time", gate="bounds")
refused(confirm(oid, machine="-1"), 422)
refused(confirm(oid, who=PLANNER), 403)
refused(confirm(oid, who=ACCOUNTANT), 403)
refused(confirm("NOPE"), 404)
plain = new_order(due="2026-11-16") # created, never released: the operator sees nothing
refused(confirm(plain), 404)
assert nb.journals == {}
def test_mfg_12_an_order_without_a_routing_has_nothing_to_confirm():
roast_bom()
oid = new_order()
must(release(oid))
refused(confirm(oid), 409, "no routing", gate="routing")
def test_mfg_12_a_completed_order_takes_no_more_time():
oid = routed_order()
must(confirm(oid), 201)
must(complete(oid))
refused(confirm(oid), 404) # a completed order is not on the shop floor any more: the operator sees nothing
def test_mfg_12_an_activity_cost_no_ledger_amount_can_hold_is_a_bound_not_a_500():
from datetime import date
from brick_manufacturing.domain import routing as rt
from brick_manufacturing.domain.errors import OutOfBounds
rates = [{"work_centre_id": "W", "rate_kind": "machine", "valid_from": date(2026, 1, 1), "rate_amount": Decimal("999999999999999.9999")}]
try:
rt.activity_cost(rates, "W", date(2026, 11, 16), Decimal("999999999999999"), Decimal("0"))
except OutOfBounds as e:
assert "ledger amount" in str(e)
else:
raise AssertionError("no bound")
def test_mfg_12_gl_down_keeps_the_confirmation_pending_and_the_order_open(nb):
oid = routed_order()
nb.status["gl"] = 503
got = must(confirm(oid), 201)
assert got["sent"]["pending"] == 1 and nb.journals == {}
must(api("POST", f"/production-orders/{oid}/issue", OPERATOR, {"location_id": "ROASTERY/RAW", "lines": [{"product_id": "GREEN-ETH", "quantity": "100", "lot_code": "G"}]}))
must(complete(oid))
refused(close(oid), 409, "Not every movement", gate="outbox_empty")
nb.status["gl"] = None
assert must(api("POST", "/outbox/dispatch", SYSTEM, {}))["sent"] >= 1
assert len(nb.journals) == 1
def test_mfg_12_a_journal_gl_refuses_voids_the_confirmation_it_carried(nb):
oid = routed_order()
nb.status["gl"] = 422
got = must(confirm(oid), 201)
assert got["sent"]["rejected"] and nb.journals == {}
view = must(api("GET", f"/production-orders/{oid}", ACCOUNTANT))
assert view["confirmations"][0]["status"] == "void" and view["wip_balance"] == "0.00" # counts for nothing, shown as such
refused(reverse(oid, f"mfg:{oid}:activity:10:1"), 409, "nothing to reverse", gate="not_taken")
# ── reverse_entry ──────────────────────────────────────────────────────────────────────────────
def test_reverse_a_wrong_confirmation_by_a_mirror_never_an_edit(nb):
oid = routed_order()
must(confirm(oid), 201)
got = must(reverse(oid, f"mfg:{oid}:activity:10:1"), 201)
rows = must(api("GET", f"/production-orders/{oid}", ACCOUNTANT))["confirmations"]
assert [(r["confirmation_no"], r["cost"], r["reverses_confirmation_no"]) for r in rows] == [(1, "180.00", 0), (2, "-180.00", 1)]
assert must(api("GET", f"/production-orders/{oid}", ACCOUNTANT))["wip_balance"] == "0.00" and got["reversed"] == f"mfg:{oid}:activity:10:1"
mirror = nb.journals[str(derived_id(f"{oid}:activity:10:1:reverse", "activity"))]
assert mirror["lines"] == [{"purpose": "activity_absorption", "debit": "180.00", "cost_centre": "ROASTERY"}, {"purpose": "wip", "credit": "180.00"}]
assert len(nb.journals) == 2 and str(derived_id(f"{oid}:activity:10:1", "activity")) != str(derived_id(f"{oid}:activity:10:1:reverse", "activity"))
refused(reverse(oid, f"mfg:{oid}:activity:10:1"), 409, "already reversed", gate="already_reversed")
refused(reverse(oid, f"mfg:{oid}:activity:10:1:reverse"), 409, gate="reversal")
again = must(confirm(oid, machine="150", labour="150"), 201) # the right one is the next number
assert again["confirmation_no"] == 3
def test_reverse_is_your_own_entry_the_same_business_day_while_the_order_is_open(nb):
oid = routed_order()
must(confirm(oid), 201)
other = as_role("operator", "second.operator@roastery.example")
refused(reverse(oid, f"mfg:{oid}:activity:10:1", who=other), 409, "someone else", gate="own_entry")
refused(reverse(oid, f"mfg:{oid}:activity:10:1", day="2026-11-17T09:00:00+00:00"), 409, "same business day", gate="same_day")
same_sub = person("renamed@roastery.example", ["mfg:order:reverse", "mfg:order:read"], sub="operator@roastery.example")
must(reverse(oid, f"mfg:{oid}:activity:10:1", who=same_sub), 201) # the same person under their other identity
refused(reverse(oid, f"mfg:{oid}:activity:10:1", who=PLANNER), 403)
refused(reverse(oid, "mfg:other:issue:1:1"), 404)
refused(reverse(oid, f"mfg:{oid}:activity:10:1", reason=" "), 422)
must(complete(oid))
refused(reverse(oid, f"mfg:{oid}:activity:10:1"), 404)
def test_reverse_the_business_day_is_the_one_in_business_timezone(nb):
oid = routed_order()
got = must(confirm(oid, day="2027-07-14T23:30:00+00:00"), 201) # 00:30 BST on 2027-07-15
assert must(api("GET", f"/production-orders/{oid}", ACCOUNTANT))["confirmations"][0]["business_date"] == "2027-07-15" and got["confirmation_no"] == 1
refused(reverse(oid, f"mfg:{oid}:activity:10:1", day="2027-07-15T23:10:00+00:00"), 409, "same business day", gate="same_day") # 00:10 BST on the 16th
must(reverse(oid, f"mfg:{oid}:activity:10:1", day="2027-07-15T08:00:00+00:00"), 201)
def test_reverse_an_issue_goes_back_to_stock_naming_the_original(nb):
oid = routed_order()
must(issue(oid))
over = must(issue(oid, qty="3.000"))
assert [m["status"] for m in over["movements"]] == ["sent", "sent"] and over["movements"][1]["over_issue"] is True
got = must(reverse(oid, f"mfg:{oid}:issue:1:2"), 201)
receipt = [b for k, b in nb.calls if k == "receipt"][-1]
assert (receipt["move_ref"], receipt["reverses_move_ref"], receipt["source_kind"], receipt["quantity"], receipt["lot_code"]) == (
f"mfg:{oid}:issue:1:2:reverse", f"mfg:{oid}:issue:1:2", "production_receipt", "3.0000", "G-2611-ETH-07")
assert receipt["unit_cost"] == "6.2000" and receipt["location_id"] == "ROASTERY/RAW"
assert got["planned_components"][0]["issued_base"] == "100.0000" # net of the return
assert got["movements"][-1]["kind"] == "issue_reverse" and got["movements"][-1]["status"] == "sent"
refused(reverse(oid, f"mfg:{oid}:issue:1:2"), 409, "already reversed", gate="already_reversed")
again = must(issue(oid, qty="4.000")) # 100 + 4 is within 5% again: the 3 came back
assert again["planned_components"][0]["issued_base"] == "104.0000"
def test_reverse_an_issue_inventory_refuses_is_recorded_rejected_and_can_be_tried_again(nb):
oid = routed_order()
must(issue(oid))
nb.refuse_receipts = {"gate": "reverses_known", "reason": "inventory does not know that move", "fix": "check the reference"}
refused(reverse(oid, f"mfg:{oid}:issue:1:1"), 409, "inventory does not know", gate="reverses_known")
view = get_order(oid)
assert view["movements"][-1]["status"] == "rejected" and view["planned_components"][0]["issued_base"] == "100.0000" # counts for nothing
nb.refuse_receipts = None
must(reverse(oid, f"mfg:{oid}:issue:1:1"), 201)
assert get_order(oid)["movements"][-1]["move_ref"] == f"mfg:{oid}:issue:1:1:reverse:2"
assert get_order(oid)["planned_components"][0]["issued_base"] == "0.0000"
def test_reverse_an_issue_inventory_never_took_has_nothing_to_return(nb):
oid = routed_order()
nb.status["inventory"] = 503
assert must(issue(oid))["sent"]["pending"] == 1 # inventory did not answer: the issue stays pending
nb.status["inventory"] = None
refused(reverse(oid, f"mfg:{oid}:issue:1:1"), 409, "with inventory", gate="not_taken")
def test_reverse_scrap_is_voided_and_output_has_no_route_back(nb):
oid = routed_order()
must(issue(oid))
must(output(oid))
refused(reverse(oid, f"mfg:{oid}:output:1"), 409, "no route", gate="output_not_reversible")
refused(reverse(oid, f"mfg:{oid}:byproduct:2:1"), 409, gate="output_not_reversible")
with on_day(D16):
must(api("POST", f"/production-orders/{oid}/scrap", OPERATOR, {"quantity": "1.000", "reason_code": "BURNT"}), 201)
got = must(reverse(oid, f"mfg:{oid}:scrap:1"), 201)
assert got["scraps"][0]["status"] == "void"
refused(reverse(oid, f"mfg:{oid}:scrap:1"), 409, "already reversed", gate="already_reversed")
# ── MFG-14 ─────────────────────────────────────────────────────────────────────────────────────
def scrap(oid, qty="6.000", reason="BURNT", who=None):
with on_day(D16):
return api("POST", f"/production-orders/{oid}/scrap", who or OPERATOR, {"quantity": qty, "reason_code": reason})
def test_mfg_14_a_burnt_batch_is_scrap_never_received_and_big_scrap_waits_for_a_planner(nb):
oid = routed_order(chaff=None)
must(issue(oid))
must(output(oid, qty="78.000", chaff=None))
got = must(scrap(oid), 201)
assert (got["scrap_no"], got["scrap_status"]) == (1, "pending_approval") # 6.000 x 9.3899 = 56.34 > 50.00
row = must(api("GET", f"/production-orders/{oid}", ACCOUNTANT))["scraps"][0]
assert (row["standard_value"], row["quantity"], row["reason_code"], row["status"]) == ("56.34", "6.0000", "BURNT", "pending_approval")
assert [b for k, b in nb.calls if k == "receipt"][-1]["quantity"] == "78.000" # only the good output reached stock
assert "standard_value" not in got["scraps"][0]
refused(api("POST", f"/production-orders/{oid}/scrap/1/approve", OPERATOR), 403)
must(confirm(oid, machine="150", labour="150"), 201)
must(complete(oid))
refused(close(oid), 409, "waiting for a planner", gate="scrap_approved")
approved = must(api("POST", f"/production-orders/{oid}/scrap/1/approve", PLANNER))
assert approved["scraps"][0]["status"] == "approved"
refused(api("POST", f"/production-orders/{oid}/scrap/1/approve", PLANNER), 409, gate="state")
closed = close(oid)
s = must(closed)["settlements"][0]
# SC-MFG-02: WIP debit 620.00 + 168.75 = 788.75, credit 732.41 -> 56.34, all of it abnormal scrap; the usage variance is what is left: nothing
assert (s["wip_debit"], s["wip_credit"], s["abnormal_scrap"], s["material_usage_variance"], s["activity_variance"]) == (
"788.75", "732.41", "56.34", "0.00", "0.00")
assert must(api("GET", f"/production-orders/{oid}", ACCOUNTANT))["wip_balance"] == "56.34"
journal = next(j for j in nb.journals.values() if j["event"] == "settle")
assert {x["purpose"]: x.get("debit") or x.get("credit") for x in journal["lines"]} == {"abnormal_scrap": "56.34", "wip": "56.34"}
def test_mfg_14_small_scrap_is_approved_at_once_and_still_expensed(nb):
oid = routed_order(chaff=None)
must(issue(oid))
got = must(scrap(oid, qty="1.000"), 201)
assert got["scrap_status"] == "approved" # 9.39 is under 50.00
def test_mfg_14_nobody_approves_their_own_scrap_by_email_or_sub(nb):
oid = routed_order(chaff=None)
must(issue(oid))
both = person("both@roastery.example", ["mfg:order:scrap", "mfg:order:release", "mfg:order:read"], sub="both-sub")
must(scrap(oid, who=both), 201)
refused(api("POST", f"/production-orders/{oid}/scrap/1/approve", both), 409, "reported this scrap yourself", gate="second_person")
other_email = person("alias@roastery.example", ["mfg:order:release", "mfg:order:read"], sub="both-sub")
refused(api("POST", f"/production-orders/{oid}/scrap/1/approve", other_email), 409, gate="second_person")
other_sub = person("both@roastery.example", ["mfg:order:release", "mfg:order:read"], sub="another-sub")
refused(api("POST", f"/production-orders/{oid}/scrap/1/approve", other_sub), 409, gate="second_person")
refused(api("POST", f"/production-orders/{oid}/scrap/7/approve", PLANNER), 404)
def test_mfg_14_a_reason_must_exist_and_normal_loss_is_the_yield_not_scrap(nb):
oid = routed_order(chaff=None)
must(issue(oid))
refused(scrap(oid, reason="WHIM"), 409, "not a scrap reason", gate="scrap_reason")
refused(api("POST", "/scrap-reasons/MOISTURE", PLANNER, {"description": "moisture lost", "abnormal": False}), 403)
must(api("POST", "/scrap-reasons/MOISTURE", OWNER, {"description": "moisture lost in the roast", "abnormal": False}), 201)
refused(scrap(oid, reason="moisture"), 409, "planned yield", gate="normal_loss")
assert must(api("GET", "/scrap-reasons", OPERATOR))["reasons"]["MOISTURE"] is False
refused(scrap(oid, qty="0"), 422)
refused(scrap(oid, who=PLANNER), 403)
nb.costs.pop("ROAST-ETH")
refused(scrap(oid), 409, "no cost", gate="standard_price")
# ── SC-MFG-01 ──────────────────────────────────────────────────────────────────────────────────
def test_sc_mfg_01_roast_a_batch_from_plan_to_close(nb):
oid = routed_order()
must(issue(oid)) # 100.000 KGM of lot G-2611-ETH-07
must(issue(oid, qty="3.000")) # the 3.000 over-issue ...
must(reverse(oid, f"mfg:{oid}:issue:1:2"), 201) # ... is put back the same day
got = must(output(oid)) # 82.500 KGM as lot R-20261116-01 and 1.100 KGM chaff
assert got["yield_pct"] == "82.50"
must(confirm(oid, machine="160", labour="160"), 201)
must(complete(oid))
done = must(close(oid))
s = done["settlements"][0]
assert (s["wip_debit"], s["wip_credit"], s["standard_material"]) == ("800.00", "774.67", "608.93")
assert (s["material_usage_variance"], s["activity_variance"]) == ("11.07", "14.26")
assert done["wip_balance"] == "25.33" and Decimal(s["material_usage_variance"]) + Decimal(s["activity_variance"]) == Decimal("25.33")
settle = next(j for j in nb.journals.values() if j["event"] == "settle")
assert {x["purpose"]: x.get("debit") or x.get("credit") for x in settle["lines"]} == {
"material_usage_variance": "11.07", "activity_variance": "14.26", "wip": "25.33"}
assert json.dumps(done["settlements"]) and done["status"] == "closed"