dodil/erp-brick-payroll
DODIL ERP brick: payroll
tests/fakes.py
"""In-memory stand-ins for the neighbours, registered as local seams (erp_platform.seam.register_local): hr (people, their
salaries and the payroll extract built from them per period, the employment, the work pattern), time-off (absences and a
holiday balance), gl (the ledger) and payments (the payouts). They answer the contract infrastructure/seams.py documents,
can be taken DOWN, and conftest.py checks every call against the CALLEE's signed spec."""
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
def month_bounds(period: str) -> tuple:
if "-W" in period:
y, w = period.split("-W")
first = date.fromisocalendar(int(y), int(w), 1)
return first, first + timedelta(days=6)
y, m = (int(x) for x in period.split("-"))
first = date(y, m, 1)
return first, (first.replace(day=28) + timedelta(days=4)).replace(day=1) - timedelta(days=1)
class Neighbours:
def __init__(self):
self.persons: dict = {} # person_id -> {person_id, bp_id, company_code, sub, cost_centre}
self.pay: dict = {} # person_id -> [{n, from, to, basis, amount, component}] (hr's pay slices)
self.start: dict = {} # person_id -> start date
self.last_day: dict = {} # person_id -> last day
self.extract_state: dict = {} # batch_id -> {"state", "sent_ref", "frozen": [...]}
self.extra_lines: dict = {} # period -> [extract lines hr adds for a person who has left (a late payment)]
self.ni_numbers: dict = {} # person_id -> NI number, or None (hr holds none: the FPS carries the address instead)
self.hr_status: Optional[int] = None
self.sent_calls = 0
self.absences: list = []
self.balances: dict = {} # person_id -> available days on the last day
self.journals: dict = {}
self.journal_posts = 0
self.closed_through: Optional[str] = None
self.gl_status: Optional[int] = None
self.payouts: dict = {}
self.payout_posts = 0
self.pay_status: Optional[int] = None
self.hr = self._hr()
self.time_off = self._time_off()
self.gl = self._gl()
self.payments = self._payments()
# ── builders ────────────────────────────────────────────────────────────────────────────────
def salaried(self, person_id: int, annual: str, *, sub: Optional[str] = None, start: str = "2020-01-06", last_day: Optional[str] = None,
cost_centre: str = "SALES-N", bp_id: Optional[int] = None, company: str = "UK01") -> dict:
p = {"person_id": person_id, "bp_id": bp_id if bp_id is not None else 5000 + person_id, "company_code": company,
"sub": sub, "cost_centre": cost_centre}
self.persons[person_id] = p
self.pay[person_id] = [{"n": 1, "from": date.fromisoformat(start), "to": date(2100, 1, 1), "basis": "year", "amount": annual, "component": "base"}]
self.start[person_id] = date.fromisoformat(start)
if last_day:
self.last_day[person_id] = date.fromisoformat(last_day)
return p
def late_payment(self, person_id: int, period: str, amount: str) -> None:
"""A payment hr still extracts for someone who has left (a late bonus), as one monthly pay line over the whole period."""
lo, hi = month_bounds(period)
self.extra_lines.setdefault(period, []).append({"person_id": person_id, "n": 1, "component": "base", "for_period": period, "rate_basis": "month",
"rate_amount": amount, "currency": "GBP", "prorate_from": lo.isoformat(), "prorate_to": hi.isoformat(),
"quantity_minutes": None, "quantity_days": None, "payable_amount": None, "change_kind": "pay"})
def pay_change(self, person_id: int, on: str, annual: str) -> None:
d = date.fromisoformat(on)
slices = self.pay[person_id]
slices[-1]["to"] = d - timedelta(days=1)
slices.append({**slices[-1], "from": d, "to": date(2100, 1, 1), "amount": annual})
def lines_for(self, period: str) -> dict:
lo, hi = month_bounds(period)
lines, starters, leavers = [], [], []
for pid in sorted(self.persons):
start, end = self.start[pid], self.last_day.get(pid, date(2100, 1, 1))
if start > hi or end < lo:
continue
for s in self.pay[pid]:
a, b = max(lo, s["from"], start), min(hi, s["to"], end)
if a > b:
continue
kind = "starter" if start == a and lo <= start <= hi else ("leaver" if end == b and lo <= end <= hi else "pay")
lines.append({"person_id": pid, "n": s["n"], "component": s["component"], "for_period": period, "rate_basis": s["basis"],
"rate_amount": s["amount"], "currency": "GBP", "prorate_from": a.isoformat(), "prorate_to": b.isoformat(),
"quantity_minutes": None, "quantity_days": None, "payable_amount": None, "change_kind": kind})
if lo <= start <= hi:
starters.append({"person_id": pid, "n": 1, "start_date": start.isoformat(), "worker_type": "employee"})
if lo <= end <= hi:
leavers.append({"person_id": pid, "n": 1, "last_day": end.isoformat(), "reason": "resignation"})
return {"lines": lines + self.extra_lines.get(period, []), "starters": starters, "leavers": leavers}
def sick(self, person_id: int, days: list, ref: str = "UK01-TOF-2026-00100", *, linked_to: Optional[str] = None, linked_group: Optional[str] = None,
evidence_ok: bool = True) -> None:
"""A closed sickness spell as time-off serves it (TOF-20): the days, and the facts for SSP (first and last day off, PIW, the linked spell)."""
first, last = date.fromisoformat(days[0]), date.fromisoformat(days[-1])
self.absences.append({"absence_ref": ref, "revision": 1, "person_id": person_id, "leave_type_code": "SICK", "pay_treatment": "sick",
"status": "closed", "evidence_ok": evidence_ok, "statutory": None, "changed_at": "2026-11-13T09:00:00Z",
"days": [{"date": d, "part": "full", "minutes": 450} for d in days],
"sickness": {"first_day": days[0], "last_day_off": days[-1], "self_certified": True, "qualifying_days": len(days),
"piw": (last - first).days + 1 >= 4, "linked_to": linked_to, "linked_group": linked_group, "fit_notes": []}})
def maternity(self, person_id: int, start: str, ewc: str, evidence: Optional[str], through: str, ref: str = "UK01-TOF-2026-00300") -> None:
"""Maternity leave as time-off serves it: the weekdays from `start` to `through`, with the expected week of childbirth and the MATB1 date."""
d, last, days = date.fromisoformat(start), date.fromisoformat(through), []
while d <= last:
if d.weekday() < 5:
days.append({"date": d.isoformat(), "part": "full", "minutes": 450})
d += timedelta(days=1)
self.absences.append({"absence_ref": ref, "revision": 1, "person_id": person_id, "leave_type_code": "MATERNITY", "pay_treatment": "statutory_maternity",
"status": "approved", "evidence_ok": True, "changed_at": "2026-06-02T09:00:00Z", "days": days,
"statutory": {"kind": "maternity", "ewc": ewc, "evidence_received_on": evidence, "kit_days": 0}})
def unpaid(self, person_id: int, days: list, ref: str = "UK01-TOF-2026-00200") -> None:
self.absences.append({"absence_ref": ref, "revision": 1, "person_id": person_id, "leave_type_code": "UNPAID", "pay_treatment": "unpaid",
"status": "approved", "evidence_ok": True, "statutory": None, "changed_at": "2026-11-10T09:00:00Z",
"days": [{"date": d, "part": "full", "minutes": 450} for d in days]})
# ── the apps ────────────────────────────────────────────────────────────────────────────────
def _down(self, response: Response):
response.status_code = self.hr_status
return {"detail": "down"}
def _hr(self) -> FastAPI:
app, me = FastAPI(), self
@app.post("/payroll-extracts")
def produce(body: dict, response: Response):
"""HR-60: 201 the draft batch; 200 the same re-produced; 200 the frozen batch once sent."""
if me.hr_status:
return me._down(response)
bid = f"{body['legal_entity_ref']}|{body['period']}"
st = me.extract_state.setdefault(bid, {"state": "draft", "sent_ref": None, "frozen": None, "new": True})
response.status_code = 201 if st.pop("new", False) else 200
got = st["frozen"] or me.lines_for(body["period"])
return {"batch_id": bid, "state": st["state"], "produced_at": "2026-11-24T09:00:00Z", "lines": len(got["lines"])}
@app.get("/payroll-extracts/{batch_id}")
def get_extract(batch_id: str, response: Response):
if me.hr_status:
return me._down(response)
st = me.extract_state.get(batch_id)
if st is None:
response.status_code = 404
return {"detail": "never produced"}
company, period = batch_id.split("|")
got = st["frozen"] or me.lines_for(period)
return {"batch_id": batch_id, "state": st["state"], **got, "bank_changed": []}
@app.post("/payroll-extracts/{batch_id}/sent")
def sent(batch_id: str, body: dict, response: Response):
me.sent_calls += 1
if me.hr_status:
return me._down(response)
st = me.extract_state.get(batch_id)
if st is None:
response.status_code = 404
return {"detail": "never produced"}
if st["state"] == "sent" and st["sent_ref"] != body["sent_ref"]:
return JSONResponse({"gate": "sent_once", "reason": f"sent as {st['sent_ref']}", "fix": "a later change is a retro line"}, status_code=409)
if st["state"] != "sent":
st.update(state="sent", sent_ref=body["sent_ref"], frozen=me.lines_for(batch_id.split("|")[1]))
return {"batch_id": batch_id, "state": "sent", "sent_ref": st["sent_ref"]}
@app.get("/persons")
def by_sub(response: Response, pool_user_sub: str):
if me.hr_status:
return me._down(response)
p = next((p for p in me.persons.values() if p["sub"] and p["sub"] == pool_user_sub), None)
if p is None:
response.status_code = 404
return {"detail": "no such person"}
return {"person_id": p["person_id"], "bp_id": p["bp_id"], "company_code": p["company_code"]}
@app.get("/persons/{person_id}/payroll-identity")
def identity(person_id: int, on: str, response: Response):
"""HR-61: the identity HMRC needs. Deterministic: the same person always reads the same."""
if me.hr_status:
return me._down(response)
if person_id not in me.persons:
response.status_code = 404
return {"detail": "no such person"}
ni = me.ni_numbers.get(person_id, f"QQ{person_id % 1000000:06d}C")
return {"person_id": person_id, "legal_first_names": f"Pat{person_id}", "legal_last_name": f"Example{person_id}", "date_of_birth": "1990-05-17",
"ni_number": ni, "home_address": {"line1": f"{person_id} High Street", "city": "London", "postcode": "N1 9GU", "country": "GB"},
"valid_from": "2020-01-06"}
@app.get("/persons/{person_id}/employment")
def employment(person_id: int, on: str, response: Response):
if me.hr_status:
return me._down(response)
p = me.persons.get(person_id)
d = date.fromisoformat(on)
employed = bool(p) and me.start[person_id] <= d <= me.last_day.get(person_id, date(2100, 1, 1))
return {"person_id": person_id, "on": on, "employed": employed, "engagement_id": None, "worker_type": "employee",
"company_code": p["company_code"] if p else None, "org_unit_id": None, "location_code": "LON",
"cost_centre": p["cost_centre"] if p else None, "bp_id": p["bp_id"] if p else None}
@app.get("/persons/{person_id}/work-pattern")
def work_pattern(person_id: int, request: Request, response: Response):
"""HR-62: Monday to Friday, 450 minutes; employed per day from the engagement ('from' is a keyword: read from the query)."""
if me.hr_status:
return me._down(response)
frm, to = date.fromisoformat(request.query_params["from"]), date.fromisoformat(request.query_params["to"])
days, d = [], frm
while d <= to:
employed = me.start[person_id] <= d <= me.last_day.get(person_id, date(2100, 1, 1))
days.append({"date": d.isoformat(), "employed": employed, "contracted_minutes": 450 if d.weekday() < 5 else 0,
"public_holiday": None, "location_code": "LON", "holiday_calendar": "GB-EAW", "fte": "1.0000"})
d += timedelta(days=1)
return {"person_id": person_id, "from": frm.isoformat(), "to": to.isoformat(), "days": days}
return app
def _time_off(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/absences")
def absences(request: Request):
q = request.query_params
frm, to = q["from"], q["to"]
rows = [a for a in me.absences if any(frm <= d["date"] <= to for d in a["days"])
and (q.get("person_id") is None or int(q["person_id"]) == a["person_id"])]
rows = [{**a, "days": [d for d in a["days"] if frm <= d["date"] <= to]} for a in rows]
return {"company_code": q["company_code"], "from": frm, "to": to, "as_of": "2026-11-24T09:00:00Z", "absences": rows}
@app.get("/persons/{person_id}/balance")
def balance(person_id: int, on: str, leave_type_code: str = "ANNUAL"):
avail = me.balances.get(person_id, "0.00")
return {"person_id": person_id, "on": on, "leave_type_code": leave_type_code, "unit": "days", "entitled": "25.00", "carried": "0.00",
"taken": "0.00", "booked": "0.00", "adjustments": "0.00", "available": avail, "day_minutes": 450}
return app
def _gl(self) -> FastAPI:
app, me = FastAPI(), self
@app.post("/journal-entry/journals")
def journal(body: dict, response: Response):
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": "open it"}, 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"], "entry_no": len(me.journals), "period_id": body["journal_date"][:7], "status": "posted",
"replay": replay, "lines": len(body["lines"])}
return app
def _payments(self) -> FastAPI:
app, me = FastAPI(), self
@app.post("/payouts")
def payout(body: dict, response: Response):
me.payout_posts += 1
if me.pay_status:
response.status_code = me.pay_status
return {"detail": "down"}
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
class RtiProvider:
"""A stand-in provider of hmrc.rti_submit and pension.submit_contributions (no live HMRC, no live scheme): answers by `mode`, replays a key it has already
answered terminally (the platform's rule 1), and records every call so a test can say what was handed over, and how often."""
def __init__(self):
self.mode = "accept" # accept | reject | down
self.calls: list = [] # (capability, key, input) in order, replays included
self.answers: dict = {} # (capability, key) -> the terminal answer
def _answer(self, capability: str, key: str, data: dict) -> dict:
self.calls.append((capability, key, data))
if self.mode == "down":
raise ConnectionError("the provider did not answer")
if (capability, key) in self.answers:
return self.answers[(capability, key)]
if self.mode == "reject":
got = {"status": "rejected", "errors": [{"code": "1234", "field": "TaxCode", "message": "the tax code is not valid for this employee"}]}
else:
got = {"status": "accepted", "correlation_id": f"CORR-{len(self.answers) + 1:04d}"}
self.answers[(capability, key)] = got
return got
def hmrc(self, key: str, data: dict) -> dict:
return self._answer("hmrc.rti_submit", key, data)
def pension(self, key: str, data: dict) -> dict:
return self._answer("pension.submit_contributions", key, data)
def of(self, capability: str) -> list:
return [c for c in self.calls if c[0] == capability]
class YtdReaderFake:
"""The previous provider's last FPS and year-to-date report, as the reader port answers them (no files, no live provider)."""
def __init__(self):
self.sources: dict = {}
self.reads = 0
def read(self, source_ref: str) -> dict:
self.reads += 1
return self.sources[source_ref]
@staticmethod
def source(employees: list, *, fps_over: Optional[dict] = None, tax_year: str = "2026-27", paye_ref: str = "123/AB45678") -> dict:
"""employees: [(payroll_id, tax_code, taxable, tax, ee_ni, er_ni)]; the FPS carries the same figures unless fps_over changes one: {(payroll_id, field): value}."""
over = fps_over or {}
report = [{"payroll_id": p, "tax_code": code, "basis": "cumulative", "taxable": t, "tax": x, "er_ni": er, "pension": "0.00",
"ni": {"A": {"earnings_at_lel": "0.00", "lel_to_pt": "0.00", "pt_to_uel": t, "above_uel": "0.00", "employee": ee, "employer": er}}}
for p, code, t, x, ee, er in employees]
fps = [{"payroll_id": p, "taxable": over.get((p, "taxable"), t), "tax": over.get((p, "tax"), x), "ee_ni": over.get((p, "ee_ni"), ee),
"er_ni": over.get((p, "er_ni"), er)} for p, code, t, x, ee, er in employees]
return {"tax_year": tax_year, "paye_ref": paye_ref, "tax_month": 7, "fps": {"employees": fps}, "report": {"employees": report},
"source_file_key": "raw/payroll/UK01/migration/fps-2026-27-M07.xml"}