dodil/erp-brick-inventory
DODIL ERP brick: inventory
tests/test_v6.py
"""Spec v6: INV-58 (issue from a source), INV-59 (output, by-products and restocked refunds, with lot links), INV-57 (production settlement),
INV-60 (a released roll-up's standard prices), INV-61 (replenishment counts dependent requirements and skips make/mrp), the tool names."""
import fakes
import world
from conftest import ACCOUNTANT, MANAGER, OPERATOR, OWNER, SYSTEM, api, must, refused
DAY = "2026-11-20"
def beans():
world.catalogue()
fakes.add_product("BEANS", base="KGM", units={})
world.warehouse(bins=("A/01", "USE"))
world.receive("BEANS", "3.2", "KGM", price="18.0000", location="WH1/USE")
def sale(refs=("pos:R-5521:1:1", "pos:R-5521:2:1"), qty="0.018", **kw):
return api("POST", "/goods-issues/from-source", SYSTEM, {"source_kind": "pos_sale", "source_id": "R-5521", "company_code": "UK1", "location_id": "WH1/USE",
"issued_on": DAY, "lines": [{"move_ref": r, "product_id": "BEANS", "quantity": qty, "unit_code": "KGM"} for r in refs], **kw})
def test_inv58_a_till_sale_takes_stock_out_at_cost_each_move_rounded_once_with_the_remainder_on_the_last_and_a_repeat_changes_nothing():
beans()
out = must(sale())
assert [(m["move_ref"], m["quantity_base"], m["value"]) for m in out["moves"]] == [("pos:R-5521:1:1", "0.018", "0.32"), ("pos:R-5521:2:1", "0.018", "0.33")]
(j,) = [j for j in fakes.JOURNALS.values() if j["source_document"] == "gi:src:pos_sale:R-5521"]
assert {l["purpose"]: (l.get("debit"), l.get("credit")) for l in j["lines"]} == {"cost_of_sales": ("0.65", None), "stock_valuation": (None, "0.65")}
again = must(sale())
assert again["replay"] is True and [m["move_id"] for m in again["moves"]] == [m["move_id"] for m in out["moves"]]
assert must(api("GET", "/stock/BEANS", OPERATOR))["on_hand"] == "3.164" # 3.2 - 0.036: nothing was taken twice
def test_inv58_a_till_sale_is_never_refused_for_stock_but_a_production_issue_short_of_stock_is():
beans()
out = must(sale(refs=("pos:R-9:1:1",), qty="5")) # 5 KGM sold, 3.2 on the shelf: the sale happened
assert out["moves"][0]["quantity_base"] == "5"
assert must(api("GET", "/stock/BEANS", OPERATOR))["on_hand"] == "-1.8"
s, b = api("POST", "/goods-issues/from-source", SYSTEM, {"source_kind": "production_issue", "source_id": "MO-1", "company_code": "UK1", "location_id": "WH1/A/01",
"issued_on": DAY, "account_purpose": "wip",
"lines": [{"move_ref": "mfg:MO-1:issue:1:1", "product_id": "BEANS", "quantity": "1", "unit_code": "KGM"}]})
assert s == 409 and b["detail"]["gate"] == "on_hand"
refused(api("POST", "/goods-issues/from-source", OPERATOR, {"source_kind": "pos_sale", "source_id": "R-1", "company_code": "UK1", "location_id": "WH1/USE",
"issued_on": DAY, "lines": [{"move_ref": "pos:R-1:1:1", "product_id": "BEANS", "quantity": "1", "unit_code": "KGM"}]}),
403, "inv:issue:intake")
def test_inv58_a_production_issue_posts_wip_at_the_lots_cost():
beans()
out = must(api("POST", "/goods-issues/from-source", SYSTEM, {"source_kind": "production_issue", "source_id": "MO-2", "company_code": "UK1",
"location_id": "WH1/USE", "issued_on": DAY, "account_purpose": "wip",
"lines": [{"move_ref": "mfg:MO-2:issue:1:1", "product_id": "BEANS", "quantity": "2", "unit_code": "KGM"}]}))
assert out["moves"][0]["value"] == "36.00"
(j,) = [j for j in fakes.JOURNALS.values() if j["source_document"] == "gi:src:production_issue:MO-2"]
assert {l["purpose"]: (l.get("debit"), l.get("credit")) for l in j["lines"]} == {"wip": ("36.00", None), "stock_valuation": (None, "36.00")}
def test_inv59_an_output_is_received_from_production_at_its_cost_with_a_lot_link_per_input_lot_and_the_trace_walks_both_ways():
world.catalogue()
fakes.add_product("GREEN", base="KGM", units={})
fakes.add_product("ROAST", base="KGM", units={})
world.warehouse(bins=("A/01", "A/02"))
must(api("POST", "/settings", OWNER, {"key": "lots_enabled", "value": "true"}))
for p in ("GREEN", "ROAST"):
must(api("POST", "/stocked-products", MANAGER, {"product_id": p, "tracking": "lot"}))
world.receive("GREEN", "84", "KGM", price="6.0000", location="WH1/A/01", lot_code="GE-0412")
body = {"move_ref": "mfg:MO-7:output:1", "source_kind": "production_receipt", "source_id": "MO-7", "company_code": "UK1", "product_id": "ROAST",
"quantity": "82.5", "unit_code": "KGM", "location_id": "WH1/A/02", "received_on": DAY, "lot_code": "RE-1180", "unit_cost": "9.3900",
"currency": "GBP", "input_lots": [{"product_id": "GREEN", "lot_code": "GE-0412", "quantity": "84"}]}
out = must(api("POST", "/receipts", SYSTEM, body))
assert out["quantity_base"] == "82.5" and out["value"] == "774.68" # 82.5 x 9.3900 = 774.675, rounded once half up
again = must(api("POST", "/receipts", SYSTEM, body))
assert again["replay"] is True and again["move_id"] == out["move_id"]
(j,) = [j for j in fakes.JOURNALS.values() if j["source_document"] == "rcpt:src:production_receipt:MO-7"]
assert {l["purpose"]: (l.get("debit"), l.get("credit")) for l in j["lines"]} == {"stock_valuation": ("774.68", None), "wip": (None, "774.68")}
back = must(api("GET", "/lots/ROAST/RE-1180/trace", OPERATOR))
assert [(x["product_id"], x["lot_code"], x["quantity_base"]) for x in back["made_from"]] == [("GREEN", "GE-0412", "84")]
fwd = must(api("GET", "/lots/GREEN/GE-0412/trace", OPERATOR))
assert [(x["product_id"], x["lot_code"]) for x in fwd["made_into"]] == [("ROAST", "RE-1180")]
by = dict(body, move_ref="mfg:MO-7:byproduct:1:1", source_kind="by_product_receipt", product_id="GREEN", quantity="2", unit_cost="0.5000", lot_code="CH-1", input_lots=[])
assert must(api("POST", "/receipts", SYSTEM, by))["value"] == "1.00"
refused(api("POST", "/receipts", SYSTEM, dict(body, move_ref="mfg:MO-7:output:2", lot_code=None)), 422, "names its lot")
def test_inv59_a_restocked_refund_returns_at_the_cost_the_sale_left_at_and_never_more_than_was_issued():
beans()
must(sale(refs=("pos:R-5521:3:1",), qty="1.5")) # 1.5 x 18.0000 = 27.00 left
back = {"move_ref": "pos:R-5530:1:1", "source_kind": "pos_refund_restock", "source_id": "R-5530", "company_code": "UK1", "product_id": "BEANS", "quantity": "0.5",
"unit_code": "KGM", "location_id": "WH1/USE", "received_on": DAY, "reverses_move_ref": "pos:R-5521:3:1"}
out = must(api("POST", "/receipts", SYSTEM, back))
assert out["value"] == "9.00" # one third of the 27.00 it left at
(j,) = [j for j in fakes.JOURNALS.values() if j["source_document"] == "rcpt:src:pos_refund_restock:R-5530"]
assert {l["purpose"]: (l.get("debit"), l.get("credit")) for l in j["lines"]} == {"stock_valuation": ("9.00", None), "cost_of_sales": (None, "9.00")}
s, b = api("POST", "/receipts", SYSTEM, dict(back, move_ref="pos:R-5530:2:1", quantity="1.1"))
assert s == 422 and b["detail"]["gate"] == "within_issued"
s, b = api("POST", "/receipts", SYSTEM, dict(back, move_ref="pos:R-5530:3:1", reverses_move_ref="pos:NOPE:1:1"))
assert s == 422 and b["detail"]["gate"] == "reverses_known"
def to_standard():
must(api("POST", "/settings", OWNER, {"key": "valuation_method:UK1", "value": "standard", "valid_from": "2026-12-01"}))
def test_inv60_a_released_rollup_becomes_a_scheduled_standard_price_never_over_the_accountants():
world.catalogue()
world.warehouse()
to_standard()
body = {"company_code": "UK1", "product_id": "COLA-075", "valid_from": "2026-12-01", "standard_price": "9.6100", "source_ref": "mfg:CR-2026-12",
"released_by": "costing-b@shop.example", "released_sub": "costing-b"}
out = must(api("POST", "/standard-prices", SYSTEM, body))
assert out["lifecycle"] == "scheduled" and out["standard_price"] == "9.6100"
assert must(api("POST", "/standard-prices", SYSTEM, body))["replay"] is True
s, b = api("POST", "/standard-prices", SYSTEM, dict(body, standard_price="9.7000"))
assert s == 409 and b["detail"]["gate"] == "standard_price_taken" and "costing-b@shop.example" in b["detail"]["reason"]
s, b = api("POST", "/standard-prices", SYSTEM, dict(body, valid_from="2026-11-20", product_id="SCREW-M4"))
assert s == 422 and b["detail"]["gate"] == "starts_after_today"
must(api("POST", "/standard-prices", ACCOUNTANT, {"company_code": "UK1", "product_id": "SCREW-M4", "standard_price": "0.5", "valid_from": "2026-12-01"}))
s, b = api("POST", "/standard-prices", SYSTEM, dict(body, product_id="SCREW-M4"))
assert s == 409 and b["detail"]["gate"] == "standard_price_taken" # the accountant's own row is never overwritten
refused(api("POST", "/standard-prices", ACCOUNTANT, body), 403, "inv:standard_price:intake")
def test_inv60_a_company_valued_at_average_is_refused():
world.catalogue()
world.warehouse()
s, b = api("POST", "/standard-prices", SYSTEM, {"company_code": "UK1", "product_id": "COLA-075", "valid_from": "2026-12-01", "standard_price": "1.0000",
"source_ref": "mfg:CR-1", "released_by": "b@x", "released_sub": "b"})
assert s == 422 and b["detail"]["gate"] == "values_at_standard"
def test_inv57_a_production_variance_is_spread_over_what_is_on_hand_and_the_rest_goes_to_cost_of_sales():
world.catalogue()
fakes.add_product("ROAST", base="KGM", units={})
world.warehouse(bins=("A/01",))
out = {"move_ref": "mfg:MO-9:output:1", "source_kind": "production_receipt", "source_id": "MO-9", "company_code": "UK1", "product_id": "ROAST", "quantity": "100",
"unit_code": "KGM", "location_id": "WH1/A/01", "received_on": DAY, "unit_cost": "10.0000", "currency": "GBP"}
must(api("POST", "/receipts", SYSTEM, out))
must(sale_of("ROAST", "40")) # 40 of the 100 already sold
adj = {"adjustment_ref": "mfg:MO-9:settle:1", "source_kind": "production_settlement", "production_order_id": "MO-9", "company_code": "UK1", "product_id": "ROAST",
"amount": "50.00", "currency": "GBP", "on": DAY}
got = must(api("POST", "/valuation-adjustments", SYSTEM, adj))
assert (got["on_hand_amount"], got["cost_of_sales_amount"]) == ("30.00", "20.00") # 60 of 100 still on hand
assert must(api("POST", "/valuation-adjustments", SYSTEM, adj))["replay"] is True
(j,) = [j for j in fakes.JOURNALS.values() if j["source_document"] == "vadj:mfg:MO-9:settle:1"]
assert {l["purpose"]: (l.get("debit"), l.get("credit")) for l in j["lines"]} == {"stock_valuation": ("30.00", None), "cost_of_sales": ("20.00", None), "wip": (None, "50.00")}
def sale_of(product, qty):
return api("POST", "/goods-issues/from-source", SYSTEM, {"source_kind": "pos_sale", "source_id": "R-1", "company_code": "UK1", "location_id": "WH1/A/01",
"issued_on": DAY, "lines": [{"move_ref": "pos:R-1:1:1", "product_id": product, "quantity": qty, "unit_code": "KGM"}]})
def test_the_shared_family_tools_name_the_module_noun():
from brick_inventory.application import use_cases
names = {u.name for u in use_cases.INV.use_cases}
assert {"set_inventory_account_map", "get_inventory_account_map", "sign_inventory_migration", "list_inventory_migration_runs",
"reconcile_inventory_migration"} <= names
assert not {"set_account_map", "get_account_map", "sign_odoo_run", "list_odoo_history", "reconcile_odoo_stock"} & names
# ── INV-61 ───────────────────────────────────────────────────────────────────────────────────────
def manufacturing(requirements):
"""A stand-in manufacturing that answers its signed GET /dependent-requirements (MFG-27)."""
from erp_platform import seam
from fastapi import FastAPI, Query
app = FastAPI()
asked = []
@app.get("/dependent-requirements")
def dependent(product_id: str, warehouse_id: str, frm: str = Query(alias="from"), to: str = Query()):
asked.append((product_id, warehouse_id))
return {"product_id": product_id, "warehouse_id": warehouse_id, "requirements": requirements.get(product_id, [])}
seam.register_local("manufacturing", app)
return asked
def test_inv61_replenishment_counts_what_production_will_consume_and_leaves_made_products_alone():
world.catalogue()
fakes.add_product("GREEN", base="KGM", units={})
fakes.add_product("ROAST", base="KGM", units={})
world.warehouse(bins=("A/01",))
world.receive("GREEN", "900", "KGM", price="6.0000", location="WH1/A/01")
site = {"site_id": "WH1", "valid_from": "2026-01-01", "procurement_type": "buy", "planning_method": "reorder_point", "reorder_point": "500", "max_stock": "2000",
"purchase_unit_code": "KGM", "base_unit_code": "KGM", "lot_size_method": "none", "planned_delivery_days": 3}
fakes.SITES.extend([dict(site, product_id="GREEN"), dict(site, product_id="ROAST", procurement_type="make", planning_method="mrp")])
fakes.ROUNDING[("GREEN", "WH1")] = lambda q: q
asked = manufacturing({"GREEN": [{"required_on": "2026-11-22", "quantity": "1200", "source": "planned"}]})
try:
got = must(api("POST", "/replenishment/run", SYSTEM, {}))
finally:
from erp_platform import seam
seam.unregister_local("manufacturing")
(p,) = got["proposals"] # ROAST (make, mrp) is manufacturing's: not proposed
assert (p["product_id"], p["figures"]["projected_base"]) == ("GREEN", "-300") # 900 on hand - 1,200 needed by planned orders
assert asked == [("GREEN", "WH1")]
def test_inv61_without_manufacturing_the_projection_is_as_before():
world.catalogue()
fakes.add_product("GREEN", base="KGM", units={})
world.warehouse(bins=("A/01",))
world.receive("GREEN", "900", "KGM", price="6.0000", location="WH1/A/01")
fakes.SITES.append({"product_id": "GREEN", "site_id": "WH1", "valid_from": "2026-01-01", "procurement_type": "buy", "planning_method": "reorder_point",
"reorder_point": "500", "max_stock": "2000", "purchase_unit_code": "KGM", "base_unit_code": "KGM", "lot_size_method": "none"})
assert must(api("POST", "/replenishment/run", SYSTEM, {}))["proposals"] == [] # 900 is above the reorder point of 500