dodil/erp-brick-payroll
DODIL ERP brick: payroll
tests/test_stories.py
"""One test per story of slice A, from its Given/When/Then (module.yaml stories). The worked example (PRL-04, PRL-16) is reproduced
exactly: 4410 on 36,000.00, 1257L cumulative, category A, 5% / 3% on qualifying earnings by net pay arrangement, paid the same
every month since April -> net 2,354.24 in November, and the run's journal 3,461.85 each side for 4410 alone."""
import base64
from datetime import date
from conftest import (
ADMIN,
EMP4410,
MANAGER,
PAYMENTS_CLIENT,
SYSTEM,
api,
approve,
calculate,
must,
on_day,
onboard,
open_run,
pay_month,
peer_approves_own,
refused,
setup_company,
slip,
)
APPROVE_DAYS = {"2026-04": "2026-04-22T10:00:00+01:00", "2026-05": "2026-05-20T10:00:00+01:00", "2026-06": "2026-06-22T10:00:00+01:00",
"2026-07": "2026-07-22T10:00:00+01:00", "2026-08": "2026-08-21T10:00:00+01:00", "2026-09": "2026-09-22T10:00:00+01:00",
"2026-10": "2026-10-22T10:00:00+01:00"}
def april_to_october():
for period, at in APPROVE_DAYS.items():
pay_month(period, at)
def test_prl_04_the_worked_example_line_by_line(nb):
"""Given 4410 paid the same every month since April (taxable pay to date at month 7: 20,132.00, tax to date 2,558.80) and starter
4450 from 2026-11-16 at 30,000.00, when the November run is calculated, then every figure of the spec's example."""
setup_company()
april_to_october()
nb.salaried(4450, "30000.00", start="2026-11-16")
run = open_run("2026-11")
assert (run["pay_run_id"], run["pay_date"], run["tax_period"], run["tax_year"]) == ("UK01-M-2026-27-M08", "2026-11-30", 8, "2026-27")
view = onboard(run["pay_run_id"])
p = slip(view, "4410-1")
before = p["to_date"]
assert (p["gross"], p["employee_pension"], p["employer_pension"], p["taxable"]) == ("3000.00", "124.00", "74.40", "2876.00")
detail = p["tax_detail"]
assert detail["pay_to_date"] == "23008.00" and detail["free_pay_to_date"] == "8386.00" and detail["taxable_to_date"] == "14622"
assert detail["tax_to_date_due"] == "2924.40" and detail["tax_before"] == "2558.80" # month 7's figures, from April to October here
assert (p["tax"], p["employee_ni"], p["employer_ni"], p["net"], p["cost_to_employer"]) == ("365.60", "156.16", "387.45", "2354.24", "3461.85")
assert before["taxable"] == "23008.00" and before["tax"] == "2924.40"
lines = {ln["element_code"]: ln for ln in p["lines"]}
assert lines["SALARY"]["amount"] == "3000.00" and lines["SALARY"]["source_kind"] == "hr_extract" and "UK01|2026-11" in lines["SALARY"]["source_ref"]
assert "tax_basic_rate/any@2026-04-06" in lines["TAX"]["rate_rows_used"]
assert "ni_primary_threshold/monthly@2026-04-06" in lines["EE_NI"]["rate_rows_used"]
assert "ae_qualifying_lower/monthly@2026-04-06" in lines["EE_PENSION"]["rate_rows_used"]
s4450 = slip(view, "4450-1")
assert s4450["gross"] == "1309.52" # 2,500.00 x 11 / 21 working days
(ln,) = [x for x in s4450["lines"] if x["element_code"] == "SALARY"]
assert ln["quantity"] == "11.0000" and ln["rate_amount"] == "2500.0000" and "11 of 21 working days" in ln["source_ref"]
again = must(calculate(run["pay_run_id"])) # a pure function: the same payslips
assert [(x["payroll_id"], x["net"], x["tax"]) for x in again["payslips"]] == [(x["payroll_id"], x["net"], x["tax"]) for x in view["payslips"]]
assert again["calculation"] == view["calculation"] + 1
def test_prl_16_the_journal_of_the_run_balances_per_purpose_and_cost_centre_and_a_retry_is_gls_replay(nb):
"""Given run M08 approved, for 4410 alone on SALES-N, when the post runs and runs again after a timeout, then one journal:
debit gross 3,000.00, employer NI 387.45, employer pension 74.40; credit paye_ni_payable 909.21, pension_payable 198.40,
net_pay_payable 2,354.24; 3,461.85 each side; dated the pay date."""
setup_company()
april_to_october()
for pid in (2001, 2101, 2102): # 4410 alone in November, as the story's figures are
nb.last_day[pid] = date(2026, 10, 31)
run = open_run("2026-11")["pay_run_id"]
onboard(run)
got = must(approve(run))
assert got["status"] == "approved" and got["journal_status"] == "posted"
from brick_payroll.infrastructure.seams import journal_id
j = nb.journals[str(journal_id("prl:UK01-M-2026-27-M08", "approve"))]
assert (j["source_module"], j["source_document"], j["event"], j["journal_date"]) == ("payroll", "prl:UK01-M-2026-27-M08", "approve", "2026-11-30")
assert j["lines"] == [{"purpose": "gross_pay", "cost_centre": "SALES-N", "debit": "3000.00"},
{"purpose": "employer_ni", "cost_centre": "SALES-N", "debit": "387.45"},
{"purpose": "employer_pension", "cost_centre": "SALES-N", "debit": "74.40"},
{"purpose": "paye_ni_payable", "credit": "909.21"}, {"purpose": "pension_payable", "credit": "198.40"},
{"purpose": "net_pay_payable", "credit": "2354.24"}]
assert sum(float(x.get("debit", 0)) for x in j["lines"]) == 3461.85
posts = nb.journal_posts
from sqlalchemy import update
from brick_payroll.application import use_cases
from brick_payroll.infrastructure.tables import prl_outbox
with use_cases.repository() as repo: # a timeout: the row was never marked sent
repo.conn.execute(update(prl_outbox).where((prl_outbox.c.target == "gl") & (prl_outbox.c.pay_run_id == run)).values(status="pending"))
must(api("POST", "/outbox/drain", SYSTEM))
assert nb.journal_posts == posts + 1 and len(nb.journals) == 8 # gl's replay: still one journal per run
def test_prl_16_a_closed_period_leaves_the_journal_pending_and_holds_the_payouts(nb):
setup_company()
nb.closed_through = "2026-11-30"
run = open_run("2026-11")["pay_run_id"]
onboard(run)
peer_approves_own(run)
got = must(approve(run))
assert got["status"] == "approved" and got["journal_status"] == "refused" and "period_open" in got["journal_reason"]
assert nb.payouts == {}
nb.closed_through = None
must(api("POST", "/outbox/drain", SYSTEM))
got = must(api("GET", f"/pay-runs/{run}", MANAGER))
assert got["journal_status"] == "posted" and len(nb.payouts) == 4
def test_prl_12_one_payout_per_payslip_keyed_by_run_and_person_and_the_repeat_is_payments_replay(nb):
"""Given run M08 approved with 4410's net 2,354.24 to person partner 9410, when payroll requests the payouts twice, then
payouts keyed prl:<run>:<person_id>:net, batch_ref the run, due on the pay date, clearing net_pay_payable; never for 0.00."""
setup_company()
april_to_october()
run = open_run("2026-11")["pay_run_id"]
onboard(run)
peer_approves_own(run)
must(approve(run))
p = nb.payouts["prl:UK01-M-2026-27-M08:4410:net"]
assert p == {"payout_ref": "prl:UK01-M-2026-27-M08:4410:net", "batch_ref": "UK01-M-2026-27-M08", "payee_bp_id": "9410", "person_id": "4410",
"company_code": "UK01", "currency": "GBP", "amount": "2354.24", "due_on": "2026-11-30", "reference": "SALARY 2026-11",
"source_module": "payroll", "source_document": "UK01-M-2026-27-M08", "clearing_purpose": "net_pay_payable"}
assert len([k for k in nb.payouts if "M08" in k]) == 4
posts = nb.payout_posts
from sqlalchemy import update
from brick_payroll.application import use_cases
from brick_payroll.infrastructure.tables import prl_outbox
with use_cases.repository() as repo:
repo.conn.execute(update(prl_outbox).where((prl_outbox.c.target == "payments") & (prl_outbox.c.pay_run_id == run)).values(status="pending"))
must(api("POST", "/outbox/drain", SYSTEM))
assert nb.payout_posts == posts + 4 and len([k for k in nb.payouts if "M08" in k]) == 4 # payments' replay, nothing new
def test_prl_28_payments_confirms_a_payout_once_and_refuses_another_amount(nb):
setup_company()
run = open_run("2026-11")["pay_run_id"]
onboard(run)
peer_approves_own(run)
must(approve(run))
ref = "prl:UK01-M-2026-27-M08:4410:net"
net = must(api("GET", f"/pay-runs/{run}/payslips/4410-1", MANAGER))["net"]
body = {"status": "paid", "payment_id": "PAY-UK01-2026-000881", "paid_on": "2026-11-30", "amount": net}
first = must(api("POST", f"/pay-runs/{run}/payouts/{ref}/paid", PAYMENTS_CLIENT, body))
assert first["paid_on"] == "2026-11-30" and first["payout_status"] == "paid"
assert must(api("POST", f"/pay-runs/{run}/payouts/{ref}/paid", PAYMENTS_CLIENT, body)) == first
refused(api("POST", f"/pay-runs/{run}/payouts/{ref}/paid", PAYMENTS_CLIENT, {**body, "amount": "1.00"}), 409, "amount_matches")
ref2 = "prl:UK01-M-2026-27-M08:2001:net"
net2 = must(api("GET", f"/pay-runs/{run}/payslips/2001-1", MANAGER))["net"]
back = must(api("POST", f"/pay-runs/{run}/payouts/{ref2}/paid", PAYMENTS_CLIENT,
{"status": "returned", "paid_on": "2026-12-02", "amount": net2, "reason": "account closed"}))
assert back["payout_status"] == "returned" and back["paid_on"] is None and back["payout_reason"] == "account closed"
refused(api("POST", f"/pay-runs/{run}/payouts/{ref}/paid", MANAGER, body), 403, "prl:payout:mark_paid")
def test_prl_06_the_approver_is_not_the_preparer_and_their_own_payslip_is_approved_by_a_peer_first(nb):
"""Given run M08 calculated by admin A with manager M's own payslip in it, when M tries to approve on 2026-11-25, P approves
M's payslip, and M approves the run, then the first try is refused own_payslip_by_peer and the second approves; after the
commit hr's extract is marked sent, the journal posts and the payouts are requested; the payslips are frozen."""
setup_company()
run = open_run("2026-11")["pay_run_id"]
onboard(run)
with on_day("2026-11-25T10:00:00+00:00"):
body = refused(approve(run), 409, "own_payslip_by_peer")
assert [g["gate"] for g in body["gates"] if not g["met"]] == ["own_payslip_by_peer"]
refused(api("POST", f"/pay-runs/{run}/payslips/2101-1/approve", MANAGER), 409, "own_payslip_by_peer") # M cannot approve their own
peer_approves_own(run)
got = must(approve(run))
assert got["status"] == "approved" and got["approved_by"] == "manager.m@uk.example" and got["approved_on"] == "2026-11-25"
assert nb.extract_state["UK01|2026-11"]["state"] == "sent" and nb.extract_state["UK01|2026-11"]["sent_ref"] == run
assert got["journal_status"] == "posted" and len(nb.payouts) == 4
assert {o["target"]: o["status"] for o in got["outbox"]} == {"hr": "sent", "gl": "sent", "payments": "sent", "fps": "sent", "pension": "sent"}
refused(calculate(run), 409, "not possible from approved") # frozen
def test_prl_06_an_admin_cannot_approve_and_a_manager_who_calculated_cannot_either(nb):
from conftest import person
setup_company()
run = open_run("2026-11")["pay_run_id"]
onboard(run)
refused(approve(run, ADMIN), 403, "prl:pay_run:approve")
lead = person("lead.l@uk.example", ["prl:pay_run:calculate", "prl:pay_run:approve", "prl:pay_run:read"], "l-1")
must(calculate(run, lead))
peer_approves_own(run)
body = refused(approve(run, lead), 409, "second_person")
assert "you calculated this run" in body["detail"]
def test_prl_01_a_starter_waits_for_tax_details_and_nobody_completes_their_own_record(nb):
"""Given starter 4450 in hr's extract with a P45 (1257L, pay to date 15,200.00, tax 1,363.40), when the run is calculated and
the admin completes the record, then 4450-1 waits pending_details and the run is refused naming them; the P45 figures count
in the cumulative tax but never in this employer's year to date; nobody completes their own record."""
setup_company()
nb.salaried(4450, "30000.00", start="2026-11-16")
run = open_run("2026-11")["pay_run_id"]
onboard(run) # 4450 (and everyone on the extract) gets its details here
emp = must(api("GET", "/payroll-employees/4450-1", ADMIN))
assert emp["status"] == "active" and emp["tax_codes"][0]["tax_code"] == "1257L"
nb.salaried(4451, "30000.00", start="2026-11-16")
body = refused(calculate(run), 409, "tax_details")
assert "4451-1 has no tax code or NI category" in body["detail"]
assert must(api("GET", "/payroll-employees/4451-1", ADMIN))["status"] == "pending_details"
must(api("POST", "/payroll-employees/4451-1/details", ADMIN,
{"tax_code": "1257L", "ni_category": "A", "source": "p45", "p45_pay_to_date": "15200.00", "p45_tax_to_date": "1363.40"}))
view = must(calculate(run))
s = slip(view, "4451-1")
assert s["gross"] == "1309.52" and s["to_date"]["taxable"] == "1309.52" # P45 pay is not this employer's
# cumulative with the P45 figures: 16,509.52 - 8,386.00 = 8,123 taxable to date, 1,624.60 due, 1,363.40 paid before
assert (s["tax_detail"]["pay_to_date"], s["tax_detail"]["taxable_to_date"], s["tax"]) == ("16509.52", "8123", "261.20")
refused(api("POST", "/payroll-employees/2001-1/details", ADMIN, {"tax_code": "BR", "ni_category": "A", "source": "manual"}), 409, "not_own")
def test_prl_20_the_extract_is_produced_read_and_marked_sent_once_after_approval(nb):
setup_company()
run = open_run("2026-11")
assert run["extract"]["batch_id"] == "UK01|2026-11" and run["extract"]["state"] == "draft"
onboard(run["pay_run_id"])
peer_approves_own(run["pay_run_id"])
must(approve(run["pay_run_id"]))
assert nb.sent_calls == 1 and nb.extract_state["UK01|2026-11"]["sent_ref"] == "UK01-M-2026-27-M08"
def test_prl_08_a_leavers_salary_to_the_last_day_and_holiday_in_lieu(nb):
"""Given leaver 4490 (last day 2026-10-16, 36,000.00, 1257L) with 4.79 days of annual leave owed, when October is calculated,
then salary 3,000.00 x 12 / 22 = 1,636.36, holiday in lieu 4.79 x 138.4615 = 663.23, gross 2,299.59."""
setup_company()
nb.salaried(4490, "36000.00", last_day="2026-10-16")
nb.balances[4490] = "4.79"
with on_day("2026-10-20T10:00:00+01:00"):
run = open_run("2026-10")["pay_run_id"]
view = onboard(run)
s = slip(view, "4490-1")
lines = {ln["element_code"]: ln["amount"] for ln in s["lines"]}
assert (lines["SALARY"], lines["HOLIDAY_IN_LIEU"], s["gross"], s["leaving_date"]) == ("1636.36", "663.23", "2299.59", "2026-10-16")
def test_prl_21_unpaid_leave_is_a_deduction_at_the_day_rate_and_other_statutory_pay_is_refused_by_name(nb):
"""Given 4480 (26,000.00) unpaid on 2026-11-16, then 1 x 100.0000 = 100.00 off gross with the absence named; a sickness needs
statutory paternity pay, which is not calculated - refused by name rather than paid wrongly (SSP and SMP: tests/test_statutory.py)."""
setup_company()
nb.salaried(4480, "26000.00")
nb.unpaid(4480, ["2026-11-16"])
run = open_run("2026-11")["pay_run_id"]
view = onboard(run)
s = slip(view, "4480-1")
(ln,) = [x for x in s["lines"] if x["element_code"] == "UNPAID_DEDUCTION"]
assert (ln["quantity"], ln["rate_amount"], ln["amount"], ln["source_ref"]) == ("1.0000", "100.0000", "-100.00", "UK01-TOF-2026-00200 r1")
assert s["gross"] == "2066.67" # 26,000.00 / 12 = 2,166.67 - 100.00
nb.absences.append({**nb.absences[0], "absence_ref": "UK01-TOF-2026-00201", "pay_treatment": "statutory_paternity"})
refused(calculate(run), 409, "statutory_pay") # SPP and the rest are refused by name, never guessed
def test_prl_25_a_change_after_calculation_marks_the_payslip_stale_and_blocks_approval_until_recalculated(nb):
"""Given M08 calculated, when hr records a pay change for 4410 from 2026-11-16 before approval, then 4410's payslip is flagged
stale with the cause and approval is refused not_stale; recalculated, it is approved."""
setup_company()
run = open_run("2026-11")["pay_run_id"]
onboard(run)
peer_approves_own(run)
nb.pay_change(4410, "2026-11-16", "38400.00")
body = refused(approve(run), 409, "not_stale")
assert "4410-1: changed since the calculation: extract" in body["detail"]
got = must(api("GET", f"/pay-runs/{run}", MANAGER))
assert slip(got, "4410-1")["stale"] is True and [g["met"] for g in got["readiness"] if g["gate"] == "not_stale"] == [False]
view = must(calculate(run))
assert slip(view, "4410-1")["gross"] == "3104.76" and slip(view, "4410-1")["stale"] is False # 3,000 x 10/21 + 3,200 x 11/21
peer_approves_own(run) # a recalculation needs the peer again
assert must(approve(run))["status"] == "approved"
def test_prl_11_my_payslips_from_the_pay_date_only_mine_and_a_payroll_read_is_logged(nb):
setup_company()
run = open_run("2026-11")["pay_run_id"]
onboard(run)
peer_approves_own(run)
must(approve(run))
assert must(api("GET", "/my/payslips", EMP4410))["payslips"] == [] # 2026-11-24: before the pay date
refused(api("GET", f"/pay-runs/{run}/payslips/4410-1", EMP4410), 404)
with on_day("2026-11-30T08:00:00+00:00"):
(mine,) = must(api("GET", "/my/payslips", EMP4410))["payslips"]
assert mine["pay_run_id"] == run and mine["document_key"] == "raw/payroll/UK01/2026-27/payslip/4410/UK01-M-2026-27-M08.pdf"
doc = must(api("GET", f"/pay-runs/{run}/payslips/4410-1/document", EMP4410))
assert base64.b64decode(doc["content_base64"]).startswith(b"%PDF-1.4") and doc["content_type"] == "application/pdf"
refused(api("GET", f"/pay-runs/{run}/payslips/2101-1", EMP4410), 404) # another person's: not found
must(api("GET", f"/pay-runs/{run}/payslips/4410-1/document", ADMIN)) # a payroll admin opens 4410's ...
from brick_payroll.application import use_cases
with use_cases.repository() as repo:
reads = repo.documents.reads(doc["object_key"])
assert [(r["reader"], r["person_id"]) for r in reads] == [("admin.a@uk.example", 4410)] # ... and it is in the access log; 4410's own read is not