dodil/erp-brick-expenses
DODIL ERP brick: expenses
tests/fakes.py
"""In-memory stand-ins for the neighbours, registered as local seams (erp_platform.seam.register_local): hr (the person, the
employment, the derived manager), gl (the ledger) and payments (the payout). They hold plain dicts the tests fill, answer the
contract infrastructure/seams.py documents, and can be taken DOWN. conftest.py checks every call against the CALLEE's signed spec."""
from decimal import Decimal
from typing import Optional
from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse
class Neighbours:
def __init__(self):
self.persons: dict = {} # sub -> {person_id, bp_id, company_code, email}
self.left_on: dict = {} # person_id -> the day after which hr says "not employed"
self.managers: dict = {} # person_id -> [level 1, level 2]: each {person_id, bp_id, pool_user_sub, email} or None
self.cost_centres: dict = {} # person_id -> cost centre
self.hr_status: Optional[int] = None # 500 takes hr down
self.journals: dict = {} # journal_id -> body
self.journal_posts = 0 # every POST, replays included
self.closed_through: Optional[str] = None # a journal_date on or before this is refused 409 period_open
self.gl_status: Optional[int] = None # 500 takes gl down
self.payouts: dict = {} # payout_ref -> body
self.payout_posts = 0
self.payout_refusal: Optional[dict] = None # {"gate":..., "reason":..., "fix":...} -> 422
self.pay_status: Optional[int] = None
self.exported: set = set() # payout_refs already in an exported payment run: cancel is refused 409 not_exported
self.cancelled: dict = {} # payout_ref -> reason
self.cancel_posts = 0
self.fx_rates: dict = {} # currency -> [(rate_date ISO, "0.870000")]: the newest ON OR BEFORE the date answers, as gl's GL-33
self.fx_calls: list = []
self.tax_calls: list = [] # every POST /documents/compute body
self.check_calls: list = [] # (document_ref, body) of every check-contents
self.tax_status: Optional[int] = None # 500 takes tax down
self.tax_refusal: Optional[dict] = None # {"gate": ..., "reason": ..., "fix": ...} -> 422 on compute
self.grosses: dict = {} # document_ref -> gross of the computed receipt (the content check needs it for the 250.00 rule)
self.invoices: dict = {} # source_id -> body: ar raises ONE draft per (source_kind, source_id)
self.invoice_posts = 0
self.ar_status: Optional[int] = None # 500 takes ar down
self.known_customers: Optional[set] = None # None: every bp_id is a customer; a set: others are refused 422 {gate: customer}
self.hr = self._hr()
self.gl = self._gl()
self.payments = self._payments()
self.tax = self._tax()
self.ar = self._ar()
# ── builders ────────────────────────────────────────────────────────────────────────────────
def employee(self, person_id: int, email: str, *, company: str = "UK01", bp_id: Optional[int] = None, cost_centre: str = "SALES-N",
manager: Optional[dict] = None, manager2: Optional[dict] = None) -> dict:
p = {"person_id": person_id, "bp_id": bp_id if bp_id is not None else 70000 + person_id, "company_code": company, "email": email,
"pool_user_sub": email}
self.persons[email] = p
self.cost_centres[person_id] = cost_centre
self.managers[person_id] = [manager, manager2]
return p
def as_manager(self, person_id: int, email: str) -> dict:
return {"person_id": person_id, "bp_id": 70000 + person_id, "pool_user_sub": email, "email": email}
# ── the apps ────────────────────────────────────────────────────────────────────────────────
def _hr(self) -> FastAPI:
app, me = FastAPI(), self
def person(pid: int):
return next((p for p in me.persons.values() if p["person_id"] == pid), None)
@app.get("/persons")
def by_sub(response: Response, pool_user_sub: str):
"""hr HR-58: 200 {person_id, bp_id, company_code}; 404 when no person carries the sub."""
if me.hr_status:
response.status_code = me.hr_status
return {"detail": "down"}
p = me.persons.get(pool_user_sub)
if p is None:
response.status_code = 404
return {"detail": "no such person"}
return {k: p[k] for k in ("person_id", "bp_id", "company_code")}
@app.get("/persons/{person_id}/employment")
def employment(person_id: int, on: str, response: Response):
"""hr HR-56: 200 {person_id, on, employed, engagement_id, worker_type, company_code, org_unit_id, location_code, cost_centre, bp_id}."""
if me.hr_status:
response.status_code = me.hr_status
return {"detail": "down"}
p = person(person_id)
if p is None:
response.status_code = 404
return {"detail": "no such person"}
gone = me.left_on.get(person_id)
return {"person_id": person_id, "on": on, "employed": not (gone and on > gone), "company_code": p["company_code"],
"cost_centre": me.cost_centres.get(person_id), "bp_id": p["bp_id"]}
@app.get("/persons/{person_id}/manager")
def manager(person_id: int, on: str, level: int, response: Response):
"""hr HR-57: 200 {person_id, on, level, manager: {...} | null, source}."""
if me.hr_status:
response.status_code = me.hr_status
return {"detail": "down"}
chain = me.managers.get(person_id, [None, None])
return {"person_id": person_id, "on": on, "level": level, "manager": chain[level - 1] if level <= len(chain) else None,
"source": "org_tree"}
return app
def _gl(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/fx-rates/{currency}")
def fx(currency: str, on: str, response: Response, type: str = "spot", to: str = "GBP"):
"""gl GL-33: 200 {currency, to, type, on, rate_date, rate (6 places), source}; 404 {gate: fx_rate} when none on or before `on`."""
me.fx_calls.append((currency, on, type, to))
if me.gl_status:
response.status_code = me.gl_status
return {"detail": "down"}
if currency == to:
return {"currency": currency, "to": to, "type": type, "on": on, "rate_date": on, "rate": "1", "source": "identity"}
have = sorted(r for r in me.fx_rates.get(currency, []) if r[0] <= on)
if not have:
return JSONResponse({"gate": "fx_rate", "reason": f"no {currency} rate on or before {on}", "fix": "enter the rate"}, status_code=404)
return {"currency": currency, "to": to, "type": type, "on": on, "rate_date": have[-1][0], "rate": have[-1][1], "source": "manual"}
@app.post("/journal-entry/journals")
def journal(body: dict, response: Response):
"""gl GL-30 for expenses: 201 {journal_id, replay: false}; 200 the same with replay: true; 409 {gate: period_open}."""
me.journal_posts += 1
if me.gl_status:
response.status_code = me.gl_status
return {"detail": "down"}
dr = sum(Decimal(x.get("debit", "0")) for x in body["lines"])
cr = sum(Decimal(x.get("credit", "0")) for x in body["lines"])
if dr != cr:
return JSONResponse({"gate": "balanced", "reason": f"{dr} against {cr}", "fix": "balance it"}, status_code=422)
if me.closed_through and body["journal_date"] <= me.closed_through:
return JSONResponse({"gate": "period_open", "reason": f"the period of {body['journal_date']} is closed", "fix": "post it in the next open period"},
status_code=409)
replay = body["journal_id"] in me.journals
me.journals[body["journal_id"]] = body
response.status_code = 200 if replay else 201
return {"journal_id": body["journal_id"], "status": "posted", "replay": replay, "lines": len(body["lines"])}
return app
def _ar(self) -> FastAPI:
app, me = FastAPI(), self
@app.post("/invoices/raise")
def raise_invoice(body: dict, response: Response):
"""ar AR-20: 201 the draft invoice {invoice_id, status: draft, lines, tax, totals}; 200 the same for the same source_id; 422 {gate, reason, fix}."""
me.invoice_posts += 1
if me.ar_status:
response.status_code = me.ar_status
return {"detail": "down"}
if me.known_customers is not None and int(body["bp_id"]) not in me.known_customers:
return JSONResponse({"gate": "customer", "reason": f"no customer {body['bp_id']}", "fix": "name a customer partner"}, status_code=422)
have = me.invoices.get(body["source_id"])
response.status_code = 200 if have else 201
me.invoices.setdefault(body["source_id"], body)
return {"invoice_id": f"INV-{list(me.invoices).index(body['source_id']) + 1}", "status": "draft", "lines": body["lines"], "tax": {}, "totals": {}}
return app
def _tax(self) -> FastAPI:
app, me = FastAPI(), self
rates = {"standard": Decimal("0.20"), "entertainment": Decimal("0.20"), "zero": Decimal("0"), "outside_scope": Decimal("0")}
codes = {"standard": "PS", "entertainment": "PE", "zero": "PZ", "outside_scope": "PO"}
@app.post("/documents/compute")
def compute(body: dict, response: Response):
"""tax TAX-53: a purchase document without a partner: per line {tax_code, rate_pct, net, vat, gross, vat_reclaimable, vat_non_reclaimable};
business entertainment is a blocked input code (VAT a cost, not a reclaim)."""
me.tax_calls.append(body)
if me.tax_status:
response.status_code = me.tax_status
return {"detail": "down"}
if me.tax_refusal:
return JSONResponse(me.tax_refusal, status_code=422)
out = []
for ln in body["lines"]:
cat = ln["category_code"]
rate = rates.get(cat, Decimal("0.20"))
gross = Decimal(ln["amount"])
vat = (gross * rate / (1 + rate)).quantize(Decimal("0.01"))
me.grosses[body["document_ref"]] = gross
out.append({"line_no": ln["line_no"], "tax_code": codes.get(cat, "PS"), "rate_pct": f"{rate * 100:.2f}", "net": f"{gross - vat:.2f}", "vat": f"{vat:.2f}",
"gross": f"{gross:.2f}", "vat_reclaimable": "0.00" if cat == "entertainment" else f"{vat:.2f}",
"vat_non_reclaimable": f"{vat:.2f}" if cat == "entertainment" else "0.00"})
return {"document_ref": body["document_ref"], "status": "computed", "lines": out, "totals": {}, "gates": [], "version": 1}
@app.post("/documents/{document_ref}/check-contents")
def check(document_ref: str, body: dict, response: Response):
"""tax TAX-24/TAX-50: the supplier's name, address and VAT number and the invoice number; our name as the customer above 250.00."""
me.check_calls.append((document_ref, body))
if me.tax_status:
response.status_code = me.tax_status
return {"detail": "down"}
c = body["contents"]
gates = [{"gate": g, "met": bool(c.get(g)), "reason": "" if c.get(g) else f"the receipt shows no {g.replace('_', ' ')}: input VAT cannot be reclaimed on it",
"fix": "ask the supplier for a VAT invoice"} for g in ("supplier_name", "supplier_address", "supplier_vat_number", "invoice_number")]
gross = me.grosses.get(document_ref, Decimal(0))
full = gross > Decimal("250.00")
gates.append({"gate": "customer_name", "met": bool(c.get("customer_name")) or not full,
"reason": "" if (c.get("customer_name") or not full) else "a VAT invoice over 250.00 must show your company's name", "fix": "ask for a full invoice"})
return {"document_ref": document_ref, "status": "checked", "ready": all(g["met"] for g in gates), "title": "full_invoice" if full else "less_detailed_invoice",
"gates": gates}
return app
def _payments(self) -> FastAPI:
app, me = FastAPI(), self
@app.post("/payouts/{payout_ref}/cancel")
def cancel(payout_ref: str, body: dict, response: Response):
"""payments PAY-51: 200 {payout_ref, status: cancelled} (a repeat the same); 409 {gate: not_exported} once exported; 404 an unknown ref."""
me.cancel_posts += 1
if me.pay_status:
response.status_code = me.pay_status
return {"detail": "down"}
if payout_ref not in me.payouts:
return JSONResponse({"gate": "payout", "reason": f"no payout {payout_ref}", "fix": ""}, status_code=404)
if payout_ref in me.exported:
return JSONResponse({"gate": "not_exported", "reason": f"{payout_ref} is in an exported payment run", "fix": "recover it from the employee"},
status_code=409)
me.cancelled[payout_ref] = body["reason"]
return {"payout_ref": payout_ref, "status": "cancelled"}
@app.post("/payouts")
def payout(body: dict, response: Response):
"""payments PAY-51: 201 {payout_ref, payout_id, status: requested}; 200 the same; 409 another amount; 422 {gate}."""
me.payout_posts += 1
if me.pay_status:
response.status_code = me.pay_status
return {"detail": "down"}
if me.payout_refusal:
return JSONResponse(me.payout_refusal, status_code=422)
have = me.payouts.get(body["payout_ref"])
if have is not None:
if have["amount"] != body["amount"]:
return JSONResponse({"gate": "amount_matches", "reason": "another amount", "fix": "cancel it first"}, status_code=409)
response.status_code = 200
else:
me.payouts[body["payout_ref"]] = body
response.status_code = 201
return {"payout_ref": body["payout_ref"], "payout_id": f"payout:{len(me.payouts)}", "status": "requested"}
return app