dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
tests/fakes.py
"""In-memory neighbours, each answering EXACTLY what its SPEC route answers (shape and status codes): a fake that disagrees with the real callee is
the bug that hid 31 broken seams in wave 1. business-partner, product, tax, gl, inventory, payments, ar, fixed-assets, the object store, and the
Odoo reader and id maps of the migration."""
from decimal import Decimal
from fractions import Fraction
from brick_purchasing.infrastructure.seams import (
ArPort,
BpPort,
FixedAssetsPort,
GlPort,
InventoryPort,
MapsPort,
Missing,
ObjectsPort,
OdooPort,
PaymentsPort,
ProductPort,
TaxPort,
)
D = Decimal
def gate(g, reason="", fix=""):
return {"gate": g, "met": False, "reason": reason, "fix": fix}
from erp_platform.seam import derived_id # noqa: E402
class FakeBp(BpPort):
def __init__(self):
self.promoted = {}
def promote(self, caller, *, bp_key, legal_name, country="", tax_id="", domain=""):
bp = self.promoted.setdefault(bp_key, 7000 + len(self.promoted))
return {"bp_id": bp, "bp_key": bp_key, "legal_name": legal_name, "roles": ["supplier"], "_status": 201}
def get(self, caller, bp_id):
if int(bp_id) >= 9_000_000:
raise Missing(f"partner {bp_id}", "no such partner", 404)
return {"bp_id": bp_id, "legal_name": f"Supplier {bp_id}", "country": "GB", "tax_id": "GB123456789", "roles": ["supplier"], "status": "active"}
class FakeProduct(ProductPort):
"""COLA-075 (bottles C62; PK24 = 24 bottles; a test sets rule=("PK24", 80, 80) for the pallet rule), STEEL-BAR (a bar C62; 1 KGM = 2/5 bar), LAPTOP (each), CONSULT
(hours, a service), WIDGET (C62)."""
P = None
_P = {
"COLA-075": dict(base="C62", kind="goods", tax="standard", units={"PK24": (24, 1)}, purchase="PK24", rule=None),
"STEEL-BAR": dict(base="C62", kind="goods", tax="standard", units={"KGM": (2, 5)}, purchase="C62", rule=None),
"LAPTOP": dict(base="C62", kind="goods", tax="standard", units={}, purchase="C62", rule=None),
"WIDGET": dict(base="C62", kind="goods", tax="standard", units={"BX10": (10, 1)}, purchase="C62", rule=None),
"CONSULT": dict(base="HUR", kind="service", tax="standard", units={}, purchase="HUR", rule=None),
"ZERO-1": dict(base="C62", kind="goods", tax="zero", units={}, purchase="C62", rule=None),
}
def __init__(self):
import copy
self.P = copy.deepcopy(self._P)
self.used = []
def _p(self, pid):
p = self.P.get(pid)
if p is None:
raise Missing(f"product {pid}", f"no product {pid}", 404)
return p
def get(self, caller, product_id, on=None):
p = self._p(product_id)
units = [{"unit_code": p["base"], "numerator": 1, "denominator": 1}] + [{"unit_code": u, "numerator": n, "denominator": d}
for u, (n, d) in p["units"].items()]
return {"product_id": product_id, "name": product_id, "description": f"{product_id} description", "kind": p["kind"], "product_type_code": "TRADING" if p["kind"] == "goods" else "SERVICE",
"stocked_default": p["kind"] == "goods", "valuation_class": "", "tax_category": p["tax"], "commodity_code": "", "country_of_origin": "",
"base_unit_code": p["base"], "roles": {"purchase": {"unit_code": p["purchase"]}, "sales": {"unit_code": p["base"]}}, "units": units,
"lifecycle": "active", "version": 1}
def _factor(self, p, unit):
if unit == p["base"]:
return Fraction(1)
if unit in p["units"]:
n, d = p["units"][unit]
return Fraction(n, d)
return None
def convert(self, caller, product_id, quantity, from_unit, to_unit):
p = self._p(product_id)
a, b = self._factor(p, from_unit), self._factor(p, to_unit)
if a is None or b is None:
raise Missing(f"product {product_id}", f"{product_id} is not counted in {from_unit if a is None else to_unit}", 422)
q = Fraction(D(quantity)) * a / b
return {"product_id": product_id, "from_unit": from_unit, "to_unit": to_unit, "quantity": quantity, "exact": f"{q.numerator}/{q.denominator}",
"converted": str(D(q.numerator) / D(q.denominator)), "in_base_unit": str(D(quantity) * D(a.numerator) / D(a.denominator)),
"base_unit_code": p["base"]}
def unit_used(self, product_id, unit_code, document_ref):
self._p(product_id)
self.used.append((product_id, unit_code, document_ref))
return {"product_id": product_id, "unit_code": unit_code, "used": True, "fixed_by": "its own row"}
def rounding(self, caller, product_id, quantity, unit_code, on, role="purchase"):
if "." in str(quantity) and len(str(quantity).split(".")[1]) > 4: # the real query refuses it (shared.yaml quantity: 4 places)
raise Missing("product", "a quantity has at most 4 decimal places", 422)
p = self._p(product_id)
rule = p["rule"]
q = D(quantity)
if rule is None or unit_code != rule[0]:
return {"product_id": product_id, "role": role, "rule": None, "quantity_in_rule_unit": quantity, "met": True, "next_down": None,
"next_up": None, "reason": ""}
unit, mn, mult = rule
met = q >= mn and q % mult == 0
down, up = (q // mult) * mult, (q // mult + 1) * mult
return {"product_id": product_id, "role": role, "rule": {"unit_code": unit, "min_quantity": str(mn), "rounding_multiple": str(mult), "valid_from": "2026-01-01"},
"quantity_in_rule_unit": quantity, "met": met, "next_down": None if met else str(down), "next_up": None if met else str(up),
"reason": "" if met else f"{product_id} is bought in multiples of {mult} {unit} (a full pallet): {q.normalize():f} is not one"}
class FakeTax(TaxPort):
"""Purchase direction: standard 20%, zero 0%. Records the documents; check_contents fails while `missing_vat_number` is set."""
RATES = {"standard": ("S20", D(20)), "zero": ("Z0", D(0)), "exempt": ("E0", D(0)), "reduced": ("R5", D(5))}
def __init__(self):
self.docs, self.frozen, self.missing_vat_number, self.down = {}, set(), False, False
self.cis = {} # bp_id -> {status, rate_pct, verification_number}
self.deductions, self.reversed, self.imports = {}, [], {}
def compute(self, body):
if self.down:
raise Missing("tax", "tax did not answer", 503)
lines, net, vat, recl = [], D(0), D(0), D(0)
for ln in body["lines"]:
code, rate = self.RATES.get(ln.get("category_code") or "standard", ("S20", D(20)))
n = D(ln["amount"])
v = (n * rate / 100).quantize(D("0.01"))
lines.append({"line_no": ln["line_no"], "tax_code": code, "rate_pct": str(rate), "net": str(n), "vat": str(v), "gross": str(n + v),
"vat_reclaimable": str(v), "vat_non_reclaimable": "0.00"})
net, vat, recl = net + n, vat + v, recl + v
self.docs[body["document_ref"]] = body
return {"document_ref": body["document_ref"], "status": "computed", "lines": lines, "summary": [],
"totals": {"net": str(net), "vat": str(vat), "vat_due": str(vat), "gross": str(net + vat), "method": "line", "vat_reclaimable": str(recl),
"vat_non_reclaimable": "0.00"}, "gates": [], "version": 1, "_status": 200}
def check_contents(self, document_ref, contents):
self.contents = contents
gates = [gate("supplier_vat_number", "the supplier's VAT number is missing", "ask the supplier for a valid VAT invoice")] if self.missing_vat_number else []
return {"document_ref": document_ref, "status": "computed", "ready": not gates, "title": "VAT invoice", "gates": gates}
def freeze(self, document_ref, contents=None):
self.frozen.add(document_ref)
self.frozen_contents = contents
return {"document_ref": document_ref, "status": "frozen", "_status": 200}
def party(self, bp_id):
return {"bp_id": bp_id, "current": {"treatment_code": "standard", "vat_number": "GB123", "valid_from": "2026-01-01"}, "rows": []}
def cis_subcontractor(self, bp_id, on):
c = self.cis.get(int(bp_id))
if c is None:
raise Missing("tax", "not a CIS subcontractor", 404)
return {"bp_id": bp_id, "utr": "1234567890", "verification_number": c.get("verification_number", "V1"), "status": c["status"], "rate_pct": c["rate_pct"],
"verified_on": "2026-01-01", "readiness": c.get("readiness", [{"gate": "cis_verified", "met": True, "reason": "", "fix": ""}])}
def cis_deduct(self, body):
c = self.cis[int(body["bp_id"])]
self.deductions[body["payment_ref"]] = body
return {"payment_ref": body["payment_ref"], "bp_id": body["bp_id"], "tax_month": "2026-11", "status": c["status"], "rate_pct": c["rate_pct"],
"labour": body["gross_ex_vat"], "deduction": "0", "gates": [], "_status": 200}
def cis_reverse(self, payment_ref, body):
self.reversed.append(payment_ref)
return {"payment_ref": payment_ref, "bp_id": body["bp_id"], "status": "reversed", "tax_month": "2026-11", "_status": 200}
def record_import(self, body):
self.imports[body["mrn"]] = body
return {"mrn": body["mrn"], "tax_code": "PVA", "box_vat_output": "0", "box_vat_input": "0", "box_value": "0", "_status": 201}
class FakeGl(GlPort):
"""Records the journals (a retry with the same journal_id answers replay); a closed period is a 409 {gate: period_open}."""
def __init__(self):
self.journals, self.closed, self.refuse, self.down = {}, set(), None, False
self.extra_cost_centres = {} # a test adds {"X": "closed"}
def post_journal(self, body):
if self.down:
raise Missing("the ledger", "did not answer", 503)
if self.refuse:
raise Missing("the ledger", self.refuse["reason"], self.refuse["status"], gates=[gate(self.refuse["gate"], self.refuse["reason"], "fix")])
if body.get("source_module") == "purchasing": # gl's purchasing seam: event approve | pay | reverse; journal_id = derived_id(source_document, event)
if body["event"] not in ("approve", "pay", "reverse"):
raise Missing("the ledger", f"event {body['event']} is not accepted from purchasing", 422, gates=[gate("event", body["event"], "use approve | pay | reverse")])
if str(body["journal_id"]) != str(derived_id(body["source_document"], body["event"])):
raise Missing("the ledger", "journal_id is not derived_id(source_document, event)", 422, gates=[gate("journal_id", body["journal_id"], "derive it")])
month = body["journal_date"][:7]
if month in self.closed:
raise Missing("the ledger", f"period {month} is closed", 409, gates=[gate("period_open", f"period {month} is closed", "reopen it")])
debit = sum(D(x.get("debit", "0")) for x in body["lines"])
credit = sum(D(x.get("credit", "0")) for x in body["lines"])
if debit != credit:
raise Missing("the ledger", f"debits {debit} do not equal credits {credit}", 422, gates=[gate("balanced", f"{debit} vs {credit}", "balance it")])
replay = body["journal_id"] in self.journals
self.journals.setdefault(body["journal_id"], body)
return {"journal_id": body["journal_id"], "entry_no": "J1", "period_id": month, "status": "posted", "replay": replay, "lines": len(body["lines"]),
"_status": 200 if replay else 201}
def period_open(self, on, company_code=None):
month = on.isoformat()[:7]
if month in self.closed:
return {"date": on.isoformat(), "company_code": company_code or "UK01", "period_id": month, "open": False, "status": "closed",
"reason": f"period {month} is closed", "fix": "ask the accountant to reopen it", "_status": 200}
return {"date": on.isoformat(), "company_code": company_code or "UK01", "period_id": month, "open": True, "status": "open", "reason": "", "fix": "", "_status": 200}
COMPANY = {"company_code": "UK01", "name": "Brightside Ltd", "country": "GB", "functional_currency": "GBP", "chart_id": "UK", "bp_id": 1,
"financial_year_end_month": 12, "ledgers": ["0L"], "_status": 200}
def cost_centre(self, cost_centre):
status = self.extra_cost_centres.get(cost_centre, "open") # every cost centre is open unless a test says closed or unknown
if status == "unknown":
raise Missing("the ledger", f"no cost centre {cost_centre}", 404)
return {"cost_centre": cost_centre, "name": cost_centre, "status": status, "company_code": "UK01", "_status": 200}
def company(self, company_code):
if company_code != "UK01":
raise Missing("the ledger", f"no company {company_code}", 404)
return dict(self.COMPANY)
def companies(self):
return [dict(self.COMPANY)]
def posted(self, document, event=None):
return [j for j in self.journals.values() if (j["source_document"] == document or j["source_document"].startswith(document + ":")) and (event is None or j["event"] == event)]
class FakeInventory(InventoryPort):
"""Idempotent on move_ref; a location 'BAD' is not an internal one (409 {gate: location}); 'EMPTY' has no stock to return (409 {gate: on_hand});
a component short at the subcontractor is 409 {gate: components_at_subcontractor}."""
def __init__(self):
self.moves, self.returns, self.provisions, self.adjustments = {}, {}, {}, {}
self.short_components = set()
self.down = False
def receive(self, body):
if self.down:
raise Missing("inventory", "did not answer", 503)
if body["move_ref"] in self.moves:
return {**self.moves[body["move_ref"]]["answer"], "_status": 200}
if body["location_id"] == "BAD":
raise Missing("inventory", "BAD is not an internal location", 409, gates=[gate("location", "BAD is not an internal location", "pick an internal one")])
for c in body.get("consume_provided") or []:
if c["product_id"] in self.short_components:
raise Missing("inventory", f"{c['product_id']} is short at the subcontractor", 409,
gates=[gate("components_at_subcontractor", f"{c['product_id']} is short at the subcontractor", "provide it first")])
n = len(self.moves) + 1
answer = {"move_id": f"M{n}", "quantity_base": body["quantity"]}
if body.get("consume_provided"):
answer["consumed"] = [{"move_id": f"M{n}c{i}", "product_id": c["product_id"], "quantity_base": c["quantity"]} for i, c in enumerate(body["consume_provided"])]
self.moves[body["move_ref"]] = {"body": body, "answer": answer}
return {**answer, "_status": 201}
def return_to_supplier(self, body):
if body["move_ref"] in self.returns:
return {**self.returns[body["move_ref"]]["answer"], "_status": 200}
if body["location_id"] == "EMPTY":
raise Missing("inventory", "the stock is no longer at the location", 409, gates=[gate("on_hand", "the stock is no longer at the location", "check the count")])
answer = {"move_id": f"R{len(self.returns) + 1}", "quantity_base": body["quantity"]}
self.returns[body["move_ref"]] = {"body": body, "answer": answer}
return {**answer, "_status": 201}
def locations(self, usage=None, warehouse_id=None):
return {"locations": [{"location_id": "WH/STOCK", "warehouse_id": "WH", "name": "Stock", "usage": "internal", "default": True}], "_status": 200}
def subcontract_provision(self, body):
if body["move_ref"] in self.provisions:
return {**self.provisions[body["move_ref"]]["answer"], "_status": 200}
for c in body["components"]:
if c["product_id"] in self.short_components:
raise Missing("inventory", f"{c['product_id']} is short", 409, gates=[gate("on_hand", f"{c['product_id']} is short", "receive it first")])
answer = {"move_id": f"P{len(self.provisions) + 1}", "components": [{"product_id": c["product_id"], "quantity_base": c["quantity"]} for c in body["components"]]}
self.provisions[body["move_ref"]] = {"body": body, "answer": answer}
return {**answer, "_status": 201}
def valuation_adjustment(self, body):
replay = body["adjustment_ref"] in self.adjustments
self.adjustments.setdefault(body["adjustment_ref"], body)
return {"adjustment_ref": body["adjustment_ref"], "on_hand_amount": "0", "price_difference_amount": "0", "_status": 200 if replay else 201}
class FakePayments(PaymentsPort):
def __init__(self):
self.accounts = [{"bank_account_id": "BARC-01", "name": "Barclays GBP", "currency": "GBP", "iban_or_sort_code_masked": "****1234", "gl_account_id": "1200"},
{"bank_account_id": "EUR-01", "name": "Euro account", "currency": "EUR", "iban_or_sort_code_masked": "****9999", "gl_account_id": "1210"}]
self.reads = []
def bank_accounts(self):
return {"bank_accounts": self.accounts, "_status": 200}
# PAY-27: the unmasked details of OUR paying account. `fail` is the status payments answers instead (403 | 404 | 503).
details = {"BARC-01": {"bank_account_id": "BARC-01", "company_code": "UK01", "account_name": "Barclays GBP", "sort_code": "601613",
"account_number": "87654321", "iban": "GB12BARC20000012345678", "bic": "BARCGB22", "currency": "GBP"},
"EUR-01": {"bank_account_id": "EUR-01", "company_code": "UK01", "account_name": "Euro account", "sort_code": "",
"account_number": "", "iban": "DE89370400440532013000", "bic": "COBADEFF", "currency": "EUR"}}
fail = None
def account_details(self, bank_account_id, purpose):
self.reads.append((bank_account_id, purpose))
if self.fail:
raise Missing("payments", {403: "forbidden", 404: "unknown account", 503: "cannot log the read"}[self.fail], self.fail)
if bank_account_id not in self.details:
raise Missing("payments", "unknown account", 404)
return {**self.details[bank_account_id], "_status": 200}
class FakeAr(ArPort):
def __init__(self):
self.owed = {}
def customer_status(self, bp_id):
o = self.owed.get(int(bp_id))
return {"bp_id": bp_id, "overdue_amount": str(o or "0.00"), "overdue_by_currency": {"GBP": str(o or "0.00")}, "oldest_days_overdue": 12 if o else 0,
"on_credit_hold": False, "hold_reason": "", "known": o is not None, "note": "", "_status": 200}
class FakeFa(FixedAssetsPort):
def __init__(self):
self.assets = {"UK01-2026-VEH-0003-0": {"asset_id": "UK01-2026-VEH-0003-0", "company_code": "UK01", "asset_no": "0003", "sub_no": "0", "class_code": "VEH",
"description": "Van", "status": "active", "is_auc": False, "cost_centre": "SITE-A", "capitalisation_threshold": "500.00",
"readiness": []},
"UK02-2026-VEH-0001-0": {"asset_id": "UK02-2026-VEH-0001-0", "company_code": "UK02", "asset_no": "0001", "sub_no": "0", "class_code": "VEH",
"description": "Other co van", "status": "active", "is_auc": False, "cost_centre": "X", "capitalisation_threshold": "500.00",
"readiness": []},
"UK01-2026-VEH-0009-0": {"asset_id": "UK01-2026-VEH-0009-0", "company_code": "UK01", "asset_no": "0009", "sub_no": "0", "class_code": "VEH",
"description": "Sold van", "status": "retired", "is_auc": False, "cost_centre": "X", "capitalisation_threshold": "500.00",
"readiness": []}}
self.acquisitions, self.refuse = {}, None
def asset(self, asset_id):
if asset_id not in self.assets:
raise Missing("fixed-assets", f"no asset {asset_id}", 404)
return {**self.assets[asset_id], "_status": 200}
def classes(self, on):
return {"classes": [{"class_code": "VEH", "name": "Vehicles", "is_auc": False, "is_low_value": False, "capitalisation_threshold": "500.00"}], "_status": 200}
def acquisition(self, body):
if self.refuse:
raise Missing("fixed-assets", self.refuse, 409, gates=[gate("period_open", self.refuse, "reopen")])
replay = body["acquisition_ref"] in self.acquisitions
self.acquisitions.setdefault(body["acquisition_ref"], body)
return {"acquisition_ref": body["acquisition_ref"], "transaction_id": f"T{len(self.acquisitions)}", "asset_id": body.get("asset_id") or "NEW", "areas": [],
"replay": replay, "_status": 200 if replay else 201}
class FakeObjects(ObjectsPort):
def __init__(self):
self.store = {}
def put(self, key, body, content_type):
self.store.setdefault(key, body)
return key
def get(self, key):
return self.store.get(key)
class FakeOdoo(OdooPort):
"""Flat Odoo rows per model (module.yaml migration.odoo.reads); `rows[model]` is what the reader returns."""
def __init__(self):
self.db = "acme"
self.rows = {}
def read(self, model):
return [dict(r) for r in self.rows.get(model, [])]
class FakeMaps(MapsPort):
def __init__(self):
self.db = "acme"
self.partners, self.products, self.units, self.taxes = {}, {}, {}, {}
def partner(self, odoo_id):
return self.partners.get(int(odoo_id))
def product(self, odoo_id):
return self.products.get(int(odoo_id))
def unit(self, odoo_uom_id):
return self.units.get(int(odoo_uom_id))
def tax(self, odoo_id):
return self.taxes.get(int(odoo_id))