dodil/erp-brick-payroll
DODIL ERP brick: payroll
tests/test_filing.py
"""The files that go out after an approval - the FPS, the pension contribution file and the EPS - as deterministic documents handed to a capability from
pending rows, never submitted from here (PRL-13, PRL-14, PRL-15, and SC-PRL-01 to its close)."""
import base64
import csv
import hashlib
import io
import xml.etree.ElementTree as ET
from datetime import date
from conftest import (
ADMIN,
EMP4410,
MANAGER,
SYSTEM,
api,
approve,
must,
on_day,
onboard,
open_run,
pay_month,
peer_approves_own,
refused,
setup_company,
)
from brick_payroll.infrastructure import rti as files
RUN = "UK01-M-2026-27-M08"
def approved(nb, at="2026-11-24T10:00:00+00:00"):
"""Run M08 approved with the four people of conftest (4410 in the pension scheme)."""
setup_company()
with on_day(at):
run = open_run()["pay_run_id"]
onboard(run)
peer_approves_own(run)
return must(approve(run))
def subs(run=RUN):
return {s["submission_id"]: s for s in must(api("GET", "/rti-submissions", ADMIN, {"pay_run_id": run}))["submissions"]}
def test_the_fps_is_a_deterministic_document_of_the_committed_run_handed_over_once_and_accepted(nb, rti):
view = approved(nb)
(call,) = rti.of("hmrc.rti_submit")
cap, key, data = call
assert key == f"fps:{RUN}:1" and data["kind"] == "fps" and data["paye_ref"] == "123/AB45678" and data["accounts_office_ref"] == "123PA00045678"
assert (data["tax_year"], data["tax_period"], data["payment_date"]) == ("2026-27", 8, "2026-11-30")
assert data["file_object_key"] == f"raw/payroll/UK01/rti/fps/{RUN}-1.xml"
s = subs()[f"fps:{RUN}:1"]
assert (s["status"], s["correlation_id"], s["sha256"], s["due_on"]) == ("accepted", "CORR-0001", data["sha256"], "2026-11-30")
assert view["status"] == "submitted" # the answer arrives after the commit and moves the run: HMRC accepted the FPS
assert {f["kind"]: f["status"] for f in must(api("GET", f"/pay-runs/{RUN}", MANAGER))["filings"]} == {"fps": "accepted", "pension": "accepted"}
must(api("POST", "/outbox/drain", SYSTEM)) # the drain again sends nothing: the claim is the submission's row
assert len(rti.of("hmrc.rti_submit")) == 1
def test_the_fps_year_to_date_equals_the_payslips_year_to_date_to_the_penny(nb, rti):
approved(nb)
from brick_payroll.application import use_cases
with use_cases.repository() as repo:
doc = repo.documents.get(f"raw/payroll/UK01/rti/fps/{RUN}-1.xml")
root = ET.fromstring(base64.b64decode(doc["content_b64"]))
assert doc["sha256"] == hashlib.sha256(base64.b64decode(doc["content_b64"])).hexdigest() and doc["content_type"] == "application/xml"
employees = {e.get("payrollId"): e for e in root.findall("Employee")}
assert sorted(employees) == ["2001-1", "2101-1", "2102-1", "4410-1"] and root.find("Employer/PayeReference").text == "123/AB45678"
assert (root.find("TaxYear").text, root.find("TaxPeriod").text, root.find("PaymentDate").text) == ("2026-27", "8", "2026-11-30")
got = must(api("GET", f"/pay-runs/{RUN}", MANAGER, {"lines": True}))
for p in got["payslips"]:
e = employees[p["payroll_id"]]
assert e.find("Pay").get("taxablePayToDate") == p["to_date"]["taxable"] and e.find("Tax").get("toDate") == p["to_date"]["tax"]
assert e.find("Tax").get("inPeriod") == p["tax"] and e.find("NIC").get("employeeToDate") == p["to_date"]["ee_ni"] and e.find("NetPay").text == p["net"]
assert e.find("NIC").get("employerInPeriod") == p["employer_ni"] and e.find("Pension").get("employeeInPeriod") == p["employee_pension"]
e4410 = employees["4410-1"]
assert (e4410.find("NINO").text, e4410.find("TaxCode").text, e4410.find("TaxCode").get("basis"), e4410.find("Name/Surname").text) == ("QQ004410C", "1257L", "cumulative", "Example4410")
def test_the_same_run_builds_the_same_bytes(nb):
approved(nb)
from brick_payroll.application import use_cases
with use_cases.repository() as repo:
run = repo.runs.get(RUN, lines=True)
emps = repo.employees.many([p.payroll_id for p in run.payslips])
ident = {p.payroll_id: {"ni_number": None, "legal_first_names": "A", "legal_last_name": "B", "date_of_birth": "1990-01-01",
"home_address": {"line1": "1 High Street", "postcode": "N1 9GU"}} for p in run.payslips}
a = files.fps_xml(run, 1, "123PA00045678", run.payslips, ident, emps)
b = files.fps_xml(run, 1, "123PA00045678", list(reversed(run.payslips)), ident, emps)
assert a == b and b"<Address>" in a and b"<NINO>" not in a # no NI number: HMRC takes the address
assert files.fps_xml(run, 2, "123PA00045678", run.payslips, ident, emps) != a
def test_with_no_provider_every_file_is_a_manual_task_with_its_deadline_and_the_run_closes_when_they_are_done(nb):
"""SC-PRL-01 to its close: approved -> the FPS done by hand -> submitted -> every payout paid -> paid -> the pension file done by hand -> closed."""
view = approved(nb)
assert view["status"] == "approved"
got = subs()
fps, pen = got[f"fps:{RUN}:1"], got["pension:UK01-QS:" + RUN]
assert (fps["status"], pen["status"], fps["due_on"], pen["due_on"]) == ("manual", "manual", "2026-11-30", "2026-12-22")
assert fps["message"].startswith("Send this FPS to HMRC: upload raw/payroll/UK01/rti/fps/") and "by 2026-11-30" in fps["message"]
assert [w["submission_id"] for w in must(api("GET", "/rti-submissions", ADMIN))["submissions"]] == [fps["submission_id"], pen["submission_id"]]
refused(api("POST", f"/rti-submissions/{fps['submission_id']}/done", EMP4410, {"reference": "X"}), 403, "prl:pay_run:calculate")
must(api("POST", f"/rti-submissions/{fps['submission_id']}/done", ADMIN, {"reference": "HMRC-REF-1"}))
assert must(api("GET", f"/pay-runs/{RUN}", MANAGER))["status"] == "submitted"
refused(api("POST", f"/rti-submissions/{fps['submission_id']}/done", ADMIN, {"reference": "HMRC-REF-1"}), 409, "only a manual task")
from conftest import PAYMENTS_CLIENT
for p in must(api("GET", f"/pay-runs/{RUN}", MANAGER))["payslips"]:
must(api("POST", f"/pay-runs/{RUN}/payouts/{p['payout_ref']}/paid", PAYMENTS_CLIENT, {"status": "paid", "paid_on": "2026-11-30", "amount": p["net"], "payment_id": "pay-" + p["payroll_id"]}))
assert must(api("GET", f"/pay-runs/{RUN}", MANAGER))["status"] == "paid"
must(api("POST", f"/rti-submissions/{pen['submission_id']}/done", ADMIN, {"reference": "SCHEME-UPLOAD-9"}))
closed = must(api("GET", f"/pay-runs/{RUN}", MANAGER))
assert closed["status"] == "closed" and closed["filings"][0]["correlation_id"]
assert not must(api("GET", "/rti-submissions", ADMIN))["submissions"] # nothing waiting
def test_a_rejected_fps_keeps_the_run_approved_and_a_corrected_one_is_n_plus_1(nb, rti):
rti.mode = "reject"
view = approved(nb)
s = subs()[f"fps:{RUN}:1"]
assert (s["status"], s["errors"][0]["code"], s["errors"][0]["field"]) == ("rejected", "1234", "TaxCode")
assert must(api("GET", f"/pay-runs/{RUN}", MANAGER))["status"] == "approved" and view["status"] == "approved"
refused(api("POST", f"/pay-runs/{RUN}/fps/resend", EMP4410), 403, "prl:pay_run:calculate")
rti.mode = "accept"
again = must(api("POST", f"/pay-runs/{RUN}/fps/resend", ADMIN))
assert {k: v["status"] for k, v in subs().items() if v["kind"] == "fps"} == {f"fps:{RUN}:1": "rejected", f"fps:{RUN}:2": "accepted"}
assert [x["submission_id"] for x in again["submissions"] if x["kind"] == "fps"] == [f"fps:{RUN}:1", f"fps:{RUN}:2"]
assert must(api("GET", f"/pay-runs/{RUN}", MANAGER))["status"] == "submitted"
refused(api("POST", f"/pay-runs/{RUN}/fps/resend", ADMIN), 409, "only a rejected FPS")
def test_a_provider_that_does_not_answer_is_retried_with_the_same_key_and_acts_once(nb, rti):
rti.mode = "down"
approved(nb)
assert subs()[f"fps:{RUN}:1"]["status"] == "pending" and subs()[f"fps:{RUN}:1"]["attempts"] == 1
assert must(api("GET", f"/pay-runs/{RUN}", MANAGER))["status"] == "approved"
rti.mode = "accept"
must(api("POST", "/outbox/drain", SYSTEM))
assert subs()[f"fps:{RUN}:1"]["status"] == "accepted"
assert {c[1] for c in rti.of("hmrc.rti_submit")} == {f"fps:{RUN}:1"} and len({k for k in rti.answers if k[0] == "hmrc.rti_submit"}) == 1
def test_nothing_is_handed_over_when_the_approval_is_refused(nb, rti):
setup_company()
run = open_run()["pay_run_id"]
onboard(run)
refused(approve(run), 409, "own_payslip_by_peer")
assert rti.calls == [] and not must(api("GET", "/rti-submissions", ADMIN))["submissions"]
def test_the_pension_file_is_one_row_per_member_with_the_run_and_scheme_in_its_key(nb, rti):
approved(nb)
(call,) = rti.of("pension.submit_contributions")
_, key, data = call
assert key == f"pension:UK01-QS:{RUN}" and data["format"] == "papdis" and data["scheme_code"] == "UK01-QS" and data["pay_date"] == "2026-11-30"
assert data["file_object_key"] == f"raw/payroll/UK01/pension/UK01-QS/{RUN}.csv" and data["totals"] == {"employee": "124.00", "employer": "74.40"}
from brick_payroll.application import use_cases
with use_cases.repository() as repo:
doc = repo.documents.get(data["file_object_key"])
assert doc["sha256"] == data["sha256"] and doc["content_type"] == "text/csv"
(row,) = list(csv.DictReader(io.StringIO(base64.b64decode(doc["content_b64"]).decode())))
assert (row["payroll_id"], row["ni_number"], row["surname"], row["pensionable_earnings"], row["employee_contribution"], row["employer_contribution"],
row["enrolment_status"], row["enrolment_date"]) == ("4410-1", "QQ004410C", "Example4410", "2480.00", "124.00", "74.40", "enrolled", "2026-04-06")
assert subs()[key]["status"] == "accepted" and subs()[key]["due_on"] == "2026-12-22"
def test_prl_14_the_eps_recovers_92_percent_of_smp_once_with_its_journal(nb, rti):
"""4431's SMP of 2,423.57 in November: recovered 92% = 2,229.68 on the EPS for tax month 8, sent after the month ends, once; its journal debits
paye_ni_payable and credits statutory_pay_recovery; a month with no pay date sends the no-payment EPS and no second journal."""
setup_company()
nb.salaried(4431, "33800.00", start="2023-03-01")
for month, at in (("2026-06", "2026-06-24T10:00:00+01:00"), ("2026-07", "2026-07-27T10:00:00+01:00")):
pay_month(month, at)
nb.maternity(4431, "2026-11-02", "2026-11-15", "2026-06-01", "2027-10-29")
pay_month("2026-11", "2026-11-24T10:00:00+00:00")
with on_day("2026-12-05T10:00:00+00:00"):
refused(api("POST", "/rti/eps", SYSTEM, {"paye_ref": "123/AB45678", "tax_year": "2026-27", "tax_month": 8}), 409, "month_open")
with on_day("2026-12-06T10:00:00+00:00"):
refused(api("POST", "/rti/eps", MANAGER, {"paye_ref": "123/AB45678", "tax_year": "2026-27", "tax_month": 8}), 403, "prl:rti:submit")
refused(api("POST", "/rti/eps", SYSTEM, {"paye_ref": "123/AB45678", "tax_year": "26-27", "tax_month": 8}), 422, "tax year")
got = must(api("POST", "/rti/eps", SYSTEM, {"paye_ref": "123/AB45678", "tax_year": "2026-27", "tax_month": 8}))
sub = got["submission"]
assert sub["submission_id"] == "eps:123/AB45678:2026-27:M08:1" and sub["status"] == "accepted" and sub["totals"]["ytd_smp_recovered"] == "2229.68"
assert sub["due_on"] == "2026-12-19" and sub["object_key"] == "raw/payroll/UK01/rti/eps/123_AB45678-2026-27-M08-1.xml"
eps_journal = [j for j in nb.journals.values() if j["event"] == "eps"]
assert len(eps_journal) == 1 and eps_journal[0]["lines"] == [{"purpose": "paye_ni_payable", "debit": "2229.68"}, {"purpose": "statutory_pay_recovery", "credit": "2229.68"}]
assert eps_journal[0]["journal_date"] == "2026-12-05" and eps_journal[0]["source_document"] == "prl:eps:123/AB45678:2026-27:M08:1"
again = must(api("POST", "/rti/eps", SYSTEM, {"paye_ref": "123/AB45678", "tax_year": "2026-27", "tax_month": 8}))
assert again["sent"] is False
assert len([c for c in rti.of("hmrc.rti_submit") if c[2]["kind"] == "eps"]) == 1 and len([j for j in nb.journals.values() if j["event"] == "eps"]) == 1
with on_day("2027-01-06T10:00:00+00:00"):
nothing = must(api("POST", "/rti/eps", SYSTEM, {"paye_ref": "123/AB45678", "tax_year": "2026-27", "tax_month": 9}))["submission"]
assert nothing["submission_id"].endswith(":M09:1") and nothing["totals"] == {"ytd_smp_recovered": "2229.68", "added_by_this": "0.00", "no_payment": True}
assert len([j for j in nb.journals.values() if j["event"] == "eps"]) == 1 # nothing new to recover: no second journal
from brick_payroll.application import use_cases
with use_cases.repository() as repo:
eps = ET.fromstring(base64.b64decode(repo.documents.get(sub["object_key"])["content_b64"]))
assert (eps.find("TaxMonth").text, eps.find("NoPaymentForPeriod").text, eps.find("RecoverableAmounts").get("smpRecoveredToDate")) == ("8", "false", "2229.68")
def test_tax_month_dates():
from brick_payroll.application.filing import tax_month_dates
assert tax_month_dates("2026-27", 8) == (date(2026, 11, 6), date(2026, 12, 5))
assert tax_month_dates("2026-27", 1) == (date(2026, 4, 6), date(2026, 5, 5))
assert tax_month_dates("2026-27", 12) == (date(2027, 3, 6), date(2027, 4, 5))
assert files.pension_due(date(2026, 11, 30)) == date(2026, 12, 22) and files.pension_due(date(2026, 12, 31)) == date(2027, 1, 22)