dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
tests/test_budget.py
"""Statement budgets: every statement is a round trip (100-300 ms on the live pg wire), so a request's statements must not grow with the
number of components or of unrelated rows (erp_platform.testing.count_statements, one request at a time)."""
from decimal import Decimal
import pytest
from conftest import (
ACCOUNTANT,
D16,
ENGINEER,
OPERATOR,
PLANNER,
SYSTEM,
api,
close,
complete,
get_order,
line,
must,
on_day,
release,
)
from erp_platform.testing import count_statements
@pytest.fixture
def counted(engine):
return lambda: count_statements(engine)
def _show(label, c):
print(f"BUDGET {label}: statements={c.statements} round_trips={c.round_trips}")
def bom_with(nb, n, product="ROAST-ETH", version=1, release_it=True, start="2026-11-16"):
"""BOM `product` with n components C01..Cnn (a bag of ingredients), 1 KGM each, 1 KGM out."""
comps = []
for k in range(1, n + 1):
pid = f"C{k:02d}"
nb.products[pid] = ("KGM", {"KGM": __import__("fractions").Fraction(1)})
nb.costs[pid] = Decimal("1.0000")
nb.available[(pid, "ROASTERY")] = Decimal("1000")
nb.onhand[(pid, "ROASTERY/RAW")] = Decimal("1000")
comps.append(line(pid, "1.000"))
must(api("POST", f"/boms/{product}:production/versions", ENGINEER,
{"version": version, "valid_from": start, "output_quantity": "1.000", "output_unit_code": "KGM", "lines": comps, "company_code": "UK01"}))
if release_it:
must(api("POST", f"/boms/{product}:production/versions/{version}/release", ENGINEER))
return [c["product_id"] for c in comps]
def test_writing_and_releasing_a_bom_does_not_grow_with_its_components(nb, counted):
with counted() as few:
bom_with(nb, 2, "ROAST-ETH")
with counted() as many:
bom_with(nb, 20, "ROAST-HOUSE")
_show("write+release BOM (2 components)", few)
_show("write+release BOM (20 components)", many)
assert many.statements <= few.statements and few.statements <= 24, (few.statements, many.statements)
def _order_of(nb, n, product):
comps = bom_with(nb, n, product)
oid = must(api("POST", "/production-orders", PLANNER, {"company_code": "UK01", "product_id": product, "warehouse_id": "ROASTERY",
"planned_quantity": "1.000", "due_on": "2026-11-16"}))["production_order_id"]
with on_day(D16):
must(api("POST", f"/production-orders/{oid}/release", PLANNER))
return oid, comps
def test_issuing_all_components_does_not_grow_with_their_number(nb, counted):
nb.costs["ROAST-ETH"] = Decimal("9.3899")
few_id, few_c = _order_of(nb, 2, "ROAST-ETH")
many_id, many_c = _order_of(nb, 20, "ROAST-HOUSE")
nb.costs["ROAST-HOUSE"] = Decimal("1.0000")
with on_day(D16):
with counted() as few:
must(api("POST", f"/production-orders/{few_id}/issue", OPERATOR,
{"location_id": "ROASTERY/RAW", "lines": [{"product_id": p, "quantity": "1.000"} for p in few_c]}))
with counted() as many:
must(api("POST", f"/production-orders/{many_id}/issue", OPERATOR,
{"location_id": "ROASTERY/RAW", "lines": [{"product_id": p, "quantity": "1.000"} for p in many_c]}))
_show("issue (2 components)", few)
_show("issue (20 components)", many)
assert many.statements <= few.statements and few.statements <= 24, (few.statements, many.statements)
with counted() as r_few:
must(api("GET", f"/production-orders/{few_id}", ACCOUNTANT))
with counted() as r_many:
must(api("GET", f"/production-orders/{many_id}", ACCOUNTANT))
_show("get order (2 components)", r_few)
_show("get order (20 components)", r_many)
assert r_many.statements <= r_few.statements and r_few.statements <= 6
def test_release_does_not_grow_with_components_and_close_does_not_grow_with_unrelated_orders(nb, counted):
nb.costs["ROAST-ETH"] = Decimal("9.3899")
first, comps = _order_of(nb, 2, "ROAST-ETH")
with on_day(D16):
must(api("POST", f"/production-orders/{first}/issue", OPERATOR, {"location_id": "ROASTERY/RAW", "lines": [{"product_id": p, "quantity": "1.000"} for p in comps]}))
must(api("POST", f"/production-orders/{first}/output", OPERATOR, {"location_id": "ROASTERY/FG", "quantity": "1.000", "lot_code": "R-1"}))
must(api("POST", f"/production-orders/{first}/complete", ACCOUNTANT))
with counted() as few:
must(close(first))
# ten unrelated orders with movements: closing another order costs the same statements
nb.costs["ROAST-HOUSE"] = Decimal("1.0000")
other, ocomps = _order_of(nb, 20, "ROAST-HOUSE")
for _ in range(10):
extra = must(api("POST", "/production-orders", PLANNER, {"company_code": "UK01", "product_id": "ROAST-HOUSE", "warehouse_id": "ROASTERY",
"planned_quantity": "1.000", "due_on": "2026-11-16"}))["production_order_id"]
with on_day(D16):
must(api("POST", f"/production-orders/{extra}/release", PLANNER))
must(api("POST", f"/production-orders/{extra}/issue", OPERATOR, {"location_id": "ROASTERY/RAW", "lines": [{"product_id": ocomps[0], "quantity": "1.000"}]}))
with on_day(D16):
must(api("POST", f"/production-orders/{other}/issue", OPERATOR, {"location_id": "ROASTERY/RAW", "lines": [{"product_id": p, "quantity": "1.000"} for p in ocomps]}))
must(api("POST", f"/production-orders/{other}/output", OPERATOR, {"location_id": "ROASTERY/FG", "quantity": "1.000", "lot_code": "B-1"}))
must(api("POST", f"/production-orders/{other}/complete", ACCOUNTANT))
with counted() as many:
status, body = close(other)
assert status in (200, 409), body
if status == 409: # a variance needing an approval: approve, then count the close
must(api("POST", f"/production-orders/{other}/approve-variance", __import__("conftest").ACCOUNTANT2))
with counted() as many:
must(close(other))
_show("close (2 orders)", few)
_show("close (12 orders, 20 components)", many)
assert many.statements <= few.statements + 2 and few.statements <= 30, (few.statements, many.statements)
assert get_order(other)["status"] == "closed" and complete and release
def test_explode_is_a_flat_read(nb, counted):
bom_with(nb, 2, "ROAST-ETH")
bom_with(nb, 20, "ROAST-HOUSE")
q = {"quantity": "3", "usage": "production", "on": "2026-11-16"}
with counted() as few:
must(api("GET", "/boms/ROAST-ETH/explode", SYSTEM, q))
with counted() as many:
must(api("GET", "/boms/ROAST-HOUSE/explode", SYSTEM, q))
_show("explode (2 components)", few)
_show("explode (20 components)", many)
assert many.statements <= few.statements and few.statements <= 4