dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
tests/fakes.py
"""In-memory stand-ins for the neighbours, registered as local seams (erp_platform.seam.register_local): product (units and exact
conversion), inventory (availability, valuation, the issue / receipt / valuation-adjustment doors) and gl (the journal door).
They hold plain dicts the tests fill and answer the contract the callee's signed spec states (tests/conftest.py checks every call
against that spec). A test can take one DOWN, or make one refuse."""
from decimal import ROUND_CEILING, ROUND_HALF_UP, Decimal
from fractions import Fraction
from typing import Optional
from fastapi import FastAPI, Request, Response
def q2(v: Decimal) -> Decimal:
return v.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
class Neighbours:
def __init__(self):
# product: id -> {base unit, units {code: Fraction of the base unit}}
self.products = {"GREEN-ETH": ("KGM", {"KGM": Fraction(1), "GRM": Fraction(1, 1000)}),
"ROAST-ETH": ("KGM", {"KGM": Fraction(1), "GRM": Fraction(1, 1000)}),
"CHAFF": ("KGM", {"KGM": Fraction(1)}), "BAG-ETH-250": ("C62", {"C62": Fraction(1)}),
"POUCH-250": ("C62", {"C62": Fraction(1)}), "LABEL-ETH": ("C62", {"C62": Fraction(1)}),
"LATTE-REG": ("C62", {"C62": Fraction(1)}), "ROAST-HOUSE": ("KGM", {"KGM": Fraction(1)}),
"MILK-WHOLE": ("LTR", {"LTR": Fraction(1)}), "MILK-OAT": ("LTR", {"LTR": Fraction(1)}),
"CUP-12OZ": ("C62", {"C62": Fraction(1)}), "LID-12OZ": ("C62", {"C62": Fraction(1)}),
"THIRD": ("KGM", {"KGM": Fraction(1), "PCT": Fraction(1, 3)})}
self.costs = {"GREEN-ETH": Decimal("6.2000"), "ROAST-ETH": Decimal("9.3899"), "CHAFF": Decimal("0.0000")}
self.available = {("GREEN-ETH", "ROASTERY"): Decimal("140.000")}
self.onhand = {("GREEN-ETH", "ROASTERY/RAW"): Decimal("500.000")}
self.moves: dict = {} # move_ref -> answer
self.calls: list = [] # (target, body) of every write
self.journals: dict = {}
self.adjustments: dict = {}
self.status = {"inventory": None, "gl": None, "product": None} # an HTTP status to answer with (a neighbour down)
self.refuse_receipts: Optional[dict] = None # {"gate": ..., "reason": ...}: inventory answers 422
self.period_closed = False
self.cost_centres = {"ROASTERY": "open", "PACKING": "open", "OLD": "closed"}
self.standard_prices: dict = {} # (company, product, valid_from, source_ref) -> body
self.refuse_standard: Optional[dict] = None
self.traces: dict = {} # (product, lot) -> inventory's trace answer
self.round_multiple: dict = {} # product -> Decimal (round UP to a multiple)
self.sites: list = [] # product-sites rows
self.demand: dict = {} # (product, warehouse) -> [{needed_on, open_base}]
self.product = self._product()
self.sales = self._sales()
self.inventory = self._inventory()
self.gl = self._gl()
# ── product ──────────────────────────────────────────────────────────────────────────────
def _product(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/products/{product_id}")
def get_product(response: Response, product_id: str, on: Optional[str] = None):
if me.status["product"]:
response.status_code = me.status["product"]
return {"detail": "down"}
if product_id not in me.products:
response.status_code = 404
return {"detail": "no such product"}
base, units = me.products[product_id]
return {"product_id": product_id, "base_unit_code": base, "kind": "goods", "roles": {},
"units": [{"unit_code": u} for u in units], "lifecycle": "active"}
@app.get("/products/{product_id}/convert")
def convert(response: Response, product_id: str, quantity: str, from_unit: str, to_unit: str):
base, units = me.products[product_id]
exact = Decimal(quantity) * units[from_unit].numerator / units[from_unit].denominator / (
Decimal(units[to_unit].numerator) / Decimal(units[to_unit].denominator))
if exact != exact.quantize(Decimal("0.0001")):
response.status_code = 422
return {"detail": "does not fit", "gate": "units_exact", "reason": "does not fit"}
return {"product_id": product_id, "from_unit": from_unit, "to_unit": to_unit, "quantity": quantity,
"converted": str(exact.quantize(Decimal("0.0001"))), "in_base_unit": str(exact), "base_unit_code": base}
@app.get("/products/{product_id}/round-proposal")
def round_proposal(response: Response, product_id: str, site_id: str, quantity: str, unit_code: str, on: Optional[str] = None):
if me.status["product"]:
response.status_code = me.status["product"]
return {"detail": "down"}
q, m = Decimal(quantity), me.round_multiple.get(product_id)
rounded = (q / m).to_integral_value(rounding=ROUND_CEILING) * m if m else q
return {"product_id": product_id, "proposed": quantity, "rounded": str(rounded), "unit_code": unit_code, "steps_applied": [],
"purchase_rule": {"met": True, "reason": ""}}
@app.get("/product-sites")
def product_sites(response: Response, site_id: str, planning: Optional[str] = None, on: Optional[str] = None):
if me.status["product"]:
response.status_code = me.status["product"]
return {"detail": "down"}
return [r for r in me.sites if r["site_id"] == site_id]
return app
# ── sales-order ──────────────────────────────────────────────────────────────────────────
def _sales(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/demand")
def demand(product_id: str, warehouse_id: str, limit: int = 500, offset: int = 0):
rows = me.demand.get((product_id, warehouse_id), [])
page = rows[offset:offset + limit]
return {"product_id": product_id, "warehouse_id": warehouse_id, "from": "", "to": "",
"demand": [{"needed_on": r["needed_on"], "sales_order_id": "SO-1", "line_no": 1, "open_quantity": str(r["open_base"]),
"unit_code": "C62", "open_base": str(r["open_base"]), "confirmed": True} for r in page],
"more": offset + limit < len(rows)}
return app
# ── inventory ────────────────────────────────────────────────────────────────────────────
def _inventory(self) -> FastAPI:
app, me = FastAPI(), self
def down(response):
if me.status["inventory"]:
response.status_code = me.status["inventory"]
return {"detail": "down"}
@app.get("/availability/{product_id}")
def availability(response: Response, product_id: str, warehouse_id: Optional[str] = None):
if (d := down(response)):
return d
have = me.available.get((product_id, warehouse_id), Decimal("0"))
return {"product_id": product_id, "base_unit_code": me.products[product_id][0],
"warehouses": [{"warehouse_id": warehouse_id, "on_hand": str(have), "reserved": "0", "available": str(have),
"in_transit_in": "0"}]}
@app.get("/valuation/products/{product_id}")
def valuation(response: Response, product_id: str, company_code: str, as_of: Optional[str] = None):
if (d := down(response)):
return d
cost = me.costs.get(product_id)
return {"product_id": product_id, "company_code": company_code, "quantity_base": "0", "value_amount": "0.00",
"average_cost": None if cost is None else str(cost), "as_of": as_of}
@app.post("/goods-issues/from-source")
async def issue(request: Request, response: Response):
if (d := down(response)):
return d
body = await request.json()
me.calls.append(("issue", body))
moves, fresh = [], False
for ln in body["lines"]:
ref = ln["move_ref"]
if ref not in me.moves:
key = (ln["product_id"], body["location_id"])
have = me.onhand.get(key, Decimal("0"))
if have < Decimal(ln["quantity"]):
response.status_code = 409
return {"detail": "short", "gate": "on_hand", "reason": f"only {have} of {ln['product_id']} at {body['location_id']}",
"fix": "receive stock"}
me.onhand[key] = have - Decimal(ln["quantity"])
me.moves[ref] = {"move_ref": ref, "move_id": f"M{len(me.moves) + 1}", "quantity_base": ln["quantity"],
"lot_code": ln.get("lot_code"),
"value": str(q2(Decimal(ln["quantity"]) * me.costs.get(ln["product_id"], Decimal("0"))))}
fresh = True
moves.append(me.moves[ref])
response.status_code = 201 if fresh else 200
return {"issue_id": body["source_id"], "moves": moves}
@app.post("/receipts")
async def receipt(request: Request, response: Response):
if (d := down(response)):
return d
body = await request.json()
me.calls.append(("receipt", body))
if me.refuse_receipts:
response.status_code = 422
return {"detail": me.refuse_receipts["reason"], **me.refuse_receipts}
ref = body["move_ref"]
fresh = ref not in me.moves
if fresh:
me.moves[ref] = {"move_id": f"M{len(me.moves) + 1}", "quantity_base": body["quantity"],
"value": str(q2(Decimal(body["quantity"]) * Decimal(body["unit_cost"])))}
response.status_code = 201 if fresh else 200
return me.moves[ref]
@app.post("/standard-prices")
async def standard_price(request: Request, response: Response):
if (d := down(response)):
return d
body = await request.json()
me.calls.append(("standard_price", body))
if me.refuse_standard:
response.status_code = 409
return {"detail": me.refuse_standard["reason"], **me.refuse_standard}
key = (body["company_code"], body["product_id"], body["valid_from"], body["source_ref"])
fresh = key not in me.standard_prices
me.standard_prices[key] = body
response.status_code = 201 if fresh else 200
return {"company_code": body["company_code"], "product_id": body["product_id"], "valid_from": body["valid_from"],
"standard_price": body["standard_price"], "lifecycle": "scheduled"}
@app.get("/lots/{product_id}/{lot_code}/trace")
def lot_trace(response: Response, product_id: str, lot_code: str):
if (d := down(response)):
return d
got = me.traces.get((product_id, lot_code))
if got is None:
response.status_code = 404
return {"detail": "unknown lot"}
return got
@app.post("/valuation-adjustments")
async def adjustment(request: Request, response: Response):
if (d := down(response)):
return d
body = await request.json()
me.calls.append(("adjustment", body))
fresh = body["adjustment_ref"] not in me.adjustments
me.adjustments[body["adjustment_ref"]] = body
response.status_code = 201 if fresh else 200
return {"adjustment_ref": body["adjustment_ref"], "on_hand_amount": body["amount"], "cost_of_sales_amount": "0.00"}
return app
# ── gl ───────────────────────────────────────────────────────────────────────────────────
def _gl(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/cost-centres/{cost_centre}")
def cost_centre(response: Response, cost_centre: str):
if cost_centre not in me.cost_centres:
response.status_code = 404
return {"detail": "unknown"}
return {"cost_centre": cost_centre, "name": cost_centre, "status": me.cost_centres[cost_centre], "company_code": "UK01"}
@app.post("/journal-entry/journals")
async def post_journal(request: Request, response: Response):
if me.status["gl"]:
response.status_code = me.status["gl"]
return {"detail": "down"}
body = await request.json()
me.calls.append(("journal", body))
if me.period_closed:
response.status_code = 409
return {"detail": "period closed", "gate": "period_open", "reason": "the period is closed", "fix": "reopen it"}
debit = sum(Decimal(x.get("debit", "0")) for x in body["lines"])
credit = sum(Decimal(x.get("credit", "0")) for x in body["lines"])
if debit != credit:
response.status_code = 422
return {"detail": "unbalanced", "gate": "balanced", "reason": f"{debit} != {credit}"}
replay = body["journal_id"] in me.journals
me.journals[body["journal_id"]] = body
response.status_code = 200 if replay else 201
return {"journal_id": body["journal_id"], "status": "posted", "replay": replay, "lines": len(body["lines"])}
return app