dodil/erp-brick-payroll
DODIL ERP brick: payroll
tests/conftest.py
"""Shared wiring: the real app with storage on a SQLite file (each test its own), personas that carry exactly the permissions
of a default role, the two doors - `api(...)` and `tool(...)` - and in-memory neighbours (tests/fakes.py) whose every call is
checked against the callee's signed spec. Plus the story's company: UK01, PAYE 123/AB45678, the monthly schedule M, the
pension scheme UK01-QS (qualifying earnings, 5% / 3%, net pay arrangement), and 4410 on 36,000.00 a year."""
import base64
import json
import os
import pathlib
from contextlib import contextmanager
from datetime import datetime, timezone
import anyio
import httpx
import httpx2 # the MCP client's HTTP library
import pytest
import yaml
from erp_platform import contract as _contract
from erp_platform.contract import checked_seams
from fakes import Neighbours
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_payroll.app import app
from brick_payroll.application import use_cases
from brick_payroll.infrastructure.tables import Repository, md
HERE = pathlib.Path(__file__).resolve().parent
METHOD_DIR = os.environ.get("ERP_METHOD_DIR") or str(HERE.parents[1] / "erp-method")
if not (pathlib.Path(METHOD_DIR) / "modules" / "shared.yaml").exists():
raise RuntimeError(f"seam contract checks need an erp-method checkout: set ERP_METHOD_DIR (tried {METHOD_DIR})")
_load_callee = _contract.load_callee
def _names_payroll(entry: dict) -> bool:
return any(str(c).strip().split(" ")[0] == "payroll" for c in entry.get("callers") or [])
def _load_with_callers(callee, method_dir=None):
"""hr's HR-56 and HR-58 entries say `with: expenses` and HR-62 `with: time-off`, while their `callers:` list names payroll
(PRL-20, PRL-04); contract.check matches `with` only, so those calls would be reported as not listed for payroll. Read an
entry whose callers name payroll as ours too. (Spec gap, reported: hr v3 should list payroll in `with`.)"""
spec, shared = _load_callee(callee, method_dir)
seams = []
for e in spec.get("seams") or []:
if _names_payroll(e) and "payroll" not in str(e.get("with", "")):
seams.append({**e, "with": "payroll"})
seams.append(e)
return {**spec, "seams": seams}, shared
_contract.load_callee = _load_with_callers
checked_seams = checked_seams(caller="payroll", method_dir=METHOD_DIR) # autouse: any real Seam call
ROLES = yaml.safe_load((HERE.parent / "src" / "brick_payroll" / "roles.yaml").read_text())["roles"]
@pytest.fixture(autouse=True)
def engine(tmp_path, monkeypatch):
e = create_engine(f"sqlite:///{tmp_path / 'bucket.db'}", connect_args={"check_same_thread": False})
md.create_all(e)
with e.begin() as conn:
Repository(conn).seed()
@contextmanager
def repo():
with e.begin() as conn:
yield Repository(conn)
monkeypatch.setattr(use_cases, "repository", repo)
yield e
e.dispose()
def person(email, permissions, sub=None):
payload = base64.urlsafe_b64encode(json.dumps({"permissions": list(permissions)}).encode()).decode()
return {"X-Dodil-User": json.dumps({"sub": sub or email, "email": email, "app_roles": []}),
"X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
def as_role(role, email=None, sub=None):
return person(email or f"{role}@uk.example", ROLES[role], sub)
# The people of the stories: admin A (person 2001) calculates; manager M (2101) approves, peer P (2102) approves M's payslip;
# 4410 is the employee of the worked example; H is an hr manager with no payroll permission at all.
ADMIN = as_role("payroll_admin", "admin.a@uk.example", "a-1")
ADMIN2 = as_role("payroll_admin", "admin.b@uk.example", "b-1") # completes admin A's own record
MANAGER = as_role("payroll_manager", "manager.m@uk.example", "m-1")
PEER = as_role("payroll_manager", "peer.p@uk.example", "p-1")
EMP4410 = as_role("employee", "emp4410@uk.example", "e-4410")
HR_MANAGER = person("hr.h@uk.example", ["hr:employment:read", "hr:person:write"], "h-1")
OWNER, SYSTEM = as_role("owner", "owner@uk.example", "o-1"), as_role("system", "payroll-worker@uk.example", "svc-payroll")
PAYMENTS_CLIENT = as_role("payments_client", "payments-service@uk.example", "svc-payments")
NOBODY = person("nobody@uk.example", [])
async def _call(method, path, headers, body=None):
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://brick") as c:
if method == "GET":
return await c.get(path, headers=headers, params=body)
return await c.request(method, path, headers=headers, json=body)
def api(method, path, headers, body=None):
r = anyio.run(_call, method, path, headers, body)
return r.status_code, r.json()
async def _mcp(headers, fn):
async with app.router.lifespan_context(app):
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), headers=headers, base_url="http://brick") as http:
async with streamable_http_client("http://brick/mcp", http_client=http) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
return await fn(s)
def tool(headers, name, args):
res = anyio.run(_mcp, headers, lambda s: s.call_tool(name, args))
text = res.content[0].text
return ("refused", text.split(": ", 1)[1]) if res.is_error else ("ok", json.loads(text))
def tools(headers):
listed = anyio.run(_mcp, headers, lambda s: s.list_tools())
return sorted(t.name for t in listed.tools)
# ── a frozen business clock, the neighbours, and small builders the tests share ─────────────────
@pytest.fixture(autouse=True)
def clock():
"""Business 'today' is 2026-11-24 (run M08 is calculated), 09:00 UTC = 09:00 in London."""
from erp_platform import work
work.set_clock(at=datetime(2026, 11, 24, 9, 0, tzinfo=timezone.utc), freeze=True)
yield
work.reset_clock()
@contextmanager
def on_day(when):
"""Run a block with the business clock at `when` (an ISO instant with its offset), and put it back."""
from erp_platform import work
before = work.now()
work.set_clock(at=datetime.fromisoformat(when), freeze=True)
try:
yield
finally:
work.set_clock(at=before, freeze=True)
@pytest.fixture(autouse=True)
def nb():
"""hr, time-off, the ledger and payments, in memory. hr holds the payroll people: admin A, managers M and P (on the payroll
themselves), and 4410 on 36,000.00 since 2020."""
from erp_platform import seam
n = Neighbours()
n.salaried(2001, "40000.00", sub="a-1", cost_centre="FIN")
n.salaried(2101, "60000.00", sub="m-1", cost_centre="FIN")
n.salaried(2102, "60000.00", sub="p-1", cost_centre="FIN")
n.salaried(4410, "36000.00", sub="e-4410", bp_id=9410)
for name, a in (("hr", n.hr), ("time-off", n.time_off), ("gl", n.gl), ("payments", n.payments)):
seam.register_local(name, a)
yield n
for name in ("hr", "time-off", "gl", "payments"):
seam.unregister_local(name)
@pytest.fixture
def rti():
"""A provider for hmrc.rti_submit and pension.submit_contributions. Without this fixture there is NO provider: every file is a manual task."""
from fakes import RtiProvider
from brick_payroll.infrastructure import filing
p = RtiProvider()
filing.register("hmrc.rti_submit", p.hmrc)
filing.register("pension.submit_contributions", p.pension)
yield p
filing.unregister("hmrc.rti_submit")
filing.unregister("pension.submit_contributions")
@pytest.fixture
def ytd_reader():
"""A reader of the previous provider's year to date. Without this fixture there is NO reader: the load refuses by name."""
from fakes import YtdReaderFake
from brick_payroll.infrastructure import readers
r = YtdReaderFake()
readers.ports.reader = r
yield r
readers.ports.reader = None
def must(result, status=200):
got, body = result
assert got == status, (got, body)
return body
def refused(result, status=409, words=""):
got, body = result
assert got == status, (got, body)
assert words in json.dumps(body), (words, body)
return body
def setup_company():
"""The PAYE scheme, the monthly schedule and the pension scheme of UK01 (the payroll manager's rows)."""
must(api("POST", "/paye-schemes", MANAGER, {"company_code": "UK01", "paye_ref": "123/AB45678", "accounts_office_ref": "123PA00045678"}))
must(api("POST", "/pay-schedules", MANAGER, {"company_code": "UK01", "schedule_code": "M", "paye_ref": "123/AB45678"}))
must(api("POST", "/pension-schemes", MANAGER, {"company_code": "UK01", "scheme_code": "UK01-QS", "employee_pct": "5", "employer_pct": "3",
"valid_from": "2026-04-06"}), 201)
def open_run(period="2026-11", who=None):
return must(api("POST", "/pay-runs", who or ADMIN, {"company_code": "UK01", "schedule_code": "M", "period": period}), 201)
def calculate(run_id, who=None):
return api("POST", f"/pay-runs/{run_id}/calculate", who or ADMIN)
def details(payroll_id, tax_code="1257L", basis="cumulative", category="A", who=None, **kw):
body = {"tax_code": tax_code, "basis": basis, "ni_category": category, "source": "manual", "valid_from": "2026-04-06", **kw}
return must(api("POST", f"/payroll-employees/{payroll_id}/details", who or ADMIN, body))
def enrol(payroll_id, on="2026-04-06"):
return must(api("POST", f"/payroll-employees/{payroll_id}/pension", ADMIN, {"scheme_code": "UK01-QS", "enrolled_on": on}), 201)
def onboard(run_id, pension=("4410-1",)):
"""A calculation that finds people without tax details: give each 1257L cumulative, category A (pension for those named), and
calculate again."""
got = calculate(run_id)
if got[0] == 409:
for e in must(api("GET", "/payroll-employees", ADMIN, {"status": "pending_details"}))["employees"]:
details(e["payroll_id"], who=ADMIN2 if e["payroll_id"] == "2001-1" else ADMIN)
if e["payroll_id"] in pension:
enrol(e["payroll_id"])
got = calculate(run_id)
return must(got)
def approve(run_id, who=None):
return api("POST", f"/pay-runs/{run_id}/approve", who or MANAGER)
def peer_approves_own(run_id, payroll_id="2101-1", who=None):
return must(api("POST", f"/pay-runs/{run_id}/payslips/{payroll_id}/approve", who or PEER))
def slip(view, payroll_id):
return next(p for p in view["payslips"] if p["payroll_id"] == payroll_id)
def pay_month(period, approve_at):
"""Open, calculate and approve one month (the manager's own payslip approved by the peer first)."""
with on_day(approve_at):
run = open_run(period)["pay_run_id"]
onboard(run)
peer_approves_own(run)
return must(approve(run))