dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
tests/fakes.py
"""Stand-ins for the neighbours, each shaped from THAT brick's spec contract (module.yaml seams) and registered in process
(erp_platform.seam.register_local), so the real Seam code runs against them. Nothing here reads this brick's tables."""
from decimal import ROUND_HALF_UP, Decimal
from erp_platform import seam
from fastapi import FastAPI, Request, Response
RATES = {"OS": Decimal("0"), "S": Decimal("20"), "R": Decimal("5"), "Z": Decimal("0"), "E": Decimal("0"), "NV": Decimal("0")}
CATEGORY = {"outside_scope": "OS", "standard": "S", "reduced": "R", "zero": "Z", "exempt": "E", "": "S"}
def cents(d: Decimal) -> Decimal:
return d.quantize(Decimal("0.01"), ROUND_HALF_UP)
class Tax:
"""tax, as its use cases answer: compute_tax (POST /documents/compute, kept under document_ref), check_contents
(POST /documents/{ref}/check-contents), freeze_document_tax (POST /documents/{ref}/freeze), the Odoo map."""
def __init__(self):
self.missing_address = True # check_contents refuses contents with no customer address
self.calls = []
self.frozen = {} # document_ref -> the contents it was frozen with
self.migration_map = {} # odoo account.tax id -> tax code
self.down = False
self.filed = set() # tax points (ISO) inside a filed VAT return
self.treatment = {} # bp_id -> treatment_code (TAX-23); unknown = no row on file
self.tax_point_down = False
app = FastAPI()
self.app = app
@app.post("/documents/compute")
async def compute(request: Request, response: Response):
body = await request.json()
if self.down:
response.status_code = 503
return {"detail": "tax is down"}
if body.get("document_ref") in self.frozen:
response.status_code = 409
return {"detail": f"document {body['document_ref']} is issued: credit it instead"}
self.calls.append(body)
lines, summary = [], {}
for ln in body["lines"]:
code = ln["tax_code"] if ln.get("pinned_from") else CATEGORY.get(ln.get("category_code") or "", "S")
rate = RATES[code]
net = Decimal(ln["amount"])
tax = cents(net * rate / 100)
lines.append({"line_no": ln["line_no"], "tax_code": code, "rate_pct": str(rate), "charged_pct": str(rate),
"amount": str(net), "net": str(net), "vat": str(tax), "vat_due": str(tax), "refusal": None})
row = summary.setdefault(code, {"tax_code": code, "rate_pct": str(rate), "net": Decimal(0), "vat": Decimal(0)})
row["net"] += net
row["vat"] += tax
return {"document_ref": body.get("document_ref"), "complete": True, "lines": lines, "rounding_method": "per_line_nearest",
"gates": [], "summary": [{**r, "net": str(r["net"]), "vat": str(r["vat"])} for r in summary.values()]}
def gates(contents: dict) -> list:
out = [{"gate": "invoice_number", "met": bool(contents.get("invoice_number")), "reason": "a VAT invoice shows a "
"sequential number (VAT Notice 700 para 16.3.1)", "fix": "the invoice series on the invoice"}]
ok = not (self.missing_address and not contents.get("customer_address"))
out.append({"gate": "customer_address", "met": ok, "reason": "" if ok else "a VAT invoice shows the customer's address "
"(VAT Notice 700 para 16.3.1)", "fix": "the customer's address on the business partner"})
out.append({"gate": "contents_complete", "met": all(g["met"] for g in out), "reason": "", "fix": "the record that holds the item"})
return out
@app.post("/documents/{ref}/check-contents")
async def check(ref: str, request: Request):
body = await request.json()
g = gates(body.get("contents") or {})
return {"document_ref": ref, "status": "draft", "ready": all(x["met"] for x in g), "gates": g}
@app.post("/documents/{ref}/freeze")
async def freeze(ref: str, request: Request, response: Response):
contents = (await request.json()).get("contents") or {}
if ref not in self.frozen:
g = gates(contents)
if not all(x["met"] for x in g):
response.status_code = 409
return {"detail": "not frozen: " + ", ".join(x["gate"] for x in g if not x["met"])}
self.frozen[ref] = contents
return {"document_ref": ref, "status": "frozen"}
@app.get("/migration-map/{db}/{model}")
def mapped(db: str, model: str, ids: str = ""):
return [{"odoo_model": model, "odoo_id": i, "ours": self.migration_map[int(i)]}
for i in ids.split(",") if i and int(i) in self.migration_map]
@app.get("/vat-returns/tax-point-status")
def tax_point_status(date: str, response: Response):
"""TAX-33: {date, period_key | null, status, filed, gate: {gate: tax_point_open, met, reason, fix}}."""
if self.tax_point_down:
response.status_code = 503
return {"detail": "tax is down"}
filed = date in self.filed
return {"date": date, "period_key": "26AC" if filed else None, "status": "submitted" if filed else "open", "filed": filed,
"gate": {"gate": "tax_point_open", "met": not filed, "reason": "the tax point is in a filed VAT return" if filed else "",
"fix": "issue with a tax point of today" if filed else ""}}
@app.get("/parties/{bp_id}")
def party(bp_id: int):
code = self.treatment.get(bp_id)
return {"bp_id": bp_id, "current": {"treatment_code": code, "vat_number": "", "valid_from": "2020-01-01"} if code else None,
"rows": []}
class Payments:
"""payments, as its use cases answer: settlement (GET /settlement), allocate_from_source (POST /allocations/from-source),
credit_ok (GET /customers/{bp_id}/credit-ok), open_items' unallocated credit notes (GET /customers/{bp_id}/open-items).
Tests register an invoice's total (as payments would read it from ar's invoice-totals route) and receive money."""
def __init__(self):
self.totals = {} # invoice_id -> total
self.allocations = [] # (source_kind, source_id, invoice_id, amount)
self.credit_faces = {} # credit_id -> face value
self.receipts = [] # PAY-46 allocations: {source_kind, source_id, invoice_id, seq, amount, currency, allocated_on, received_on, reverses_seq}
self.hold = False
self.down = False
self.aged = {"0-30": [], "31-60": [], "61-90": [], "90+": []} # PAY-40 aging, set by the test
self.aged_out = set() # invoices written off / paid: they leave the aging
app = FastAPI()
self.app = app
@app.get("/settlement")
def settle(invoice_ids: str, response: Response):
if self.down:
response.status_code = 503
return {"detail": "down"}
out, unknown = [], []
for iid in invoice_ids.split(","):
if iid in self.totals:
paid = sum(a[3] for a in self.allocations if a[2] == iid)
out.append({"invoice_id": iid, "total": str(self.totals[iid]), "outstanding": str(self.totals[iid] - paid),
"state": "paid" if paid >= self.totals[iid] else ("part_paid" if paid else "open"), "age_days": 0})
else:
unknown.append(iid)
return {"as_at": "2026-11-20", "invoices": out, "unknown": unknown}
@app.post("/allocations/from-source")
async def allocate(request: Request, response: Response):
b = await request.json()
if self.down:
response.status_code = 503
return {"detail": "down"}
amount = Decimal(b["amount"])
key = (b["source_kind"], b["source_id"], b["invoice_id"])
if any((a[0], a[1], a[2]) == key for a in self.allocations):
return {"created": False, "allocation": {"source_kind": key[0], "source_id": key[1], "invoice_id": key[2]}}
paid = sum(a[3] for a in self.allocations if a[2] == b["invoice_id"])
if b["invoice_id"] in self.totals and amount > self.totals[b["invoice_id"]] - paid:
response.status_code = 422
return {"detail": f"invoice only has {self.totals[b['invoice_id']] - paid} outstanding"}
self.allocations.append((*key, amount))
return {"created": True, "allocation": {"source_kind": key[0], "source_id": key[1], "invoice_id": key[2],
"amount": str(amount)}}
@app.get("/allocations")
def allocations(invoice_ids: str, response: Response):
wanted = invoice_ids.split(",")
return {"allocations": [a for a in self.receipts if a["invoice_id"] in wanted]}
@app.get("/collections/aging")
def aging(as_at: str, response: Response, company_code: str = ""):
# PAY-40: the aging AS AT a date, in the business's buckets: {as_at, buckets: [{bucket, invoices: [{invoice_id, bp_id,
# outstanding, days_past_due, disputed}]}]}; tests set `aged` (bucket -> invoices) and an invoice leaves it when settled
return {"as_at": as_at, "buckets": [{"bucket": b, "invoices": [i for i in invs if i["invoice_id"] not in self.aged_out]}
for b, invs in self.aged.items()]}
@app.get("/customers/{bp_id}/credit-ok")
def status(bp_id: int):
return {"gate": "credit_ok", "met": not self.hold, "reason": "2 invoices over 60 days" if self.hold else "",
"fix": "collect the overdue invoices", "hold_mode": "block", "bp_id": bp_id}
@app.get("/customers/{bp_id}/open-items")
def open_items(bp_id: int):
credits = []
for cid, face in self.credit_faces.items():
left = face - sum(a[3] for a in self.allocations if a[1] == cid)
if left > 0:
credits.append({"kind": "credit_note", "id": cid, "number": cid, "currency": "GBP", "amount": str(-left)})
return {"bp_id": bp_id, "invoices": [], "credits": credits, "closing_balance": {}}
def receive(self, invoice_id, amount, ref="R1", received_on=None):
self.allocations.append(("receipt", ref, invoice_id, Decimal(amount)))
if received_on:
self.receipts.append({"source_kind": "receipt", "source_id": ref, "invoice_id": invoice_id, "seq": len(self.receipts) + 1,
"amount": str(amount), "currency": "GBP", "allocated_on": received_on, "received_on": received_on,
"reverses_seq": None})
class Ledger:
"""gl: POST /journal-entry/journals (balanced, idempotent on journal_id, refused in a closed period)."""
def __init__(self):
self.journals = {}
self.calls = [] # every journal_id posted, replays included
self.companies = {"UK01", "UK02"}
self.closed_before = None # journals dated on or before this ISO date are refused
app = FastAPI()
self.app = app
@app.post("/journal-entry/journals")
async def post(request: Request, response: Response):
b = await request.json()
debit = sum(Decimal(x.get("debit", "0")) for x in b["lines"])
credit = sum(Decimal(x.get("credit", "0")) for x in b["lines"])
if debit != credit:
response.status_code = 422
return {"detail": f"unbalanced journal: debit {debit} credit {credit}"}
if self.closed_before and b["journal_date"] <= self.closed_before:
response.status_code = 409
return {"detail": f"period {b['journal_date'][:7]} is closed"}
self.calls.append(b["journal_id"])
self.journals.setdefault(b["journal_id"], b)
return {"journal_id": b["journal_id"]}
@app.get("/companies/{company_code}")
def company(company_code: str, response: Response):
if company_code not in self.companies:
response.status_code = 404
return {"detail": "no such company"}
return {"company_code": company_code, "name": company_code, "country": "GB", "functional_currency": "GBP", "chart_id": "UK",
"bp_id": 1, "financial_year_end_month": 12, "ledgers": ["0L"]}
@app.get("/periods/open")
def open_(date: str):
return {"open": not (self.closed_before and date <= self.closed_before)}
class Purchasing:
"""purchasing's PUR-64: POST /bills/intercompany records a supplier bill keyed bill_id: 201 first, 200 the same for the same key."""
def __init__(self):
self.bills = {}
self.down = False
app = FastAPI()
self.app = app
@app.post("/bills/intercompany")
async def bill(request: Request, response: Response):
b = await request.json()
if self.down:
response.status_code = 503
return {"detail": "down"}
if b["bill_id"] in self.bills:
return {"bill_id": b["bill_id"], "status": "recorded"}
self.bills[b["bill_id"]] = b
response.status_code = 201
return {"bill_id": b["bill_id"], "status": "recorded"}
class Peppol:
"""An access-point provider composed beside the brick (capability convention): POST /capabilities/einvoice.send claims the key,
answers RESULT {status: done, provider_ref, output: {message_id}}; the same key answers the same message."""
def __init__(self):
self.sent, self.by_key = [], {}
self.down = False
app = FastAPI()
self.app = app
@app.post("/capabilities/einvoice.send")
async def send(request: Request, response: Response):
b = await request.json()
if self.down:
response.status_code = 503
return {"detail": "down"}
key = b["idempotency_key"]
if key not in self.by_key:
self.by_key[key] = f"MSG-{len(self.by_key) + 1}"
self.sent.append({"key": key, "message_id": self.by_key[key], **b["input"]})
return {"status": "done", "provider": "peppol", "provider_ref": self.by_key[key], "output": {"message_id": self.by_key[key]},
"idempotent_replay": False}
class Partners:
def __init__(self):
self.migration_map = {} # odoo res.partner id -> bp_id
app = FastAPI()
self.app = app
@app.get("/partners/{bp_id}")
def partner(bp_id: int, response: Response):
if bp_id >= 9_000_000:
response.status_code = 404
return {"detail": "no such partner"}
return {"bp_id": bp_id, "can_transact": True}
@app.get("/migration-map/{db}/{model}")
def mapped(db: str, model: str, ids: str = ""):
return [{"odoo_model": model, "odoo_id": i, "ours": str(self.migration_map[int(i)])}
for i in ids.split(",") if i and int(i) in self.migration_map]
class Orders:
def __init__(self):
self.lines = {"SO-1": {1: {"line_no": 1, "unit_code": "MON", "product_id": "HOST-M"}, 2: {"line_no": 2, "unit_code": "C62", "product_id": "SETUP"}}}
app = FastAPI()
self.app = app
@app.get("/sales-orders/{order_id}")
def order(order_id: str, response: Response):
"""sales-order's get_sales_order: the order, whole, with its lines."""
if order_id not in self.lines:
response.status_code = 404
return {"detail": f"no sales order {order_id}"}
return {"sales_order_id": order_id, "lifecycle": "confirmed", "lines": list(self.lines[order_id].values())}
class Products:
def __init__(self):
self.used = []
self.rows = {"HOST-M": {"product_id": "HOST-M", "name": "Managed hosting, per month", "sales_unit_code": "MON",
"base_unit_code": "MON", "tax_category": "standard", "category_code": "HOSTING"},
"SETUP": {"product_id": "SETUP", "name": "Setup", "sales_unit_code": "C62", "base_unit_code": "C62",
"tax_category": "zero"}}
app = FastAPI()
self.app = app
@app.get("/products/{pid}")
def product(pid: str, response: Response):
if pid not in self.rows:
response.status_code = 404
return {"detail": "no product"}
return self.rows[pid]
@app.post("/products/{pid}/units/{unit}/used")
def used(pid: str, unit: str):
self.used.append((pid, unit))
return {"ok": True}
@app.get("/migration-map/{db}/{model}")
def mapped(db: str, model: str, ids: str = ""):
return []
class World:
"""All the neighbours, registered; `World().install(...)` picks which are present."""
def __init__(self):
self.tax, self.payments, self.gl = Tax(), Payments(), Ledger()
self.bp, self.orders, self.products = Partners(), Orders(), Products()
self.pur = Purchasing()
self.peppol = Peppol()
# registered under the composition's member names: the names ar's seams ask for
self._names = {"tax": self.tax, "payments": self.payments, "gl": self.gl, "business-partner": self.bp, "sales-order": self.orders,
"product": self.products, "purchasing": self.pur, "peppol": self.peppol}
def install(self, *names):
for n in names:
seam.register_local(n, self._names[n].app)
return self
def remove(self, *names):
for n in names:
seam.unregister_local(n)
def everything(self):
return self.install(*self._names)
# ── Odoo: rows as search_read returns them (many2one = [id, name]); no live Odoo anywhere ─────────────────────────────
def _m2o(i, name=""):
return [i, name] if i else False
def tax_line(move_id, lid, tax_id, base, amount, sign=-1):
return {"id": lid, "move_id": [move_id, ""], "display_type": "tax", "tax_line_id": [tax_id, "VAT"], "tax_base_amount": base,
"balance": sign * amount, "amount_currency": sign * amount, "tax_ids": []}
def product_line(move_id, lid, seq, name, qty, price, net, total, tax_ids=(1,), sale_line=None):
return {"id": lid, "move_id": [move_id, ""], "sequence": seq, "display_type": "product", "product_id": False, "name": name, "quantity": qty,
"product_uom_id": False, "price_unit": price, "discount": 0.0, "price_subtotal": net, "price_total": total,
"tax_ids": list(tax_ids), "tax_line_id": False, "tax_base_amount": 0, "balance": -net, "amount_currency": -net,
"sale_line_ids": [sale_line] if sale_line else []}
def move(mid, name, mtype="out_invoice", state="posted", partner=10, date="2026-03-01", due="2026-03-31", cur=(1, "GBP"), net=1000.0, tax=200.0,
residual=0.0, pay="paid", reversed_=0, signed=None, origin="", term=1):
total = round(net + tax, 2)
return {"id": mid, "name": name, "move_type": mtype, "state": state, "payment_state": pay, "partner_id": [partner, "P"],
"commercial_partner_id": [partner, "P"], "invoice_date": date, "invoice_date_due": due, "date": date,
"invoice_payment_term_id": _m2o(term, "30 days"), "currency_id": list(cur), "amount_untaxed": net, "amount_tax": tax,
"amount_total": total, "amount_residual": residual, "amount_untaxed_signed": net, "amount_total_signed": signed or total,
"amount_residual_signed": residual, "ref": f"PO-{mid}", "payment_reference": "", "invoice_origin": origin,
"reversed_entry_id": _m2o(reversed_, ""), "journal_id": [1, "Sales"], "company_id": [1, "Co"], "write_date": "2026-10-31 10:00:00"}
class Odoo:
"""A small Odoo: two partners, two taxes, one payment term, and moves chosen to exercise every reconcile check."""
db = "odooprod"
def __init__(self):
self.cutover = "2026-10-31"
self.move_rows = [
move(100, "INV/2026/00001", net=1000.0, tax=200.0, residual=700.0, pay="partial", origin="SO-A"),
move(101, "INV/2026/00002", partner=11, cur=(2, "EUR"), date="2026-04-01", due="2026-04-15", net=500.0, tax=0.0,
residual=500.0, pay="not_paid", signed=430.0),
move(102, "INV/2026/00004", date="2026-05-02", due="2026-06-01", net=100.0, tax=19.99, residual=0.0, pay="paid"),
move(103, "RINV/2026/00001", "out_refund", date="2026-06-10", due="2026-06-10", net=100.0, tax=20.0, residual=0.0, pay="paid", reversed_=100),
move(104, "INV/2026/00005", state="cancel", date="2026-07-01", due="2026-07-31", net=50.0, tax=10.0, residual=0.0, pay="not_paid"),
move(105, "/", state="draft", date="2026-10-20", due="2026-11-20", net=70.0, tax=14.0, residual=70.0, pay="not_paid"),
move(106, "INV/2018/00001", date="2018-02-01", due="2018-03-03", net=10.0, tax=2.0, residual=0.0, pay="paid"),
move(107, "BILL/2026/00001", "in_invoice", date="2026-02-01", due="2026-03-03", net=10.0, tax=2.0, residual=0.0, pay="paid"),
]
self.line_rows = {
100: [product_line(100, 1001, 10, "Hosting", 10.0, 100.0, 1000.0, 1200.0, sale_line=55), tax_line(100, 1002, 1, 1000.0, 200.0)],
101: [product_line(101, 1011, 10, "Export service", 5.0, 100.0, 500.0, 500.0, tax_ids=(2,)), tax_line(101, 1012, 2, 500.0, 0.0)],
102: [product_line(102, 1021, 10, "Support", 1.0, 100.0, 100.0, 119.99), tax_line(102, 1022, 1, 100.0, 19.99)],
103: [product_line(103, 1031, 10, "Credit", 1.0, 100.0, 100.0, 120.0), tax_line(103, 1032, 1, 100.0, 20.0, sign=1)],
104: [product_line(104, 1041, 10, "Cancelled", 1.0, 50.0, 50.0, 60.0), tax_line(104, 1042, 1, 50.0, 10.0)],
105: [product_line(105, 1051, 10, "Draft", 1.0, 70.0, 70.0, 84.0), tax_line(105, 1052, 1, 70.0, 14.0)],
106: [product_line(106, 1061, 10, "Old", 1.0, 10.0, 10.0, 12.0), tax_line(106, 1062, 1, 10.0, 2.0)],
}
self.partner_rows = [{"id": 10, "name": "Acme Ltd", "email": "ap@acme.example", "street": "1 High St", "street2": False, "zip": "N1",
"city": "London", "property_payment_term_id": [1, "30 days"]},
{"id": 11, "name": "Beta GmbH", "email": "ap@beta.example", "street": "2 Allee", "street2": False, "zip": "10115",
"city": "Berlin", "property_payment_term_id": [1, "30 days"]}]
self.receivable = {"GBP": "700.00", "EUR": "500.00"}
# the OdooReader port
def company_currency(self):
return "GBP"
def moves(self, company_id, include_drafts):
return [dict(m) for m in self.move_rows]
def lines(self, ids):
return {i: [dict(x) for x in self.line_rows.get(i, [])] for i in ids}
def payment_terms(self, ids):
return [{"id": 1, "name": "30 days", "active": True}], [{"payment_id": [1, ""], "nb_days": 30, "delay_type": "days_after", "value": "balance"}]
def partners(self, ids):
return [p for p in self.partner_rows if p["id"] in ids]
def receivable_balance(self, cutover):
return dict(self.receivable)