dodil/erp-brick-pos
DODIL ERP brick: point of sale
tests/fakes.py
"""In-memory stand-ins for the neighbours, registered as local seams (erp_platform.seam.register_local): product (its price
list and item master), tax (its rates) and inventory (goods issues). They hold plain dicts the tests fill and answer the
contract infrastructure/seams.py documents; conftest checks every request against the CALLEE's signed spec. A test can take
one DOWN."""
from decimal import Decimal
from typing import Optional
from fastapi import FastAPI, Response
class Neighbours:
def __init__(self):
# price_list_id -> {"currency", "incl", "prices": [(product_id, unit_code, from_quantity, unit_price[, valid_from])]}
self.price_lists: dict = {}
self.products: dict = {} # product_id -> {"name", "tax_category"}
self.rates: dict = {} # country -> [{"category_code", "supply", "consumption", "tax_code", "rate_pct"}]
self.issues: dict = {} # source_id -> the goods issue body, as inventory stored it
self.issue_calls: list = []
self.rate_calls: list = []
self.price_calls: list = []
self.product_calls: list = []
self.product_status: Optional[int] = None
self.tax_status: Optional[int] = None
self.inventory_status: Optional[int] = None # 500 takes inventory down; 422 makes it refuse the request
self.receipts: dict = {} # move_ref -> the restock body inventory stored (INV-59)
self.receipt_calls: list = []
self.companies: dict = {} # gl: company_code -> {"country", "functional_currency"}
self.warehouses: dict = {} # inventory: warehouse_id -> {"company_code", "status", "locations": [(location_id, usage)]}
self.company_calls: list = []
self.warehouse_calls: list = []
self.location_calls: list = []
self.setup_status: Optional[int] = None # 500 takes gl and inventory's setup reads down
self.boms: dict = {} # product_id -> manufacturing's usage-sale bill of materials (see bom())
self.bom_calls: list = []
self.manufacturing_status: Optional[int] = None
self.convert_calls: list = []
self.journals: dict = {} # journal_id -> the body gl stored (the books)
self.journal_calls: list = []
self.gl_status: Optional[int] = None # 500 takes gl down; 409 is a closed period; 422 an unmapped purpose
self.batches: dict = {} # batch_ref -> the body payments stored
self.batch_calls: list = []
self.payments_status: Optional[int] = None
self.product = self._product()
self.tax = self._tax()
self.inventory = self._inventory()
self.manufacturing = self._manufacturing()
self.gl = self._gl()
self.payments = self._payments()
# ── builders ────────────────────────────────────────────────────────────────────────────────
def price_list(self, price_list_id: str, currency: str, prices: list, incl: bool = True) -> None:
self.price_lists[price_list_id] = {"currency": currency, "incl": incl,
"prices": [(row[0], row[1], str(Decimal(row[2])), str(Decimal(row[3])), *row[4:]) for row in prices]}
def product_row(self, product_id: str, tax_category: str, name: str = "", *, stocked: bool = True, base_unit: str = "C62",
units: Optional[dict] = None) -> None:
"""`units`: unit_code -> the factor to the base unit (GRM -> "0.001" of a KGM); the base unit holds three decimals."""
self.products[product_id] = {"name": name or product_id, "tax_category": tax_category, "stocked": stocked, "base_unit": base_unit,
"units": units or {}}
def company(self, company_code: str, country: str = "GB", currency: str = "GBP") -> None:
self.companies[company_code] = {"country": country, "functional_currency": currency}
def warehouse(self, warehouse_id: str, company_code: str, locations: tuple = (), status: str = "active") -> None:
self.warehouses[warehouse_id] = {"company_code": company_code, "status": status, "locations": [(loc, "internal") for loc in locations]}
def bom(self, product_id: str, components: list, *, bom_id: str = "", version: int = 1, valid_from: str = "2026-01-01",
modifiers: Optional[dict] = None) -> None:
"""manufacturing's usage-sale BOM: components [(product_id, quantity per one, unit_code)]; modifiers {modifier_id: {"replaces":
product_id or None, "adds": (product_id, quantity, unit_code)}}."""
self.boms[product_id] = {"bom_id": bom_id or f"{product_id}:sale", "version": version, "valid_from": valid_from,
"components": components, "modifiers": modifiers or {}}
def rate(self, country: str, category: str, tax_code: str, rate_pct: str, consumption: str = "any", supply: str = "goods",
valid_from: str = "2000-01-01") -> None:
self.rates.setdefault(country, []).append({"category_code": category, "supply": supply, "consumption": consumption,
"tax_code": tax_code, "rate_pct": str(Decimal(rate_pct)), "valid_from": valid_from})
# ── the apps ────────────────────────────────────────────────────────────────────────────────
def _product(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/price-lists/{price_list_id}/prices")
def prices(price_list_id: str, response: Response, on: Optional[str] = None, limit: int = 500, offset: int = 0):
"""product PRD-40: 200 {price_list_id, currency, prices_include_tax, on, next_change_on, prices: [...], more}."""
me.price_calls.append({"price_list_id": price_list_id, "on": on, "limit": limit, "offset": offset})
if me.product_status:
response.status_code = me.product_status
return {"detail": "down"}
pl = me.price_lists.get(price_list_id)
if pl is None:
response.status_code = 404
return {"detail": f"no price list {price_list_id}"}
dated = lambda row: row[4] if len(row) > 4 else "2026-01-01" # noqa: E731
started = [row for row in pl["prices"] if on is None or dated(row) <= on] # a price dated later is not yet
live = [row for row in started if dated(row) == max(dated(x) for x in started if x[:3] == row[:3])] # the row in force per product, unit, from_quantity
later = sorted(row[4] for row in pl["prices"] if len(row) > 4 and on is not None and row[4] > on)
rows = sorted(live)[offset:offset + limit]
return {"price_list_id": price_list_id, "currency": pl["currency"], "prices_include_tax": pl["incl"], "on": on,
"next_change_on": later[0] if later else None,
"prices": [{"product_id": row[0], "unit_code": row[1], "from_quantity": row[2], "unit_price": row[3], "pricing_quantity": 1,
"pricing_unit_code": row[1], "valid_from": row[4] if len(row) > 4 else "2026-01-01", "valid_to": None}
for row in rows],
"more": offset + limit < len(live)}
@app.get("/products/{product_id}")
def product(product_id: str, response: Response, on: Optional[str] = None):
"""product PRD-28 (the part pos reads): {product_id, name, tax_category, base_unit_code}."""
me.product_calls.append(product_id)
if me.product_status:
response.status_code = me.product_status
return {"detail": "down"}
p = me.products.get(product_id)
if p is None:
response.status_code = 404
return {"detail": f"no product {product_id}"}
return {"product_id": product_id, "name": p["name"], "tax_category": p["tax_category"], "base_unit_code": p["base_unit"],
"stocked_default": p["stocked"]}
@app.get("/products/{product_id}/convert")
def convert(product_id: str, response: Response, quantity: str, from_unit: str, to_unit: str):
"""product PRD-21: 200 {product_id, from_unit, to_unit, quantity, exact, converted, in_base_unit, base_unit_code}; 422 when the
exact fraction does not fit the target unit's three decimals."""
me.convert_calls.append({"product_id": product_id, "quantity": quantity, "from_unit": from_unit, "to_unit": to_unit})
p = me.products.get(product_id)
factor = Decimal(1) if p and from_unit == p["base_unit"] else Decimal(p["units"][from_unit]) if p and from_unit in p["units"] else None
if p is None or factor is None:
response.status_code = 404
return {"detail": f"no unit {from_unit} for {product_id}"}
converted = Decimal(quantity) * factor
if converted.as_tuple().exponent < -3 and converted != converted.quantize(Decimal("0.001")):
response.status_code = 422
return {"gate": "units_exact", "reason": "the exact fraction does not fit the unit's decimals"}
return {"product_id": product_id, "from_unit": from_unit, "to_unit": to_unit, "quantity": quantity, "exact": "n/d",
"converted": format(converted, "f"), "in_base_unit": format(converted, "f"), "base_unit_code": p["base_unit"]}
return app
def _tax(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/rates")
def rates(response: Response, country: str, categories: str, on: str, consumption: Optional[str] = None,
treatment_code: Optional[str] = None):
"""tax TAX-52: 200 {country, on, treatment_code, valid_until, rates: [...], missing: [category_code]}."""
me.rate_calls.append({"country": country, "categories": categories, "on": on, "consumption": consumption})
if me.tax_status:
response.status_code = me.tax_status
return {"detail": "down"}
wanted = categories.split(",")
known = [r for r in me.rates.get(country, []) if r["category_code"] in wanted
and (consumption is None or r["consumption"] in (consumption, "any"))]
rows = []
for r in known: # per category and consumption: the latest rule in force on `on`
same = [x for x in known if (x["category_code"], x["supply"], x["consumption"]) == (r["category_code"], r["supply"], r["consumption"])
and x["valid_from"] <= on]
if same and r is max(same, key=lambda x: x["valid_from"]):
rows.append({k: v for k, v in r.items() if k != "valid_from"})
later = sorted(r["valid_from"] for r in known if r["valid_from"] > on)
return {"country": country, "on": on, "treatment_code": "uk_consumer", "valid_until": later[0] if later else None, "rates": rows,
"missing": [c for c in wanted if c not in {r["category_code"] for r in rows}]}
return app
def _inventory(self) -> FastAPI:
app, me = FastAPI(), self
@app.post("/goods-issues/from-source")
def issue(body: dict, response: Response):
"""inventory INV-58: 201 {issue_id, moves: [...]}; 200 the same for known move_refs; 422 {gate, reason, fix}."""
me.issue_calls.append(body)
if me.inventory_status == 422:
response.status_code = 422
return {"detail": "units are not exact", "gate": "units_exact"}
if me.inventory_status:
response.status_code = me.inventory_status
return {"detail": "down"}
known = body["source_id"] in me.issues
me.issues.setdefault(body["source_id"], body)
response.status_code = 200 if known else 201
return {"issue_id": f"GI-{body['source_id']}", "moves": [{"move_ref": ln["move_ref"], "move_id": f"M-{ln['move_ref']}",
"quantity_base": ln["quantity"], "value": "0.00"} for ln in body["lines"]]}
@app.get("/warehouses/{warehouse_id}")
def warehouse(warehouse_id: str, response: Response):
"""inventory INV-01: 200 {warehouse_id, company_code, status}; 404 for an unknown warehouse."""
me.warehouse_calls.append(warehouse_id)
if me.setup_status:
response.status_code = me.setup_status
return {"detail": "down"}
w = me.warehouses.get(warehouse_id)
if w is None:
response.status_code = 404
return {"detail": f"no warehouse {warehouse_id}"}
return {"warehouse_id": warehouse_id, "company_code": w["company_code"], "status": w["status"]}
@app.get("/locations")
def locations(response: Response, usage: str, warehouse_id: Optional[str] = None):
"""inventory INV-01: 200 {locations: [{location_id, warehouse_id, name, usage, default}]}."""
me.location_calls.append({"usage": usage, "warehouse_id": warehouse_id})
if me.setup_status:
response.status_code = me.setup_status
return {"detail": "down"}
rows = [{"location_id": loc, "warehouse_id": wid, "name": loc, "usage": u, "default": False}
for wid, w in sorted(me.warehouses.items()) if warehouse_id in (None, wid) for loc, u in w["locations"] if u == usage]
return {"locations": rows}
@app.post("/receipts")
def receive(body: dict, response: Response):
"""inventory INV-59: 201 {move_id, quantity_base, value}; 200 for a repeated move_ref; 422 {gate: reverses_known | within_issued}."""
me.receipt_calls.append(body)
if me.inventory_status == 422:
response.status_code = 422
return {"detail": "units are not exact", "gate": "units_exact"}
if me.inventory_status:
response.status_code = me.inventory_status
return {"detail": "down"}
issued = {ln["move_ref"]: ln for issue in me.issues.values() for ln in issue["lines"]}
was = issued.get(body["reverses_move_ref"])
if was is None:
response.status_code = 422
return {"gate": "reverses_known", "reason": f"no move {body['reverses_move_ref']}"}
if Decimal(body["quantity"]) > Decimal(was["quantity"]):
response.status_code = 422
return {"gate": "within_issued", "reason": "more than that move issued"}
known = body["move_ref"] in me.receipts
me.receipts.setdefault(body["move_ref"], body)
response.status_code = 200 if known else 201
return {"move_id": f"R-{body['move_ref']}", "quantity_base": body["quantity"], "value": "0.00"}
return app
def _manufacturing(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/boms/{product_id}/explode")
def explode(product_id: str, response: Response, usage: str, quantity: str, unit_code: str, modifiers: str, on: str):
"""manufacturing MFG-30: 200 {bom_id, version, valid_from, components: [{product_id, quantity (base unit), unit_code}]}; 404 when
no released version is valid on the date. Deterministic for the same query."""
me.bom_calls.append({"product_id": product_id, "usage": usage, "quantity": quantity, "unit_code": unit_code, "modifiers": modifiers,
"on": on})
if me.manufacturing_status:
response.status_code = me.manufacturing_status
return {"detail": "down"}
b = me.boms.get(product_id)
if b is None or usage != "sale" or b["valid_from"] > on:
response.status_code = 404
return {"detail": f"no released bill of materials of {product_id} on {on}"}
parts = list(b["components"])
for m in [x for x in modifiers.split(",") if x]:
mod = b["modifiers"][m]
if mod.get("replaces"):
parts = [c for c in parts if c[0] != mod["replaces"]]
parts.append(mod["adds"])
qty = Decimal(quantity)
return {"bom_id": b["bom_id"], "version": b["version"], "valid_from": b["valid_from"],
"components": [{"product_id": p, "quantity": format((Decimal(q) * qty).normalize(), "f"), "unit_code": u} for p, q, u in parts]}
return app
def _gl(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/companies/{company_code}")
def company(company_code: str, response: Response):
"""gl GL-61: 200 {company_code, name, country, functional_currency, chart_id, bp_id, financial_year_end_month, ledgers}; 404 unknown."""
me.company_calls.append(company_code)
if me.setup_status:
response.status_code = me.setup_status
return {"detail": "down"}
c = me.companies.get(company_code)
if c is None:
response.status_code = 404
return {"detail": f"no company {company_code}"}
return {"company_code": company_code, "name": company_code, **c, "chart_id": "UK", "bp_id": None, "financial_year_end_month": 12,
"ledgers": ["L1"]}
@app.post("/journal-entry/journals")
def post(body: dict, response: Response):
"""gl GL-30: 201 {journal_id, entry_no, status: posted, replay: false}; 200 replay for the same body; 409 {gate: period_open |
journal_conflict}; 422 {gate: balanced} when debit and credit differ."""
me.journal_calls.append(body)
if me.gl_status == 409:
response.status_code = 409
return {"gate": "period_open", "reason": "the period of 2026-11 is closed", "fix": "date it in an open period"}
if me.gl_status:
response.status_code = me.gl_status
return {"detail": "down", "gate": "purposes_mapped" if me.gl_status == 422 else None}
debit = sum((Decimal(ln.get("debit", "0")) for ln in body["lines"]), Decimal(0))
credit = sum((Decimal(ln.get("credit", "0")) for ln in body["lines"]), Decimal(0))
if debit != credit:
response.status_code = 422
return {"gate": "balanced", "reason": f"debit {debit} credit {credit}"}
jid = body["journal_id"]
if jid in me.journals and me.journals[jid] != body:
response.status_code = 409
return {"gate": "journal_conflict", "reason": "a different body for the same journal_id"}
known = jid in me.journals
me.journals.setdefault(jid, body)
response.status_code = 200 if known else 201
return {"journal_id": jid, "entry_no": len(me.journals), "period_id": "2026-11", "status": "posted", "replay": known,
"lines": len(body["lines"])}
return app
def _payments(self) -> FastAPI:
app, me = FastAPI(), self
@app.post("/deposit-batches/expected")
def expected(body: dict, response: Response):
"""payments PAY-49: 201 {batch_id (= batch_ref), status: deposited, total}; 200 the same for the same batch_ref; 422 {gate:
bank_account_receives | bounds}."""
me.batch_calls.append(body)
if me.payments_status == 422:
response.status_code = 422
return {"gate": "bank_account_receives", "reason": "the account does not receive cash", "fix": "pick a receiving account"}
if me.payments_status:
response.status_code = me.payments_status
return {"detail": "down"}
known = body["batch_ref"] in me.batches
me.batches.setdefault(body["batch_ref"], body)
response.status_code = 200 if known else 201
return {"batch_id": body["batch_ref"], "status": "deposited", "total": body["amount"]}
return app
def seed_uk(nb: Neighbours) -> None:
"""The London shop's world: a GBP price list (prices include VAT) and UK VAT at 20% on drinks and food."""
nb.company("UK01")
nb.company("DE01", "DE", "EUR")
nb.warehouse("LON-SOHO", "UK01", ("LON-SOHO/BAR",))
nb.warehouse("BER-MITTE", "DE01", ("BER-MITTE/BAR",))
nb.price_list("LIST-UK-SHOP", "GBP", [("LATTE-REG", "C62", "1", "3.60"), ("LATTE-OAT", "C62", "1", "4.00"),
("CROISSANT", "C62", "1", "2.40"), ("FLAT-WHITE", "C62", "1", "3.40"),
("BEANS-1KG", "C62", "1", "18.99")])
for pid, cat in (("LATTE-REG", "hot_drink"), ("LATTE-OAT", "hot_drink"), ("FLAT-WHITE", "hot_drink"),
("CROISSANT", "food_takeaway"), ("BEANS-1KG", "packaged_food")):
nb.product_row(pid, cat)
for cat in ("hot_drink", "food_takeaway"):
nb.rate("GB", cat, "S", "20.00")
nb.rate("GB", "packaged_food", "Z", "0.00")
def seed_de(nb: Neighbours) -> None:
"""A German shop: EUR prices, and food at 19% to eat in but 7% to take away (the rate depends on the answer)."""
nb.price_list("LIST-DE-SHOP", "EUR", [("LATTE-REG", "C62", "1", "3.90"), ("PRETZEL", "C62", "1", "2.50")])
nb.product_row("LATTE-REG", "hot_drink")
nb.product_row("PRETZEL", "food")
nb.rate("DE", "hot_drink", "S19", "19.00")
nb.rate("DE", "food", "S19", "19.00", consumption="eat_in")
nb.rate("DE", "food", "R7", "7.00", consumption="takeaway")