dodil/erp-brick-payments
DODIL ERP brick: payments
tests/fakes.py
"""In-memory stand-ins for the neighbours, registered as local seams (erp_platform.seam.register_local): invoicing (ar),
the partner register (bp) and the ledger (gl). They hold plain dicts the tests fill, and answer the contract
infrastructure/seams.py documents. A test can also take one DOWN."""
from decimal import Decimal
from typing import Optional
from fastapi import FastAPI, Response
class Neighbours:
def __init__(self):
self.invoices: dict = {}
self.credit_notes: dict = {}
self.writeoffs: list = []
self.journals: dict = {}
self.ar_balances: dict = {}
self.ar_status: Optional[int] = None # set to 500 to take invoicing down
self.unknown_partners_from = 9_000_000
self.discount_calls: list = []
self.discount_refusal: Optional[dict] = None # set to {"gate":..., "reason":..., "fix":...} to make ar answer 422
self.liabilities: list = []
self.tax_calls: list = []
self.vat_rate = Decimal("0.20")
self.bills: list = []
self.bills_status: Optional[int] = None # set to 500 to take purchasing down
self.cis_due: list = []
self.requests: dict = {}
self.supplier_payments: dict = {}
self.confirmations: list = []
self.confirm_calls: list = []
self.settled: dict = {}
self.cis: dict = {}
self.cis_calls: list = []
self.book_opening: dict = {}
self.companies = {"UK01": "GBP", "UK02": "GBP", "IE01": "EUR"}
self.tax = self._tax()
self.purchasing = self._purchasing()
self.ar = self._ar()
self.bp = self._bp()
self.gl = self._gl()
# ── builders ────────────────────────────────────────────────────────────────────────────────
def invoice(self, invoice_id: str, bp_id: int, total: str, *, currency: str = "GBP", number: Optional[str] = None,
issued_on: str = "2026-10-01", due_date: str = "2026-10-31", state: str = "issued",
discount_pct: Optional[str] = None, discount_until: Optional[str] = None,
fx_rate: Optional[str] = None) -> str:
self.invoices[invoice_id] = {"invoice_id": invoice_id, "number": number or invoice_id, "bp_id": bp_id, "currency": currency,
"total": str(Decimal(total)), "issued_on": issued_on, "due_date": due_date, "state": state,
"discount_pct": discount_pct, "discount_until": discount_until, "fx_rate": fx_rate}
return invoice_id
def supplier_payment(self, payment_id: str, amount: str, *, status: str = "sent", bank_account_id: str = "BANK1", bp_id: int = 9001,
reference: str = "PAY-RUN-0045-03") -> str:
self.supplier_payments[payment_id] = {"payment_id": payment_id, "run_id": "RUN-0045", "bp_id": bp_id, "bank_account_id": bank_account_id,
"amount": str(Decimal(amount)), "currency": "GBP", "reference": reference, "status": status,
"paid_on": None, "bills": [{"bill_id": "B-1", "amount": str(Decimal(amount))}]}
return payment_id
def credit_note(self, cn_id: str, bp_id: int, total: str, *, currency: str = "GBP", issued_on: str = "2026-11-01") -> str:
self.credit_notes[cn_id] = {"credit_note_id": cn_id, "number": cn_id, "bp_id": bp_id, "currency": currency,
"total": str(Decimal(total)), "issued_on": issued_on}
return cn_id
# ── the apps ────────────────────────────────────────────────────────────────────────────────
def _ar(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/invoice-totals")
def invoice_totals(response: Response, invoice_ids: Optional[str] = None, bp_id: Optional[int] = None,
limit: int = 1000, offset: int = 0):
"""ar's invoice_totals: by ids, or the ISSUED invoices (of one customer), a page at a time."""
if me.ar_status:
response.status_code = me.ar_status
return {"detail": "down"}
if invoice_ids:
rows = [me.invoices[i] for i in invoice_ids.split(",") if i in me.invoices]
else:
rows = [i for i in me.invoices.values() if (bp_id is None or i["bp_id"] == bp_id) and i["state"] == "issued"]
rows, more = rows[offset:offset + limit], offset + limit < len(rows)
return {"invoices": {i["invoice_id"]: {"bp_id": i["bp_id"], "lifecycle": i["state"], "total": i["total"], "currency": i["currency"],
"issue_date": i["issued_on"], "due_date": i["due_date"], "invoice_no": i["number"],
**({"fx_rate": i["fx_rate"]} if i.get("fx_rate") else {}),
**({"settlement_discount_pct": i["discount_pct"], "settlement_discount_until": i["discount_until"]}
if i.get("discount_pct") else {})}
for i in rows}, "more": more if not invoice_ids else False}
@app.get("/credit-notes")
def credit_notes(response: Response, bp_id: Optional[int] = None, status: Optional[str] = None, limit: int = 50, offset: int = 0):
"""ar's list_credit_notes: its own field names (credit_id, credit_no, credit_date)."""
if me.ar_status:
response.status_code = me.ar_status
return {"detail": "down"}
rows = [{"credit_id": c["credit_note_id"], "credit_no": c["number"], "bp_id": c["bp_id"], "invoice_id": c.get("invoice_id"),
"credit_date": c["issued_on"], "currency": c["currency"], "status": "approved", "total": c["total"]}
for c in me.credit_notes.values() if bp_id is None or c["bp_id"] == bp_id]
return {"credit_notes": rows[offset:offset + limit]}
@app.get("/migration/open-balances")
def balances(db: str):
return {"run_id": f"odoo:{db}:2026-10-31", "status": "reconciled",
"balances": [{"bp_id": bp, "currency": cur, "open_balance": v} for (bp, cur), v in me.ar_balances.items()]}
@app.post("/invoices/{invoice_id}/settlement-discount")
def discount(invoice_id: str, body: dict, response: Response):
"""ar AR-38: 201 {credit_id, credit_no, net, tax, total, approved_by}; 200 the same for the same payment_id; 422 {gate, reason, fix}."""
if me.discount_refusal:
response.status_code = 422
return me.discount_refusal
for c in me.discount_calls:
if c["payment_id"] == body["payment_id"] and c["invoice_id"] == invoice_id:
response.status_code = 200
return c["credit"]
total = Decimal(body["discount_amount"])
net = (total / Decimal("1.2")).quantize(Decimal("0.01"))
credit = {"credit_id": f"cn:{len(me.discount_calls) + 1}", "credit_no": f"CN-{len(me.discount_calls) + 1}", "net": str(net),
"tax": str(total - net), "total": str(total), "approved_by": "terms"}
me.discount_calls.append({"invoice_id": invoice_id, "payment_id": body["payment_id"], "body": body, "credit": credit})
response.status_code = 201
return credit
@app.get("/payment-requests/{request_id}")
def get_request(request_id: str, response: Response):
"""ar AR-36's read: {request_id, bp_id, currency, remaining, ...}; 404 for an unknown request."""
req = me.requests.get(request_id)
if req is None:
response.status_code = 404
return {"detail": "no such request"}
return {"request_id": request_id, "bp_id": req["bp_id"], "currency": req["currency"], "remaining": str(req["remaining"])}
@app.post("/payment-requests/{request_id}/settle")
def settle_request(request_id: str, body: dict, response: Response):
"""ar AR-36: 200 {request_id, deposit_invoice_id, invoice_no, amount, remaining, status}; the same payment settles once; 422 {gate: within_request}."""
req = me.requests.get(request_id)
if req is None:
response.status_code = 404
return {"detail": "no such request"}
if body["payment_id"] in me.settled:
return me.settled[body["payment_id"]]
amount = Decimal(body["amount"])
if amount > req["remaining"]:
response.status_code = 422
return {"gate": "within_request", "met": False, "reason": f"{amount} is more than the {req['remaining']} left on the request",
"fix": "Settle at most what remains"}
req["remaining"] -= amount
req["n"] += 1
inv = f"dep:{request_id}:{req['n']}"
me.invoice(inv, req["bp_id"], str(amount), currency=req["currency"], number=f"INV-{inv}")
out = {"request_id": request_id, "deposit_invoice_id": inv, "invoice_no": f"INV-{inv}", "amount": str(amount),
"remaining": str(req["remaining"]), "status": "settled" if req["remaining"] == 0 else "issued"}
me.settled[body["payment_id"]] = out
return out
@app.post("/writeoffs")
def propose(body: dict):
"""ar's propose_writeoff: {invoice_id, amount, reason}."""
me.writeoffs.append(body)
return {"writeoff_id": f"wo:{len(me.writeoffs)}", "status": "proposed", **body}
return app
def _tax(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/vat-returns/liabilities")
def liabilities():
return {"liabilities": me.liabilities, "cis_due": me.cis_due}
@app.post("/cis/suffered")
def cis(body: dict, response: Response):
"""tax TAX-41: 201 {recorded: true}; 200 {replay: true} for the same statement; 422 {gate: within_cis_rate, reason, fix}."""
key = (body["contractor_bp_id"], body["statement_ref"])
me.cis_calls.append(body)
if key in me.cis:
return {"replay": True}
labour = Decimal(body["gross"]) - Decimal(body["materials"])
if Decimal(body["deduction"]) > labour * Decimal("0.30"):
response.status_code = 422
return {"gate": "within_cis_rate", "met": False, "reason": f"{body['deduction']} is more than 30% of the labour {labour}",
"fix": "Check the statement"}
me.cis[key] = body
response.status_code = 201
return {"recorded": True}
@app.post("/documents/compute")
def compute(body: dict):
"""tax TAX-23, direction purchase, prices include tax: the line's vat / vat_reclaimable / vat_non_reclaimable."""
me.tax_calls.append(body)
ln = body["lines"][0]
gross = Decimal(ln["amount"])
vat = (gross - gross / (1 + me.vat_rate)).quantize(Decimal("0.01")) if body.get("treatment_code") == "S" else Decimal("0.00")
return {"document_ref": body["document_ref"], "status": "computed",
"lines": [{"line_no": 1, "tax_code": body.get("treatment_code"), "rate_pct": "20.00", "net": str(gross - vat), "vat": str(vat),
"gross": str(gross), "vat_reclaimable": str(vat), "vat_non_reclaimable": "0.00"}]}
return app
def _bp(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/partners/{bp_id}")
def partner(bp_id: int, response: Response):
if bp_id >= me.unknown_partners_from:
response.status_code = 404
return {"detail": "no such partner"}
return {"bp_id": bp_id, "can_transact": True}
return app
def _purchasing(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/supplier-payments")
def supplier_payments(bank_account_id: Optional[str] = None, status: Optional[str] = None):
"""purchasing PUR-66: 200 {payments: [{payment_id, run_id, bp_id, bank_account_id, amount, currency, reference, status, paid_on, bills}]}."""
return {"payments": [p for p in me.supplier_payments.values()
if (bank_account_id is None or p["bank_account_id"] == bank_account_id) and (status is None or p["status"] == status)]}
@app.post("/supplier-payments/{payment_id}/confirm")
def confirm(payment_id: str, body: dict, response: Response):
"""purchasing PUR-66: 200 {payment_id, status: confirmed, replay}; the same again replay: true; 409 {gate: payment_open}."""
me.confirm_calls.append(payment_id)
p = me.supplier_payments.get(payment_id)
if p is None or p["status"] == "rejected":
response.status_code = 409
return {"gate": "payment_open", "met": False, "reason": f"payment {payment_id} cannot be confirmed", "fix": "Check the payment run"}
replay = p["status"] == "confirmed"
if not replay:
p["status"], p["paid_on"] = "confirmed", body["paid_on"]
me.confirmations.append({"payment_id": payment_id, **body})
return {"payment_id": payment_id, "status": "confirmed", "replay": replay}
@app.get("/bills/open")
def bills_open(response: Response, as_of: str, bp_id: Optional[int] = None):
"""purchasing PUR-14: 200 {as_of, items: [{bill_id, bill_no, bp_id, currency, total, outstanding, due_date}]}."""
if me.bills_status:
response.status_code = me.bills_status
return {"detail": "down"}
return {"as_of": as_of, "items": [b for b in me.bills if bp_id is None or b["bp_id"] == bp_id]}
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."""
if company_code not in me.companies:
response.status_code = 404
return {"detail": "no such company"}
return {"company_code": company_code, "name": f"Company {company_code}", "country": "GB", "functional_currency": me.companies[company_code],
"chart_id": "UK", "bp_id": 1, "financial_year_end_month": 12, "ledgers": ["L1"]}
@app.get("/accounts/{account_id}/balance")
def account_balance(account_id: str, as_of: str, response: Response, company_code: Optional[str] = None):
"""gl GL-61: 200 {account_id, company_code, as_of, balance, currency} = the sum of the posted lines up to as_of; 404 unknown."""
if account_id not in me.book_opening and not any(ln.get("account_id") == account_id for j in me.journals.values() for ln in j["lines"]):
response.status_code = 404
return {"detail": "no such account"}
bal = me.book_opening.get(account_id, Decimal("0.00"))
for j in me.journals.values():
if j["journal_date"] <= as_of:
for ln in j["lines"]:
if ln.get("account_id") == account_id:
bal += Decimal(ln.get("debit", "0")) - Decimal(ln.get("credit", "0"))
return {"account_id": account_id, "company_code": company_code or "UK01", "as_of": as_of, "balance": str(bal), "currency": "GBP"}
@app.post("/journal-entry/journals")
def journal(body: dict):
assert str(body["journal_id"]).isdigit() and body["source_module"] == "payments" # gl's shape: a derived BIGINT id
for ln in body["lines"]:
assert ("account_id" in ln) != ("purpose" in ln) and set(ln) <= {"account_id", "purpose", "tax_code", "debit",
"credit", "memo", "bp_id", "cost_centre", "against_document"}
me.journals[f'{body["event"]}:{body["source_document"]}'] = body # derived id: posting twice is the same journal
return {"journal_id": body["journal_id"]}
return app