dodil/erp-brick-sales-order
DODIL ERP brick: sales order
tests/fakes.py
"""In-memory neighbours, shaped from THEIR spec's contract: product (find_price), tax (compute), payments (credit_status),
invoicing (raise_draft, plan_billing, billed, cancel_drafts), business-partner (the customer can transact)."""
from datetime import date, timedelta
from decimal import Decimal
from contract_checks import spec_check
from brick_sales_order.infrastructure.seams import (
GlPort,
InventoryPort,
InvoicingPort,
MapsPort,
Missing,
OdooPort,
PartnersPort,
PaymentsPort,
ProductPort,
PurchasingPort,
TaxPort,
credit_body,
expected_return_body,
raise_body,
)
D = Decimal
class FakeProduct(ProductPort):
"""COLA-075: 1.20 a bottle (C62) and 24.00 a pack (PK24) from 2026-10-01, 1.30 a bottle from 2026-12-01.
SETUP by the hour (HUR), delivered policy. WIDGET each (C62) 10.00, delivered. HOST per month, recurring."""
PRODUCTS = {
"COLA-075": dict(sales="C62", policy="ordered", kind="", plan="", tax="standard",
prices={"C62": [("2026-10-01", "1.20"), ("2026-12-01", "1.30")], "PK24": [("2026-10-01", "24.00")]},
conv={"C62": (1, 1), "PK24": (24, 1)}),
"SETUP": dict(sales="HUR", policy="delivered", kind="", plan="", tax="standard", prices={"HUR": [("2026-10-01", "80.00")]}, conv={"HUR": (1, 1)}),
"WIDGET": dict(sales="C62", policy="delivered", kind="", plan="", tax="standard", prices={"C62": [("2026-10-01", "10.00")]}, conv={"C62": (1, 1)}),
"FEE": dict(sales="C62", policy="ordered", kind="", plan="", tax="standard", prices={"C62": [("2026-10-01", "1000.00")]}, conv={"C62": (1, 1)}),
"JUICE-075": dict(sales="C62", policy="ordered", kind="", plan="", tax="standard", base="LTR", base_decimals=3,
prices={"C62": [("2026-10-01", "2.00")]}, conv={"C62": (3, 4)}),
"HOST": dict(sales="MON", policy="ordered", kind="recurring", plan="monthly", tax="standard", prices={"MON": [("2026-10-01", "120.00")]}, conv={"MON": (1, 1)}),
"PROJ-INSTALL": dict(sales="C62", policy="ordered", kind="", plan="", tax="standard", prices={"C62": [("2026-10-01", "20000.00")]}, conv={"C62": (1, 1)}),
"DRILL-500": dict(sales="C62", policy="ordered", kind="", plan="", tax="standard", prices={"C62": [("2026-10-01", "320.00")]}, conv={"C62": (1, 1)}),
"PRINTER-X": dict(sales="C62", policy="ordered", kind="", plan="", tax="standard", prices={"C62": [("2026-10-01", "180.00")]}, conv={"C62": (1, 1)}),
"COLA-150": dict(sales="C62", policy="ordered", kind="", plan="", tax="standard", prices={"C62": [("2026-10-01", "1.80")]}, conv={"C62": (1, 1)}),
# priced PER 100 C62 on the customer's list (PRD-31); a box of 500 has no price row of its own and is priced in the row's unit
"SCREW-M4": dict(sales="C62", policy="ordered", kind="", plan="", tax="standard", prices={"C62": [("2026-10-01", "2.4500")]},
conv={"C62": (1, 1), "BX500": (500, 1)}, pricing_quantity=100, priced_in="C62"),
}
TYPES = {"SETUP": "SERVICE", "HOST": "SERVICE", "FEE": "SERVICE", "PROJ-INSTALL": "SERVICE"} # every other product is TRADING
CATEGORY = {"COLA-075": "drinks", "COLA-150": "drinks", "JUICE-075": "drinks", "SCREW-M4": "hardware"}
WEIGHTS = {"COLA-075": "0.79", "COLA-150": "1.60"} # gross KGM per ONE C62, row in force from 2026-10-01
DELIVERY_RULES = {"COLA-075": (6, "C62")} # delivered in multiples of 6 C62 (PRD-30)
def __init__(self):
self.calls = []
def find_price(self, caller, *, product_id, quantity, unit_code, bp_id, currency, priced_on):
self.calls.append((product_id, quantity, unit_code, bp_id, currency, priced_on))
spec_check("product", "GET", f"/products/{product_id}/price",
{"currency": currency, "quantity": str(quantity), "bp_id": str(bp_id), "priced_on": priced_on.isoformat(),
**({"unit_code": unit_code} if unit_code else {})})
p = self.PRODUCTS.get(product_id)
if p is None:
raise Missing(f"product {product_id}", f"no product {product_id}", 404)
unit = unit_code or p["sales"]
row_unit = unit if unit in p["prices"] else p.get("priced_in") if unit in p["conv"] else None
if row_unit is None:
raise Missing(f"product {product_id}", f"{product_id} is not counted in {unit}: its units are {', '.join(p['prices'])}", 422)
rows = [r for r in p["prices"][row_unit] if date.fromisoformat(r[0]) <= priced_on]
if not rows:
raise Missing(f"product {product_id}", f"no {currency} price for {product_id} per {unit} on {priced_on}", 404)
frm, price = rows[-1]
n, d = p["conv"][unit]
pn, pd = p["conv"][row_unit]
weight = self.WEIGHTS.get(product_id)
return {"product_id": product_id, "quantity": str(quantity), "unit_code": unit, "bp_id": bp_id, "currency": currency,
"priced_on": priced_on.isoformat(), "unit_price": price, "conversion": {"unit_code": unit, "numerator": n, "denominator": d},
"pricing_quantity": p.get("pricing_quantity", 1), "pricing_unit_code": row_unit, "pricing_numerator": pn, "pricing_denominator": pd,
"priced_from": {"price_list_id": "LIST-GBP", "rule_seq": None, "row": {"product_id": product_id, "unit_code": unit, "valid_from": frm}},
"line": {"product_id": product_id, "name": product_id, "description": f"{product_id} description",
"sales_unit_code": p["sales"], "tax_category": p["tax"], "billing_kind": p["kind"], "billing_plan": p["plan"],
"invoicing_policy": p["policy"], "product_type_code": self.TYPES.get(product_id, "TRADING"),
"category_code": self.CATEGORY.get(product_id, ""),
**({"measures": {"gross_weight": weight, "weight_unit_code": "KGM", "valid_from": "2026-10-01"}} if weight else {})}}
def rounding(self, caller, *, product_id, quantity, unit_code, on):
"""product's rounding answer for role delivery (PRD-32): met, the nearest multiples either side, and the reason in the rule's own unit."""
spec_check("product", "GET", f"/products/{product_id}/rounding",
{"role": "delivery", "quantity": str(quantity), "unit_code": unit_code, "on": on.isoformat()})
p = self.PRODUCTS.get(product_id)
if p is None:
raise Missing(f"product {product_id}", f"no product {product_id}", 404)
rule = self.DELIVERY_RULES.get(product_id)
if rule is None:
return {"product_id": product_id, "role": "delivery", "rule": None, "quantity_in_rule_unit": str(quantity), "met": True,
"next_down": None, "next_up": None, "reason": "no rule for role delivery"}
multiple, rule_unit = rule
n, d = p["conv"][unit_code]
rn, rd = p["conv"][rule_unit]
q = D(quantity) * n * rd / (d * rn)
met = q % multiple == 0
down, up = (q // multiple) * multiple, (q // multiple + 1) * multiple
def fmt(x):
return str(int(x)) if x == x.to_integral() else str(x.normalize())
return {"product_id": product_id, "role": "delivery", "rule": {"unit_code": rule_unit, "min_quantity": None, "rounding_multiple": str(multiple), "valid_from": "2026-10-01"},
"quantity_in_rule_unit": fmt(q), "met": met, "next_down": None if met else fmt(down), "next_up": None if met else fmt(up),
"reason": "" if met else f"{product_id} is delivered in multiples of {multiple} {rule_unit}"}
def convert(self, caller, *, product_id, quantity, unit_code):
"""product's convert (unit -> the same unit): in_base_unit is the exact conversion rounded half up to the base unit's decimals."""
spec_check("product", "GET", f"/products/{product_id}/convert", {"quantity": str(quantity), "from_unit": unit_code, "to_unit": unit_code})
p = self.PRODUCTS.get(product_id)
if p is None:
raise Missing(f"product {product_id}", f"no product {product_id}", 404)
n, d = p["conv"][unit_code]
places = p.get("base_decimals", 0)
return {"product_id": product_id, "from_unit": unit_code, "to_unit": unit_code, "quantity": str(quantity), "exact": f"{n}/{d}",
"converted": str(quantity), "base_unit_code": p.get("base", "C62"),
"in_base_unit": str((D(quantity) * n / d).quantize(D(1).scaleb(-places)))}
class FakeTax(TaxPort):
"""20% on 'standard', nothing on 'exempt'."""
def __init__(self):
self.calls = []
self.down = False
def compute(self, caller, *, lines, bp_id, tax_point, currency, document_ref="", ship_to_country=""):
self.calls.append((lines, bp_id, tax_point, currency))
self.countries = getattr(self, "countries", []) + [ship_to_country]
spec_check("tax", "POST", "/documents/compute",
{"lines": [{"line_no": int(ln["ref"]), "category_code": ln.get("tax_category") or None, "amount": str(ln["net_amount"])} for ln in lines],
"document_ref": document_ref, "direction": "sale", "tax_point": tax_point.isoformat(), "bp_id": str(bp_id),
"currency": currency, **({"ship_to_country": ship_to_country} if ship_to_country else {})})
if self.down:
raise Missing("tax", "tax did not answer", 503)
out, total = [], D(0)
for ln in lines:
rate = D(0) if ln["tax_category"] == "exempt" else D(20)
amount = (D(ln["net_amount"]) * rate / 100).quantize(D("0.01"))
total += amount
out.append({"ref": ln["ref"], "tax_code": "E" if rate == 0 else "S", "rate": str(rate), "tax_amount": str(amount)})
return {"lines": out, "tax_total": str(total), "method": "line"}
class FakePayments(PaymentsPort):
def __init__(self):
self.status = {}
def credit_status(self, caller, bp_id):
spec_check("payments", "GET", f"/customers/{int(bp_id)}/credit-ok")
return self.status.get(int(bp_id), {"hold": False, "reason": "", "mode": "block"})
class FakeInvoicing(InvoicingPort):
"""Keeps a draft per source_id (asking twice returns the same one); `issue()` makes a draft an issued invoice."""
def __init__(self):
self.drafts, self.schedules, self.issued, self.cancelled = {}, {}, {}, []
self.issued_ids = set() # SOURCE ids of the drafts that were issued
self.invoice_ids = {} # source id -> the invoice id ar minted (AR-0001): never equal to the source id
self.down = False
self.billed_rows, self.credits = [], {} # billed: invoice and credit-note lines issued (the rebate basis); credits by source_id
def raise_draft(self, *, source_id, bp_id, currency, lines, kind, customer_po_ref, sold_to_bp_id=0, ship_to_bp_id=0, bill_to_bp_id=0,
deductions=None, issue_hint=""):
spec_check("ar", "POST", "/invoices/raise", raise_body(source_id=source_id, bp_id=bp_id, currency=currency, lines=lines, kind=kind,
customer_po_ref=customer_po_ref, sold_to_bp_id=sold_to_bp_id, ship_to_bp_id=ship_to_bp_id,
bill_to_bp_id=bill_to_bp_id, deductions=deductions, issue_hint=issue_hint))
if self.down:
raise Missing("invoicing", "invoicing did not answer", 503)
self.calls_raise = getattr(self, "calls_raise", []) + [source_id]
invoice_id = self.invoice_ids.setdefault(source_id, f"AR-{len(self.invoice_ids) + 1:04d}")
self.drafts.setdefault(source_id, {"invoice_id": invoice_id, "bp_id": bp_id, "currency": currency, "lines": lines, "kind": kind,
"customer_po_ref": customer_po_ref, "sold_to_bp_id": sold_to_bp_id, "ship_to_bp_id": ship_to_bp_id,
"bill_to_bp_id": bill_to_bp_id, "deductions": deductions or [], "issue_hint": issue_hint})
if issue_hint == "cash": # invoicing's owner control auto_issue_sources contains sales_order:cash: issued at once
self.issue(source_id)
return {"invoice_id": invoice_id, "status": "issued" if source_id in self.issued_ids else "draft"}
def plan_billing(self, *, schedule_id, bp_id, currency, line):
src, line_id = schedule_id.rsplit("|", 1)
spec_check("ar", "POST", "/schedules",
{"source_kind": "sales_order", "source_id": src, "line_id": line_id, "bp_id": str(bp_id), "currency": currency,
"starts_on": line["starts_on"], "periods": int(D(str(line.get("quantity") or 1))), "unit_price": str(line["unit_price"]),
"billing_kind": {"recurring": "recurring", "metered": "usage", "instalment": "instalment"}.get(line.get("billing_kind") or "", "recurring"),
"billing_plan": {"monthly": "monthly", "quarterly": "quarterly", "annual": "annual", "periodic": "periodic",
"arrears": "periodic"}.get(line.get("billing_plan") or "", "monthly")})
self.schedules.setdefault(schedule_id, {"bp_id": bp_id, "currency": currency, **line})
return {"schedule_id": schedule_id}
def billed(self, sales_order_id):
spec_check("ar", "GET", f"/sources/sales_order/{sales_order_id}/billed")
if self.down:
raise Missing("invoicing", "invoicing did not answer", 503)
return dict(self.issued.get(sales_order_id, {}))
def cancel_drafts(self, source_ids):
for s in source_ids:
spec_check("ar", "POST", f"/invoices/{s}/cancel", {})
self.drafts.pop(s, None)
self.cancelled.append(s)
return {}
def issued_sources(self, invoice_ids):
spec_check("ar", "GET", "/invoice-totals", {"invoice_ids": ",".join(invoice_ids)})
if self.down:
raise Missing("invoicing", "invoicing did not answer", 503)
by_invoice = {v: k for k, v in self.invoice_ids.items()}
return {i: by_invoice.get(i) in self.issued_ids for i in invoice_ids}
def add_billed(self, document_id, bp_id, issue_date, lines, kind="invoice"):
"""Invoicing issued a document: each line (product_id, category_code, net_amount, tax_category) is readable from /billed-lines."""
for n, ln in enumerate(lines, 1):
self.billed_rows.append({"document_id": document_id, "kind": kind, "line_no": n, "issue_date": issue_date, "product_id": ln.get("product_id", "COLA-075"),
"category_code": ln.get("category_code", "drinks"), "net_amount": ln["net_amount"], "tax_category": ln.get("tax_category", "standard"),
"sales_order_id": ln.get("sales_order_id", ""), "so_line_no": ln.get("so_line_no", 0), "_bp": int(bp_id)})
def billed_lines(self, *, bp_id, issued_from, issued_to):
spec_check("ar", "GET", "/billed-lines", {"bp_id": str(bp_id), "issued_from": issued_from.isoformat(), "issued_to": issued_to.isoformat()})
if self.down:
raise Missing("invoicing", "invoicing did not answer", 503)
return [{k: v for k, v in x.items() if k != "_bp"} for x in self.billed_rows
if x["_bp"] == int(bp_id) and issued_from.isoformat() <= x["issue_date"] <= issued_to.isoformat()]
def credit_note(self, *, source_id, bp_id, currency, reason, lines, account_purpose="", invoice_id=""):
spec_check("ar", "POST", "/credit-notes", credit_body(source_id=source_id, bp_id=bp_id, currency=currency, reason=reason, lines=lines,
account_purpose=account_purpose, invoice_id=invoice_id))
if self.down:
raise Missing("invoicing", "invoicing did not answer", 503)
if getattr(self, "credit_down", False):
raise Missing("invoicing", "invoicing did not answer the credit note", 503)
self.calls_credit = getattr(self, "calls_credit", []) + [source_id]
if source_id not in self.credits:
self.credits[source_id] = {"credit_id": f"cn:{len(self.credits) + 1}", "status": "draft", "lines": lines, "bp_id": bp_id,
"account_purpose": account_purpose, "invoice_id": invoice_id, "reason": reason}
return {"credit_id": self.credits[source_id]["credit_id"], "status": "draft", "lines": lines, "tax": {}}
def issue(self, source_id):
"""Invoicing issues the draft: its lines now count as invoiced against the order lines."""
d = self.drafts[source_id]
self.issued_ids.add(source_id)
order_id = source_id.split(":")[1] if source_id.startswith("so:") else source_id
order_id = source_id[3:source_id.rindex(":")]
for ln in d["lines"]:
if ln.get("line_no"):
self.issued.setdefault(order_id, {})
self.issued[order_id][ln["line_no"]] = self.issued[order_id].get(ln["line_no"], D(0)) + D(ln["quantity"])
class FakeGl(GlPort):
"""The ledger's journal intake as its use case answers it: idempotent on journal_id (replay true), a different body for the same id is
a 409 journal_conflict, an unbalanced one 422 balanced, a date in a closed period 409 period_open."""
def __init__(self):
self.journals, self.closed_before, self.down, self.calls = {}, None, False, []
def post_journal(self, *, journal_id, source_document, event, company_code, journal_date, currency, memo, lines):
body = {"journal_id": journal_id, "source_module": "sales-order", "source_document": source_document, "event": event,
"journal_date": journal_date.isoformat(), "currency": currency, "memo": memo, "lines": lines, "company_code": company_code}
spec_check("gl", "POST", "/journal-entry/journals", body)
self.calls.append(journal_id)
if self.down:
raise Missing("the ledger", "the ledger did not answer", 503)
if sum(D(str(x.get("debit", 0))) for x in lines) != sum(D(str(x.get("credit", 0))) for x in lines):
raise Missing("the ledger", "the journal does not balance", 422)
if self.closed_before and journal_date < self.closed_before:
raise Missing("the ledger", f"period closed: {journal_date} is in a closed period", 409)
if journal_id in self.journals:
if self.journals[journal_id] != body:
raise Missing("the ledger", "journal_conflict: a different journal has this id", 409)
return {"journal_id": journal_id, "entry_no": len(self.journals), "status": "posted", "replay": True, "lines": len(lines)}
self.journals[journal_id] = body
return {"journal_id": journal_id, "entry_no": len(self.journals), "period_id": journal_date.isoformat()[:7], "status": "posted", "replay": False,
"lines": len(lines)}
def balance(self, purpose):
"""Net credit on a purpose across the journals (accrued rebates: credits less debits)."""
return sum(D(str(x.get("credit", 0))) - D(str(x.get("debit", 0))) for j in self.journals.values() for x in j["lines"] if x.get("purpose") == purpose)
class FakePartners(PartnersPort):
"""business-partner's answer: {bp_id, status, roles, can_transact, version}. 7311 Acme Ltd, 7313 York depot (ship_to), 7400 Group Treasury (payer);
7312 cannot transact (a customer on hold); 7500 is a supplier only."""
def __init__(self):
self.customers = {7311: True, 7312: False, 7313: True, 7400: True, 7500: True, 8420: True}
self.roles = {7311: ["customer"], 7312: ["customer"], 7313: ["customer", "ship_to"], 7400: ["customer"], 7500: ["supplier"], 8420: ["customer"]}
self.countries: dict = {}
def partner(self, caller, bp_id):
spec_check("business-partner", "GET", f"/partners/{int(bp_id)}")
if int(bp_id) not in self.customers:
raise Missing("partner", "no such partner", 404)
# the real GET /partners/{bp_id}: ids as strings, and the country a ship-to's place of supply reads (v7)
return {"bp_id": str(int(bp_id)), "bp_key": f"BP{int(bp_id)}", "legal_name": f"Partner {int(bp_id)}",
"country": self.countries.get(int(bp_id), "GB"), "tax_id": None, "kind": "organisation",
"status": "active" if self.customers[int(bp_id)] else "inactive", "roles": self.roles.get(int(bp_id), ["customer"]),
"can_transact": self.customers[int(bp_id)], "version": 1}
customer = partner # one route, GET /partners/{bp_id}: the same answer whichever question the order asks of it
class FakeInventory(InventoryPort):
"""inventory's reservation routes as its use case answers them: keyed (order, revision), a repeat changes nothing, a service line
comes back `ignored`, stock short of the ask comes back short_base (never an error)."""
SERVICES = {"SETUP", "HOST", "FEE", "PROJ-INSTALL"}
def __init__(self):
self.stock = {"COLA-075": D("1000"), "COLA-150": D("1000"), "WIDGET": D("1000"), "JUICE-075": D("1000"), "SCREW-M4": D("100000"),
"DRILL-500": D("50"), "PRINTER-X": D("0")}
self.reserved, self.calls, self.released, self.down = {}, [], [], False
self.supply, self.pick_pack, self.atp_calls, self.atp_down, self.reserve_bodies = {}, 1, [], False, []
self.consignment = {} # (bp_id, product_id) -> quantity_base held at the customer, ours
self.expected, self.cancelled_returns = {}, []
def atp(self, *, product_id, warehouse_id, quantity, unit_code, requested_on, min_shelf_life_days=0, exclude_sales_order_id=""):
"""inventory's GET /atp as its spec answers it: derived on read, nothing reserved. `supply[product]` is [(on, quantity_base, expires or
None)]: a lot that expires before its date plus the minimum shelf life is not counted. Unset: today's stock, all of it."""
q = {"warehouse_id": warehouse_id, "quantity": str(quantity), "unit_code": unit_code, "requested_on": requested_on,
"min_shelf_life_days": int(min_shelf_life_days or 0)}
if exclude_sales_order_id:
q["exclude_sales_order_id"] = exclude_sales_order_id
spec_check("inventory", "GET", f"/atp/{product_id}", q)
self.atp_calls.append({"product_id": product_id, **q})
if self.atp_down:
raise Missing("inventory", "inventory did not answer", 503)
if product_id in self.SERVICES:
raise Missing("inventory", f"{product_id} is not a stocked product", 422)
from erp_platform import work
now = work.now().date()
rows = self.supply.get(product_id) or [(now, self.stock.get(product_id, D(0)), None)]
keep = [(on, qty) for on, qty, exp in rows if exp is None or exp >= on + timedelta(days=int(min_shelf_life_days or 0))]
return {"product_id": product_id, "base_unit_code": "C62", "available_now": str(sum((qty for on, qty in keep if on <= now), D(0))),
"confirmations": [{"available_on": on.isoformat(), "quantity_base": str(qty)} for on, qty in sorted(keep)],
"shortfall_base": "0", "sources": [{"kind": "on_hand", "ref": "", "on": on.isoformat(), "quantity_base": str(qty)} for on, qty in sorted(keep)],
"pick_pack_days": self.pick_pack}
def reserve(self, *, sales_order_id, revision, lines, schedule_version=0, priority=""):
body = {"revision": int(revision), "lines": lines}
if schedule_version or priority == "rush":
body |= {"priority": priority or "normal"}
if schedule_version:
body |= {"schedule_version": int(schedule_version)}
spec_check("inventory", "POST", f"/reservations/sales-orders/{sales_order_id}", body)
self.reserve_bodies.append((sales_order_id, body))
self.calls.append(("reserve", sales_order_id, int(revision), [dict(x) for x in lines]))
if self.down:
raise Missing("inventory", "inventory did not answer", 503)
out = []
for ln in lines:
if ln["product_id"] in self.SERVICES:
out.append({"line_no": ln["line_no"], "ignored": "a service line is not stocked: nothing to reserve", "reserved_base": "0",
"short_base": "0", "base_unit_code": "HUR", "readiness": []})
continue
base = D(ln["quantity"]) * ln["unit_numerator"] / ln["unit_denominator"]
held = min(base, self.stock.get(ln["product_id"], D(0)))
self.reserved.setdefault((sales_order_id, int(revision)), {})[ln["line_no"]] = held
out.append({"line_no": ln["line_no"], "reserved_base": str(held), "short_base": str(base - held), "base_unit_code": "C62",
"readiness": [{"gate": "stock_reserved", "met": base == held, "reason": "", "fix": ""}]})
return {"sales_order_id": sales_order_id, "revision": int(revision), "lines": out}
def release(self, *, sales_order_id, reason):
spec_check("inventory", "POST", f"/reservations/sales-orders/{sales_order_id}/release", {"reason": reason})
self.calls.append(("release", sales_order_id, reason))
if self.down:
raise Missing("inventory", "inventory did not answer", 503)
self.released.append(sales_order_id)
return {"sales_order_id": sales_order_id, "released_base_by_line": []}
def expect_return(self, *, rma_ref, bp_id, warehouse_id, lines):
spec_check("inventory", "POST", "/expected-returns", expected_return_body(rma_ref=rma_ref, bp_id=bp_id, warehouse_id=warehouse_id, lines=lines))
if self.down:
raise Missing("inventory", "inventory did not answer", 503)
self.calls.append(("expect_return", rma_ref))
self.expected.setdefault(rma_ref, {"bp_id": bp_id, "warehouse_id": warehouse_id, "lines": lines})
return {"rma_ref": rma_ref, "expected": [{"line_no": x["line_no"], "quantity_base": str(D(x["quantity"]) * x["unit_numerator"] / x["unit_denominator"])} for x in lines]}
def cancel_expected_return(self, *, rma_ref):
self.cancelled_returns.append(rma_ref) # the cancel route is named only in INV-14's rule text, not listed as a seam: not contract-checked
return {"rma_ref": rma_ref, "status": "cancelled"}
def consignment_stock(self, *, bp_id, product_id):
spec_check("inventory", "GET", f"/consignment-stock/customers/{int(bp_id)}", {"product_id": product_id})
if self.down:
raise Missing("inventory", "inventory did not answer", 503)
held = self.consignment.get((int(bp_id), product_id))
return {"bp_id": int(bp_id), "products": [] if held is None else [{"product_id": product_id, "base_unit_code": "C62", "on_hand_base": str(held),
"location_id": f"UK1:CONSIGNMENT:{int(bp_id)}"}]}
class FakePurchasing(PurchasingPort):
"""purchasing's intake as PUR-65 answers it: idempotent on request_ref (201 then 200), 422 purchasable for a product it cannot buy; a request its
supplier confirmed cannot be cancelled until purchasing does it (409 with its reason)."""
NOT_PURCHASABLE = {"SETUP"}
def __init__(self):
self.requests, self.down, self.confirmed_with_supplier, self.cancelled = {}, False, {}, []
def purchase_request(self, *, request_ref, product_id, quantity, unit_code, needed_by, ship_to_bp_id, ship_to_address, sales_order_id, so_line_no):
spec_check("purchasing", "POST", "/purchase-requests/from-sales-order",
{"request_ref": request_ref, "product_id": product_id, "quantity": quantity, "unit_code": unit_code, "needed_by": needed_by,
"ship_to_bp_id": int(ship_to_bp_id), "ship_to_address": ship_to_address, "sales_order_id": sales_order_id, "so_line_no": int(so_line_no)})
if self.down:
raise Missing("purchasing", "purchasing did not answer", 503)
if product_id in self.NOT_PURCHASABLE:
raise Missing("purchasing", f"{product_id} is not purchasable", 422)
new = request_ref not in self.requests
self.requests.setdefault(request_ref, {"request_id": f"PR-{len(self.requests) + 1:04d}", "product_id": product_id, "quantity": quantity,
"ship_to_bp_id": ship_to_bp_id, "ship_to_address": ship_to_address, "needed_by": needed_by,
"sales_order_id": sales_order_id})
return {"request_ref": request_ref, "request_id": self.requests[request_ref]["request_id"], "status": "awaiting_approval", "created": new}
def cancel_purchase_request(self, *, request_id, reason):
# v8 (PUR-67): an ordered request is refused (gate not_cancellable) until the buyer cancels its order; the answer names the order and supplier
if request_id in self.confirmed_with_supplier:
raise Missing("purchasing", self.confirmed_with_supplier[request_id], 409)
self.cancelled.append(request_id)
return {"request_id": request_id, "status": "cancelled"}
class FakeMaps(MapsPort):
"""The earlier bricks' Odoo id maps: partner -> bp_id, product -> product_id (+ policy), uom -> unit, tax, payment term."""
def __init__(self):
self.partners = {101: 7311, 102: 7313}
self.products = {201: {"product_id": "WIDGET", "invoicing_policy": "delivered"}, 202: {"product_id": "FEE", "invoicing_policy": "ordered"},
203: {"product_id": "COLA-075", "invoicing_policy": "ordered"}}
self.units = {1: {"unit_code": "C62", "numerator": 1, "denominator": 1}, 2: {"unit_code": "PK24", "numerator": 24, "denominator": 1}}
self.taxes = {301: "S"}
self.terms = {401: "NET30"}
def partner(self, odoo_id):
return self.partners.get(odoo_id)
def product(self, odoo_id):
return self.products.get(odoo_id)
def unit(self, odoo_uom_id, product_id):
return self.units.get(odoo_uom_id)
def tax(self, odoo_id):
return self.taxes.get(odoo_id)
def payment_term(self, odoo_id):
return self.terms.get(odoo_id)
class FakeOdoo(OdooPort):
"""A small Odoo: open, done, locked and cancelled orders, a down payment, a line in a 24-pack, section and note lines,
two quotations. Names run S00001.. in id order; order 2020s dates decide the history window."""
db = "odoo17"
def __init__(self):
self._orders, self._lines = [], []
n = [0]
def order(state, date_order="2026-09-10 10:00:00", locked=False, partner=101, user=7, net="1000.00", tax="200.00", **kw):
n[0] += 1
o = {"id": n[0], "name": f"S{n[0]:05d}", "company_id": 1, "state": state, "locked": locked, "partner_id": partner, "client_order_ref": "",
"date_order": date_order, "user_id": user, "currency": "GBP", "pricelist_name": "Retail GBP", "payment_term_id": 401, "incoterm": "",
"amount_untaxed": net, "amount_tax": tax, "amount_total": str(Decimal(net) + Decimal(tax)), **kw}
self._orders.append(o)
return o
def line(o, product=202, qty="1", price="1000.00", disc="0", delivered="0", invoiced="0", to_invoice="0", uom=1, seq=10, **kw):
net = (Decimal(qty) * Decimal(price) * (1 - Decimal(disc) / 100)).quantize(Decimal("0.01"))
ln = {"id": 1000 + len(self._lines) + 1, "order_id": o["id"], "sequence": seq, "display_type": False, "is_downpayment": False,
"product_id": product, "name": f"line of {o['name']}", "product_uom_qty": qty, "product_uom": uom, "price_unit": price,
"discount": disc, "tax_ids": [301], "price_subtotal": str(net), "price_tax": str((net * Decimal("0.2")).quantize(Decimal("0.01"))),
"qty_delivered": delivered, "qty_invoiced": invoiced, "qty_to_invoice": to_invoice, **kw}
self._lines.append(ln)
return ln
self.quote_draft, self.quote_sent = order("draft"), order("sent") # S00001, S00002: brick-quote's
a = order("sale", partner=101, net="120.00", tax="24.00", client_order_ref="PO-77") # S00003 open, part delivered and invoiced
line(a, product=201, qty="12", price="10.00", delivered="5", invoiced="5", to_invoice="0", seq=10)
line(a, product=None, qty="0", price="0", seq=5, display_type="line_section", name="Hardware")
b = order("sale", partner=102, net="15000.00", tax="2000.00", user=8) # S00004 open, down payment, unknown salesperson
line(b, product=202, qty="10", price="1000.00", seq=10, to_invoice="10")
line(b, product=None, qty="1", price="5000.00", seq=20, is_downpayment=True, name="Down payment", invoiced_amount="5000.00",
deducted_amount="0.00", invoice_move_ids=[901], qty_invoiced="1", price_tax="0.00")
c = order("sale", net="72.00", tax="14.40") # S00005 open, 3 x 24-pack, 5% off
line(c, product=203, qty="3", price="25.2631", disc="5", uom=2, to_invoice="3")
d = order("sale", date_order="2024-03-01 09:00:00", net="1000.00", tax="200.00") # S00006 finished, old: outside a 2026 window
line(d, delivered="1", invoiced="1", to_invoice="0")
e = order("sale", date_order="2026-02-01 09:00:00", net="1000.00", tax="200.00") # S00007 finished (sale, all delivered and invoiced)
line(e, delivered="1", invoiced="1")
line(e, product=None, qty="0", price="0", seq=20, display_type="line_note", name="Thanks!")
f = order("done", locked=True, date_order="2026-05-01 09:00:00", net="2000.00", tax="400.00") # S00008 done and locked
line(f, qty="2", delivered="2", invoiced="2")
g = order("cancel", date_order="2026-06-01 09:00:00", net="1000.00", tax="200.00") # S00009 cancelled
line(g)
self.opens = [a, b, c]
self.history = [d, e, f, g]
def orders(self):
return [dict(o) for o in self._orders]
def lines(self, order_ids):
return [dict(x) for x in self._lines if x["order_id"] in order_ids]