dodil/erp-brick-inventory
DODIL ERP brick: inventory
tests/fakes.py
"""Neighbours as their SPEC routes answer them (shape and status codes): product, purchasing, sales-order, the ledger.
A fake that disagrees with the real callee is the bug that hid 31 broken seams in wave 1, so each one is a small FastAPI app
registered as the seam's neighbour (erp_platform.seam.register_local) and answers exactly what the spec's route answers.
The fake purchasing calls THIS brick's POST /receipts for every line it records, as the real one does."""
from decimal import Decimal
from fractions import Fraction
from erp_platform import seam
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
# ── product ──────────────────────────────────────────────────────────────────────────────────
PRODUCTS: dict = {}
def add_product(product_id, base="LTR", units=None, kind="goods", warehouse=None, issue=None, valuation_class="3000",
lifecycle="active", decimals=None, sales=None):
"""units: {code: (numerator, denominator)} = base units per one of it; measures per unit may be given as a 3rd item."""
rows = []
for code, spec in (units or {}).items():
n, d, *rest = spec
rows.append({"unit_code": code, "numerator": n, "denominator": d,
"quantity_decimals": (decimals or {}).get(code, 0), "gtins": [], "measures": rest[0] if rest else None})
PRODUCTS[product_id] = {"product_id": product_id, "name": product_id, "kind": kind, "valuation_class": valuation_class,
"base_unit_code": base, "lifecycle": lifecycle, "units": rows,
"roles": {"sales": sales, "purchase": None, "issue": issue, "production": None, "warehouse": warehouse,
"delivery": None}}
return PRODUCTS[product_id]
def _factor(p, code):
if code == p["base_unit_code"]:
return Fraction(1)
for u in p["units"]:
if u["unit_code"] == code:
return Fraction(u["numerator"], u["denominator"])
return None
product_app = FastAPI()
USED: list = []
@product_app.get("/products/{pid}/convert")
def convert(pid: str, quantity: str, from_unit: str, to_unit: str):
p = PRODUCTS.get(pid)
if p is None:
return JSONResponse({"detail": f"no product {pid}"}, status_code=404)
a, b = _factor(p, from_unit), _factor(p, to_unit)
for f, u in ((a, from_unit), (b, to_unit)):
if f is None:
return JSONResponse({"detail": f"{pid} is not counted in {u}"}, status_code=422)
exact = Fraction(Decimal(quantity)) * a / b
return {"product_id": pid, "from_unit": from_unit, "to_unit": to_unit, "quantity": quantity,
"exact": f"{exact.numerator}/{exact.denominator}", "converted": str(Decimal(exact.numerator) / Decimal(exact.denominator)),
"in_base_unit": str(Decimal(exact.numerator) / Decimal(exact.denominator)), "base_unit_code": p["base_unit_code"]}
@product_app.post("/products/{pid}/units/{unit}/used")
def used(pid: str, unit: str, body: dict):
USED.append((pid, unit, body.get("document_ref")))
return {"product_id": pid, "unit_code": unit, "used": True, "fixed_by": "its own row"}
@product_app.get("/products/{pid}")
def get_product(pid: str, on: str = ""):
p = PRODUCTS.get(pid)
return p if p else JSONResponse({"detail": f"no product {pid}"}, status_code=404)
SITES: list = []
@product_app.get("/product-sites")
def sites(site_id: str, planning: str = "active", on: str = ""):
return [s for s in SITES if s["site_id"] == site_id]
ROUNDING: dict = {}
MIN_ORDER: dict = {}
@product_app.get("/products/{pid}/round-proposal")
def round_proposal(pid: str, site_id: str, quantity: str, unit_code: str, on: str = ""):
rounded = ROUNDING.get((pid, site_id), lambda q: q)(Decimal(quantity))
return {"product_id": pid, "proposed": quantity, "rounded": str(rounded), "unit_code": unit_code, "steps_applied": [],
"purchase_rule": ({"met": True, "reason": ""} if Decimal(quantity) >= MIN_ORDER.get(pid, 0)
else {"met": False, "reason": f"the minimum order is {MIN_ORDER[pid]}"})}
# ── purchasing ───────────────────────────────────────────────────────────────────────────────
POS: dict = {}
RECEIPTS: dict = {}
EXPECTED_RECEIPTS: list = []
REQUESTS: dict = {}
REQUEST_REFUSAL: list = []
WITHDRAWALS: dict = {}
INV_CALLS: list = []
purchasing_app = FastAPI()
INV = seam.Seam("inventory", "INVENTORY_BASE_URL", roles=["service"], permissions=["inv:receipt:intake"])
def add_po(po, lines, bp_id=9100):
"""lines: {line_no: dict(product_id, unit_code, ordered, price, [pricing_unit_code, pricing_numerator, pricing_denominator, pricing_quantity])}"""
POS[po] = {"bp_id": bp_id, "lines": {n: {"received": Decimal(0), **l} for n, l in lines.items()}}
@purchasing_app.get("/purchase-orders/{po}/expected")
def expected(po: str):
o = POS.get(po)
if o is None:
return JSONResponse({"detail": f"no purchase order {po}"}, status_code=404)
out = []
for n, l in sorted(o["lines"].items()):
p = PRODUCTS[l["product_id"]]
outstanding = Decimal(l["ordered"]) - l["received"]
out.append({"line_no": n, "product_id": l["product_id"], "unit_code": l["unit_code"], "ordered": str(l["ordered"]),
"received": str(l["received"]), "outstanding": str(outstanding),
"quantity_base_outstanding": str(Decimal(outstanding) * Decimal(_factor(p, l["unit_code"]).numerator) / Decimal(_factor(p, l["unit_code"]).denominator)),
"expected_on": "2026-11-20"})
return {"purchase_order_id": po, "order_no": po, "bp_id": o["bp_id"], "lines": out}
@purchasing_app.post("/purchase-orders/{po}/receipts")
async def receive(po: str, request: Request):
body = await request.json()
key = (po, body["delivery_note"])
if key in RECEIPTS:
return RECEIPTS[key]
o = POS[po]
receipt_id = f"R-{len(RECEIPTS) + 1}"
lines, readiness = [], []
for ln in body["lines"]:
l = o["lines"][ln["line_no"]]
move_ref = f"pur:{receipt_id}:{ln['line_no']}"
post = {"move_ref": move_ref, "source_kind": l.get("source_kind", "purchase_receipt"), "source_id": receipt_id,
"purchase_order_id": po, "bp_id": str(o["bp_id"]), "product_id": l["product_id"], "quantity": ln["quantity"],
"unit_code": ln["unit_code"], "location_id": body["location_id"], "received_on": body["received_on"],
"unit_cost": str(l["price"]), "currency": "GBP"}
for k in ("pricing_unit_code", "pricing_numerator", "pricing_denominator", "pricing_quantity", "stock_owner_bp_id",
"consume_provided", "service_fee", "delivery_costs"):
if k in l:
post[k] = l[k]
if l.get("source_kind") == "consignment_receipt":
post.pop("unit_cost")
r = INV.post("/receipts", post, actor="purchasing")
INV_CALLS.append((post, r.status, r.reason))
if not r.ok:
return JSONResponse({"detail": r.reason}, status_code=r.status or 502)
l["received"] += Decimal(ln["quantity"])
lines.append({"line_no": ln["line_no"], "quantity": ln["quantity"], "quantity_base": r.data["quantity_base"],
"move_ref": move_ref, "stock_move_id": r.data["move_id"]})
out = {"receipt_id": receipt_id, "lines": lines, "readiness": readiness}
RECEIPTS[key] = out
return out
@purchasing_app.get("/expected-receipts")
def expected_receipts(product_id: str, warehouse_id: str):
return [e for e in EXPECTED_RECEIPTS if e["product_id"] == product_id and e["warehouse_id"] == warehouse_id]
@purchasing_app.post("/purchase-requests/from-replenishment")
async def from_replenishment(request: Request, response: Response):
body = await request.json()
ref = body["request_ref"]
if REQUEST_REFUSAL:
return JSONResponse({"detail": {"gate": "purchasable", "reason": "the product has no supplier", "fix": "add a supplier"}}, status_code=422)
if ref in REQUESTS:
return REQUESTS[ref]
response.status_code = 201
REQUESTS[ref] = {"request_ref": ref, "request_id": f"PR-{len(REQUESTS) + 1}", "status": "awaiting_approval", "body": body}
return {k: v for k, v in REQUESTS[ref].items() if k != "body"}
@purchasing_app.post("/consignment-withdrawals")
async def withdraw(request: Request, response: Response):
body = await request.json()
ref = body["move_ref"]
if ref in WITHDRAWALS:
return {**WITHDRAWALS[ref], "replay": True}
price = WITHDRAWALS.get("_price", Decimal("2.4500"))
if price is None:
return JSONResponse({"detail": {"gate": "consignment_price", "reason": "no consignment price", "fix": "agree one"}}, status_code=409)
qty = Decimal(body["quantity"])
amount = (qty * price / Decimal(100)).quantize(Decimal("0.01"))
response.status_code = 201
WITHDRAWALS[ref] = {"withdrawal_id": f"W-{len(WITHDRAWALS)}", "move_ref": ref, "unit_price": str(price),
"pricing_unit_code": "C62", "pricing_numerator": 1, "pricing_denominator": 1, "pricing_quantity": "100",
"amount": str(amount), "currency": "GBP"}
return WITHDRAWALS[ref]
# ── sales order ──────────────────────────────────────────────────────────────────────────────
ORDERS: dict = {}
DELIVERY_REFS: dict = {}
RETURN_INSPECTIONS: dict = {}
so_app = FastAPI()
def add_order(oid, lines, bp_id=7311, lifecycle="confirmed", **fulfilment):
"""lines: {line_no: dict(product_id, unit_code, quantity, unit_numerator, unit_denominator)}"""
ORDERS[oid] = {"sales_order_id": oid, "lifecycle": lifecycle, "revision": 1, "bp_id": bp_id,
"lines": {n: {"delivered": Decimal(0), **l} for n, l in lines.items()}, "fulfilment": fulfilment}
def _order_view(o):
return {"sales_order_id": o["sales_order_id"], "lifecycle": o["lifecycle"], "revision": o["revision"], "bp_id": str(o["bp_id"]),
"lines": [{"line_no": n, "product_id": l["product_id"], "unit_code": l["unit_code"], "quantity": str(l["quantity"]),
"unit_numerator": l.get("unit_numerator", 1), "unit_denominator": l.get("unit_denominator", 1),
"delivered": str(l["delivered"])} for n, l in sorted(o["lines"].items())]}
@so_app.get("/sales-orders/{oid}")
def get_order(oid: str):
o = ORDERS.get(oid)
return _order_view(o) if o else JSONResponse({"detail": f"no sales order {oid}"}, status_code=404)
@so_app.get("/sales-orders/{oid}/fulfilment")
def get_fulfilment(oid: str):
o = ORDERS.get(oid)
if o is None:
return JSONResponse({"detail": f"no sales order {oid}"}, status_code=404)
return {"sales_order_id": oid, "order_type_code": "OR", "priority": "normal", "ship_to_bp_id": str(o["bp_id"]),
"partial_delivery": "allowed", "max_partials": 9, "under_tolerance_pct": "0", "over_tolerance_pct": "0", "combine_allowed": True,
"special_stock": "none", "purpose": "sale", "delivery_blocked": {"met": True, "reasons": []},
"min_remaining_shelf_life_days": None, **o["fulfilment"]}
@so_app.post("/sales-orders/{oid}/deliveries")
async def record_delivery(oid: str, request: Request):
body = await request.json()
o = ORDERS.get(oid)
if o is None:
return JSONResponse({"detail": f"no sales order {oid}"}, status_code=404)
ref = body["delivery_ref"]
line = o["lines"].get(body["line_no"])
if line is None:
return JSONResponse({"detail": {"gate": "within_ordered", "reason": "no such line", "fix": ""}}, status_code=409)
if (oid, ref, body["line_no"]) in DELIVERY_REFS:
return _order_view(o)
qty = Decimal(body["quantity"])
if o["lifecycle"] != "confirmed":
return JSONResponse({"detail": {"gate": "within_ordered", "reason": "the order is no longer confirmed", "fix": ""}}, status_code=409)
if line["delivered"] + qty > Decimal(line["quantity"]) or line["delivered"] + qty < 0:
return JSONResponse({"detail": {"gate": "within_ordered", "reason": f"{line['delivered'] + qty} is more than the {line['quantity']} ordered",
"fix": "deliver no more than ordered"}}, status_code=409)
line["delivered"] += qty
DELIVERY_REFS[(oid, ref, body["line_no"])] = body
return _order_view(o)
@so_app.post("/sales-orders/{oid}/return-inspections")
async def return_inspection(oid: str, request: Request):
body = await request.json()
key = (oid, body["inspection_ref"])
if key not in RETURN_INSPECTIONS:
RETURN_INSPECTIONS[key] = body
return {"line_no": body["line_no"], "received": body["received_quantity"], "accepted": body["accepted_quantity"],
"rejected": body["rejected_quantity"], "credited": "0", "readiness": []}
# ── the ledger ───────────────────────────────────────────────────────────────────────────────
JOURNALS: dict = {}
GL_REFUSE: dict = {}
COST_CENTRES: dict = {"MAINT": {"cost_centre": "MAINT", "name": "Maintenance", "status": "open", "company_code": "UK1"}}
CLOSED_PERIODS: set = set()
gl_app = FastAPI()
@gl_app.post("/journal-entry/journals")
async def post_journal(request: Request, response: Response):
body = await request.json()
if not body.get("currency"): # the real route requires it (gl GL-30 request_schema: currency required)
return JSONResponse({"detail": [{"type": "missing", "loc": ["body", "currency"], "msg": "Field required"}]}, status_code=422)
if body.get("journal_id") != str(seam.derived_id(body.get("source_document"), body.get("event"))): # as the real route checks
return JSONResponse({"detail": "[journal_id] journal_id is not derived_id(source_document, event)"}, status_code=422)
for n, ln in enumerate(body["lines"], 1): # gl's bounds gate: an account or a purpose, exactly one
if ("account_id" in ln) == ("purpose" in ln):
return JSONResponse({"detail": f"[bounds] line {n} names an account or a purpose, exactly one"}, status_code=422)
if GL_REFUSE:
return JSONResponse({"detail": {"gate": GL_REFUSE.get("gate", "period_open"), "reason": GL_REFUSE.get("reason", "the period is closed"),
"fix": "reopen the period"}}, status_code=GL_REFUSE.get("status", 409))
d = sum(Decimal(l["debit"]) for l in body["lines"] if "debit" in l)
c = sum(Decimal(l["credit"]) for l in body["lines"] if "credit" in l)
if d != c:
return JSONResponse({"detail": {"gate": "balanced", "reason": f"{d} against {c}", "fix": ""}}, status_code=422)
jid = body["journal_id"]
if jid in JOURNALS:
return {"journal_id": jid, "entry_no": JOURNALS[jid]["entry_no"], "period_id": "2026-11", "status": "posted", "replay": True,
"lines": len(body["lines"])}
JOURNALS[jid] = {**body, "entry_no": len(JOURNALS) + 1} # the real ledger's entry_no is a number
response.status_code = 201
return {"journal_id": jid, "entry_no": JOURNALS[jid]["entry_no"], "period_id": "2026-11", "status": "posted", "replay": False,
"lines": len(body["lines"])}
@gl_app.get("/periods/open")
def period_open(date: str, company_code: str = "UK1"):
closed = date[:7] in CLOSED_PERIODS
return {"date": date, "company_code": company_code, "period_id": date[:7], "open": not closed,
"status": "closed" if closed else "open", "reason": "closed" if closed else "", "fix": "" if not closed else "reopen"}
@gl_app.get("/companies/{cc}")
def company(cc: str):
if cc not in ("UK1",):
return JSONResponse({"detail": f"no company {cc}"}, status_code=404)
return {"company_code": cc, "name": cc, "country": "GB", "functional_currency": "GBP", "chart_id": "MAIN"}
@gl_app.get("/cost-centres/{cc}")
def cost_centre(cc: str):
c = COST_CENTRES.get(cc)
return c if c else JSONResponse({"detail": f"no cost centre {cc}"}, status_code=404)
def install(product=True, purchasing=True, sales=True, ledger=True):
from brick_inventory.application.use_cases import INV as BRICK
if product:
seam.register_local("product", product_app)
if purchasing:
seam.register_local("purchasing", purchasing_app)
if sales:
seam.register_local("sales-order", so_app)
if ledger:
seam.register_local("gl", gl_app)
seam.register_local("inventory", BRICK.routes_app())
def reset():
for d in (PRODUCTS, POS, RECEIPTS, REQUESTS, WITHDRAWALS, ORDERS, DELIVERY_REFS, RETURN_INSPECTIONS, JOURNALS, GL_REFUSE,
ROUNDING):
d.clear()
for l in (USED, EXPECTED_RECEIPTS, INV_CALLS, SITES, REQUEST_REFUSAL):
l.clear()
CLOSED_PERIODS.clear()
COST_CENTRES.clear()
COST_CENTRES["MAINT"] = {"cost_centre": "MAINT", "name": "Maintenance", "status": "open", "company_code": "UK1"}
for name in ("product", "purchasing", "sales-order", "gl", "inventory", "manufacturing"):
seam.unregister_local(name)