dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
tests/test_invariants.py
"""The recurring bug classes, one test each (module.yaml tests.invariants): one claim key per stock/money operation; side effects only
after the local commit, from pending rows; state machines never go backwards; self-approval by email AND sub; business date in
business_timezone; quantities and money bounded -> 422, never 500; settlement sums to the balance."""
from decimal import Decimal
import pytest
from conftest import (
ACCOUNTANT,
ACCOUNTANT2,
OPERATOR,
OWNER,
PLANNER,
ROLES,
SYSTEM,
api,
close,
complete,
get_order,
issue,
line,
must,
new_order,
on_day,
output,
person,
refused,
release,
released_order,
roast_bom,
set_day,
)
from brick_manufacturing.application import use_cases
from brick_manufacturing.infrastructure import tables
from brick_manufacturing.infrastructure.tables import Repository
SEES_ALL_OPERATOR = person("op.lead@roastery.example", ROLES["operator"] + ["mfg:order:create"]) # works the floor and sees every order
# ── one claim key per stock / money operation ───────────────────────────────────────────────────────────
def test_an_issue_an_output_a_byproduct_and_a_settlement_never_share_a_key(nb, engine):
oid = released_order()
must(issue(oid, "50.000"))
must(issue(oid, "50.000"))
must(output(oid))
must(output(oid, "1.000", lot="R-20261116-02", chaff="0.100"))
must(complete(oid))
must(close(oid))
with engine.begin() as conn:
claims = [(r.claim_scope, r.claim_key, r.n, r.ref) for r in conn.execute(tables.select(tables.mfg_claim).order_by(
tables.mfg_claim.c.claim_scope, tables.mfg_claim.c.n))]
assert claims == [("byproduct", f"{oid}:2", 1, f"mfg:{oid}:byproduct:2:1"), ("byproduct", f"{oid}:2", 2, f"mfg:{oid}:byproduct:2:2"),
("issue", f"{oid}:1", 1, f"mfg:{oid}:issue:1:1"), ("issue", f"{oid}:1", 2, f"mfg:{oid}:issue:1:2"),
("output", oid, 1, f"mfg:{oid}:output:1"), ("output", oid, 2, f"mfg:{oid}:output:2"),
("settle", oid, 1, f"mfg:{oid}:settle:1")]
refs = [r for _, _, _, r in claims]
assert len(refs) == len(set(refs))
assert sorted(b["move_ref"] for k, b in nb.calls if k == "receipt") == sorted(r for r in refs if ":output:" in r or ":byproduct:" in r)
def test_two_writers_that_read_the_same_state_claim_the_same_key_and_the_loser_is_409(monkeypatch):
oid = released_order()
must(issue(oid, "10.000"))
monkeypatch.setattr(tables.ClaimRepository, "counts", lambda self, scope, keys: {k: 0 for k in keys}) # a stale read
status, body = issue(oid, "10.000")
assert status == 409 and "Someone else wrote the same record" in body["detail"]
def test_the_order_number_is_claimed_from_the_series_and_a_racer_loses(monkeypatch):
roast_bom()
new_order()
monkeypatch.setattr(tables.FactRepository, "next_order_no", lambda self, company, year: 1) # a stale read
status, body = api("POST", "/production-orders", PLANNER, {"company_code": "UK01", "product_id": "ROAST-ETH", "warehouse_id": "ROASTERY",
"planned_quantity": "84.000", "due_on": "2026-11-16"})
assert status == 409 and "Someone else wrote" in body["detail"]
# ── side effects only after the local commit, from pending rows ────────────────────────────────────────
def test_a_rolled_back_movement_sends_nothing_and_leaves_no_row(nb, engine, monkeypatch):
oid = released_order()
def boom(self, rows, at):
raise RuntimeError("the write failed")
monkeypatch.setattr(tables.OutboxRepository, "add", boom)
with pytest.raises(RuntimeError):
issue(oid, "100.000")
assert nb.calls == []
with engine.begin() as conn:
assert Repository(conn).movements.for_order(oid) == [] and Repository(conn).claims.counts("issue", [f"{oid}:1"]) == {f"{oid}:1": 0}
def test_the_movement_and_its_pending_row_commit_together_before_anything_is_sent(nb, engine):
oid = released_order()
seen = []
original = nb.inventory.router.routes
def peek(): # at the moment inventory is asked, the rows are already committed
with engine.begin() as conn:
repo = Repository(conn)
seen.append(([m["status"] for m in repo.movements.for_order(oid)], [r["status"] for r in repo.outbox.for_order(oid)]))
from brick_manufacturing.application import dispatch
real = dispatch.send
def send(repository, order_id=None):
peek()
return real(repository, order_id)
use_cases.dispatch.send = send
try:
must(issue(oid, "100.000"))
finally:
use_cases.dispatch.send = real
assert seen == [(["pending"], ["pending"])] and original is not None
assert get_order(oid)["outbox"][0]["status"] == "sent"
def test_a_neighbour_that_is_down_leaves_a_pending_row_and_a_retry_is_the_neighbours_replay(nb):
oid = released_order()
nb.status["inventory"] = 500
must(issue(oid))
row = get_order(oid)["outbox"][0]
assert row["status"] == "pending" and row["attempts"] == 1 and "down" in row["last_error"]
nb.status["inventory"] = None
must(api("POST", "/outbox/dispatch", SYSTEM))
assert get_order(oid)["outbox"][0]["status"] == "sent"
# sent is final: the same reference sent again is inventory's replay (200), never a second move
body = [b for k, b in nb.calls if k == "issue"][-1]
from brick_manufacturing.infrastructure import seams
assert seams.INVENTORY.post("/goods-issues/from-source", body).status == 200
assert len(nb.moves) == 1
def test_a_sent_or_rejected_outbox_row_never_goes_back_to_pending(nb, engine):
oid = released_order()
must(issue(oid))
with engine.begin() as conn:
repo = Repository(conn)
row = repo.outbox.for_order(oid)[0]
repo.outbox.mark([{**row, "status": "pending", "attempts": 9, "last_error": "again"}])
assert get_order(oid)["outbox"][0]["status"] == "sent"
def test_a_value_inventory_answers_that_no_ledger_amount_can_hold_rejects_the_movement(nb):
oid = released_order()
nb.costs["GREEN-ETH"] = Decimal("500000000000000.0000")
refused(issue(oid), words="above what a ledger amount can hold")
assert get_order(oid)["movements"][0]["status"] == "rejected"
# ── state machines never go backwards ──────────────────────────────────────────────────────────────────
def test_an_order_never_goes_back_and_a_closed_one_is_terminal(nb):
oid = released_order()
refused(release(oid), words="only a created order", gate="state") # released -> created
refused(api("POST", f"/production-orders/{oid}/complete", ACCOUNTANT), words="only an order in progress", gate="state")
must(issue(oid))
must(output(oid))
must(complete(oid))
refused(complete(oid), words="only an order in progress", gate="state")
assert issue(oid, "1.000")[0] == 404 # an operator no longer sees it
refused(issue(oid, "1.000", who=SEES_ALL_OPERATOR), words="nothing more is issued", gate="state") # completed -> in_progress
refused(output(oid, who=SEES_ALL_OPERATOR), words="nothing more is issued or reported", gate="state")
refused(api("POST", f"/production-orders/{oid}/cancel", PLANNER, {"reason": "oops"}), words="only a created or released", gate="state")
must(close(oid))
for path, who, body in (("issue", OPERATOR, {"location_id": "ROASTERY/RAW", "lines": [line("GREEN-ETH", "1")]}), ("complete", ACCOUNTANT, {}),
("close", ACCOUNTANT, {}), ("release", PLANNER, {})):
status, _ = api("POST", f"/production-orders/{oid}/{path}", who, body)
assert status in (404, 409), path
assert get_order(oid)["status"] == "closed"
def test_a_cancelled_order_is_final_and_a_moved_order_cannot_be_cancelled(nb):
roast_bom()
oid = new_order()
must(api("POST", f"/production-orders/{oid}/cancel", PLANNER, {"reason": "demand dropped"}))
refused(release(oid), words="only a created order", gate="state")
refused(api("POST", f"/production-orders/{oid}/cancel", PLANNER, {"reason": "again"}), words="only a created or released")
moved = new_order()
must(release(moved))
must(issue(moved))
refused(api("POST", f"/production-orders/{moved}/cancel", PLANNER, {"reason": "late"}), words="only a created or released")
spare = new_order()
refused(api("POST", f"/production-orders/{spare}/cancel", PLANNER, {"reason": " "}), words="reason")
def test_a_released_bom_version_is_never_a_draft_again_and_an_obsolete_one_is_never_released(nb):
roast_bom()
refused(api("POST", "/boms/ROAST-ETH:production/versions/1/release", ENGINEER_), words="only a draft")
must(api("POST", "/boms/ROAST-ETH:production/versions/1/obsolete", ENGINEER_, {"valid_to": "2026-12-31"}))
refused(api("POST", "/boms/ROAST-ETH:production/versions/1/release", ENGINEER_), words="only a draft")
refused(api("POST", "/boms/ROAST-ETH:production/versions/1/obsolete", ENGINEER_, {"valid_to": "2026-12-31"}), words="only a released version")
with on_day("2027-01-05T09:00:00+00:00"):
assert must(api("GET", "/boms/ROAST-ETH:production", ENGINEER_))["valid_today"] is None
def test_obsolete_is_refused_while_an_order_not_yet_released_falls_due_after_the_last_day(nb):
roast_bom()
oid = new_order(due="2026-12-20")
refused(api("POST", "/boms/ROAST-ETH:production/versions/1/obsolete", ENGINEER_, {"valid_to": "2026-12-01"}), words=oid, gate="orders_named")
from conftest import ENGINEER as ENGINEER_ # noqa: E402
# ── self-approval: email AND sub ───────────────────────────────────────────────────────────────────────
def big_variance_order(nb):
"""840 KGM planned (1000 in, 840 out): 1000 issued, 700 out: a variance far above 250.00."""
roast_bom()
nb.onhand[("GREEN-ETH", "ROASTERY/RAW")] = Decimal("5000.000")
nb.available[("GREEN-ETH", "ROASTERY")] = Decimal("5000.000")
oid = new_order("840.000")
must(release(oid))
must(issue(oid, "1000.000"))
must(output(oid, "700.000", chaff="10.000"))
must(complete(oid))
return oid
def test_a_variance_above_the_threshold_needs_a_second_cost_accountant_by_email_and_by_sub(nb):
oid = big_variance_order(nb)
refused(close(oid), words="above 250.00", gate="variance_approved")
must(api("POST", f"/production-orders/{oid}/approve-variance", ACCOUNTANT))
refused(close(oid), words="approved this variance yourself", gate="second_person") # the closer approved it
same_sub = person("renamed@roastery.example", ROLES["cost_accountant"], sub=ACCOUNTANT["X-Dodil-User"] and "cost_accountant@roastery.example")
refused(close(oid, who=same_sub), words="approved this variance yourself", gate="second_person") # another email, the same sub
same_email = person("cost_accountant@roastery.example", ROLES["cost_accountant"], sub="another-sub")
refused(close(oid, who=same_email), words="approved this variance yourself", gate="second_person") # the same email, another sub
must(api("POST", f"/production-orders/{oid}/approve-variance", ACCOUNTANT2))
s = must(close(oid))["settlements"][0]
assert s["approved_by"] == "second.accountant@roastery.example" and s["settled_by"] == "cost_accountant@roastery.example"
def test_an_approval_is_not_needed_within_the_threshold_and_not_given_before_completion(nb):
oid = released_order()
must(issue(oid))
refused(api("POST", f"/production-orders/{oid}/approve-variance", ACCOUNTANT), words="only a completed order")
must(output(oid, "84.000", chaff="1.200"))
must(complete(oid))
refused(api("POST", f"/production-orders/{oid}/approve-variance", ACCOUNTANT), words="no approval is needed")
must(close(oid))
def test_the_threshold_is_the_owners_to_change(nb):
oid = big_variance_order(nb)
must(api("POST", "/settings/variance_approval_threshold", OWNER, {"value": "100000"})) # the first row of a key may start today
must(close(oid))
# ── the business date is in business_timezone ───────────────────────────────────────────────────────────
def test_00_30_bst_on_the_15th_is_the_15th_even_though_it_is_23_30_utc_on_the_14th(nb):
set_day("2026-07-14T09:00:00+00:00")
roast_bom(valid_from="2026-07-15")
oid = new_order(due="2026-07-15")
at = "2026-07-14T23:30:00+00:00" # 00:30 BST on the 15th
with on_day(at):
must(api("POST", f"/production-orders/{oid}/release", PLANNER)) # the BOM is valid on the 15th in London
must(api("POST", f"/production-orders/{oid}/issue", OPERATOR, {"location_id": "ROASTERY/RAW", "lines": [{"product_id": "GREEN-ETH", "quantity": "10.000"}]}))
assert get_order(oid)["released_on"] == "2026-07-15" and get_order(oid)["movements"][0]["business_date"] == "2026-07-15"
assert [b for k, b in nb.calls if k == "issue"][0]["issued_on"] == "2026-07-15"
def test_a_change_of_zone_counts_today_in_the_zone_before_it(nb):
set_day("2026-11-10T23:30:00+00:00") # 23:30 in London, 00:30 on the 11th in Paris
assert must(api("GET", "/settings", OWNER))["today"] == "2026-11-10"
must(api("POST", "/settings/business_timezone", OWNER, {"value": "Europe/Paris"})) # first row: in force today
assert must(api("GET", "/settings", OWNER))["today"] == "2026-11-11"
refused(api("POST", "/settings/business_timezone", OWNER, {"value": "Mars/Olympus", "valid_from": "2026-12-01"}), 422, "not a valid", "bounds")
# ── money and quantities are bounded: 422, never a 500 ─────────────────────────────────────────────────
BAD = ["0", "-1", "1e20", "1000000000000000", "0.00001", "abc", "", "NaN", "Infinity", "9" * 40, 1.5, True, None]
@pytest.mark.parametrize("bad", BAD)
def test_every_quantity_door_answers_422_for_a_bad_number_never_500(bad):
oid = released_order()
for status, body in (
api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER_, {"version": 2, "valid_from": "2026-12-01", "output_quantity": bad,
"output_unit_code": "KGM", "lines": [line("GREEN-ETH", "1")]}),
api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER_, {"version": 2, "valid_from": "2026-12-01", "output_quantity": "1",
"output_unit_code": "KGM", "lines": [line("GREEN-ETH", bad)]}),
api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER_, {"version": 2, "valid_from": "2026-12-01", "output_quantity": "1",
"output_unit_code": "KGM",
"lines": [line("GREEN-ETH", "1"), line("CHAFF", "1", kind="by_product",
nrv_unit_cost=bad if bad not in ("0", None) else "x")]}),
api("POST", "/production-orders", PLANNER, {"company_code": "UK01", "product_id": "ROAST-ETH", "warehouse_id": "ROASTERY",
"planned_quantity": bad, "due_on": "2026-11-16"}),
issue(oid, bad),
output(oid, bad),
api("GET", "/boms/ROAST-ETH/explode", SYSTEM, {"quantity": bad if bad != 1.5 else "x", "usage": "production", "on": "2026-11-16"})):
assert status == 422, (bad, status, body)
def test_a_date_that_is_not_a_date_is_422_and_a_missing_permission_is_still_403_first():
assert api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER_, {"version": 1, "valid_from": "next week", "output_quantity": "1",
"output_unit_code": "KGM", "lines": [line("GREEN-ETH", "1")],
"company_code": "UK01"})[0] == 422
status, body = api("POST", "/production-orders", OPERATOR, {"company_code": "UK01", "product_id": "ROAST-ETH", "warehouse_id": "R",
"planned_quantity": "1", "due_on": "next week"})
assert status == 403 and "mfg:order:create" in str(body)
def test_a_quantity_that_does_not_convert_exactly_is_422_not_500(nb):
oid = released_order()
nb.products["GREEN-ETH"][1]["PCT"] = nb.products["THIRD"][1]["PCT"]
status, body = issue(oid, "1.000", unit_code="XYZ")
assert status == 422 and body["gate"] == "bounds" and "neither KGM" in body["detail"]
status, body = api("POST", f"/production-orders/{oid}/output", OPERATOR, {"location_id": "F", "quantity": "1", "unit_code": "GRM"})
assert status == 422 and "report in that unit" in body["detail"]
# ── a neighbour that cannot answer is its own status, never a 500 ──────────────────────────────────────
def test_product_or_inventory_down_is_502_or_409_with_the_reason_never_500(nb):
roast_bom()
oid = new_order()
nb.status["inventory"] = 503
status, body = release(oid)
assert status == 502 and "inventory failed" in body["detail"]
nb.status["inventory"] = None
nb.status["product"] = 500
status, body = api("POST", "/boms/ROAST-ETH:production/versions", ENGINEER_, {"version": 2, "valid_from": "2026-12-01", "output_quantity": "1",
"output_unit_code": "KGM", "lines": [line("GREEN-ETH", "1")]})
assert status == 502 and "product failed" in body["detail"]
def test_an_unknown_product_is_not_a_500(nb):
status, body = api("POST", "/boms/NOPE:production/versions", ENGINEER_, {"version": 1, "valid_from": "2026-12-01", "output_quantity": "1",
"output_unit_code": "KGM", "company_code": "UK01",
"lines": [line("GREEN-ETH", "1")]})
assert status == 404 and "no /products/NOPE" in body["detail"]
# ── settlement: the split sums to the balance, exactly, whatever the figures ───────────────────────────
@pytest.mark.parametrize("issued,out", [("100.000", "84.000"), ("100.000", "82.500"), ("99.999", "0.001"), ("103.000", "79.123")])
def test_the_variance_split_always_sums_to_the_wip_balance(nb, issued, out):
oid = released_order()
must(issue(oid, issued, lot="G"))
must(output(oid, out, chaff=None))
must(complete(oid))
nb.status["gl"] = None
s = _close_any(oid)["settlements"][0]
assert Decimal(s["material_usage_variance"]) + Decimal(s["activity_variance"]) + Decimal(s["abnormal_scrap"]) == \
Decimal(s["wip_debit"]) - Decimal(s["wip_credit"])
def _close_any(oid):
"""Close, with a second accountant's approval when the variance needs one."""
status, body = close(oid)
if status == 409 and body.get("gate") == "variance_approved":
must(api("POST", f"/production-orders/{oid}/approve-variance", ACCOUNTANT2))
status, body = close(oid)
assert status == 200, body
return body