dodil/erp-brick-payroll
DODIL ERP brick: payroll
tests/test_ytd.py
"""The year-to-date load (PRL-24): the previous provider's figures loaded through a reader port, reconciled to the penny against its last FPS, signed by
the owner (or discarded), then the first run here calculating cumulatively from them - PRL-04's worked example reached from the opening figures."""
import pathlib
import yaml
from conftest import ADMIN, MANAGER, OWNER, SYSTEM, api, calculate, enrol, must, onboard, open_run, peer_approves_own, refused, setup_company, slip
from fakes import YtdReaderFake
from brick_payroll.application import ytd
from brick_payroll.infrastructure import readers
LOAD = {"run_id": "YTD-2026-27-1", "tax_year": "2026-27", "paye_ref": "123/AB45678", "company_code": "UK01", "schedule_code": "M", "cutover_date": "2026-11-06",
"source_ref": "fps-2026-27-M07", "hr_migration_signed": True}
FIGURES = [("4410-1", "1257L", "20132.00", "2558.80", "1092.96", "2711.15"), ("2001-1", "1257L", "22000.00", "2900.00", "1200.50", "2800.00")]
def load(reader, body=None, who=SYSTEM, figures=FIGURES, **kw):
reader.sources[(body or LOAD)["source_ref"]] = YtdReaderFake.source(figures, **kw)
return api("POST", "/migration/ytd/load", who, body or LOAD)
def test_the_workflow_arrows_are_the_signed_ones():
spec = yaml.safe_load((pathlib.Path(__file__).resolve().parents[2] / "erp-method" / "modules" / "payroll" / "module.yaml").read_text())
wf = next(w for w in spec["workflows"] if w["entity"] == "prl_ytd_run")
assert {t["key"]: (tuple(t["from"]) if isinstance(t["from"], list) else (t["from"],), t["to"]) for t in wf["transitions"]} == ytd.YTD_MOVES
def test_with_no_reader_or_hr_unsigned_the_load_refuses_by_name(nb):
refused(api("POST", "/migration/ytd/load", SYSTEM, LOAD), 409, "no year-to-date reader")
reader = YtdReaderFake()
readers.ports.reader = reader
try:
refused(load(reader, {**LOAD, "hr_migration_signed": False}), 409, "upstream_signed")
refused(load(reader, {**LOAD, "tax_year": "26-27"}), 422, "tax year")
refused(load(reader, {**LOAD, "paye_ref": "999/ZZ99999"}), 409, "source_matches")
finally:
readers.ports.reader = None
def test_prl_24_load_reconcile_sign_then_the_first_run_calculates_from_the_opening_figures(nb, ytd_reader):
"""4410's month 7 figures (taxable 20,132.00, tax 2,558.80) loaded, reconciled and signed: M08 gives taxable 2,876.00, to date 23,008.00, tax to date 2,924.40, so
365.60 this month, and net 2,354.24 - PRL-04's own numbers."""
setup_company()
got = must(load(ytd_reader))
assert (got["status"], got["employees"], got["replay"], got["totals"]["taxable"], got["totals"]["tax"]) == ("loaded", 2, False, "42132.00", "5458.80")
assert must(api("GET", "/migration/ytd/YTD-2026-27-1", OWNER))["openings"][0]["ytd_ref"] == "ytd:2001-1:2026-27"
refused(api("POST", "/migration/ytd/YTD-2026-27-1/sign", OWNER), 409, "loaded") # not reconciled yet
rec = must(api("POST", "/migration/ytd/YTD-2026-27-1/reconcile", SYSTEM))
assert rec["status"] == "reconciled" and all(c["equal"] for c in rec["checks"]) and rec["readiness"][0]["met"] is True
refused(api("POST", "/migration/ytd/YTD-2026-27-1/sign", MANAGER), 403, "prl:migration:sign")
signed = must(api("POST", "/migration/ytd/YTD-2026-27-1/sign", OWNER))
assert (signed["status"], signed["signed_by"]) == ("signed", "owner@uk.example")
emp = must(api("GET", "/payroll-employees/4410-1", MANAGER))
assert emp["status"] == "active" and (emp["tax_codes"][0]["tax_code"], emp["tax_codes"][0]["source"], emp["tax_codes"][0]["valid_from"]) == ("1257L", "manual", "2026-11-06")
assert emp["ni_categories"][0]["category"] == "A"
enrol("4410-1")
run = open_run()["pay_run_id"]
s = slip(onboard(run), "4410-1")
assert (s["gross"], s["taxable"], s["to_date"]["taxable"], s["tax"], s["to_date"]["tax"], s["net"]) == ("3000.00", "2876.00", "23008.00", "365.60", "2924.40", "2354.24")
assert (s["employee_pension"], s["employee_ni"], s["employer_ni"]) == ("124.00", "156.16", "387.45")
def test_loading_twice_writes_nothing_and_a_held_year_is_not_loaded_by_another_run(nb, ytd_reader):
must(load(ytd_reader))
again = must(load(ytd_reader))
assert again["replay"] is True and again["employees"] == 2
refused(load(ytd_reader, {**LOAD, "run_id": "YTD-2026-27-2"}), 409, "ytd_loaded")
assert [x["run_id"] for x in must(api("GET", "/migration/ytd", OWNER))["loads"]] == ["YTD-2026-27-1"]
def test_a_difference_to_the_penny_blocks_signing_and_a_discarded_load_can_be_loaded_again(nb, ytd_reader):
must(load(ytd_reader, fps_over={("4410-1", "tax"): "2558.81"}))
rec = must(api("POST", "/migration/ytd/YTD-2026-27-1/reconcile", SYSTEM))
off = [c for c in rec["checks"] if not c["equal"]]
assert [(c["check"], c["previous"], c["ours"]) for c in off] == [("4410-1 tax to date", "2558.81", "2558.80"), ("total tax to date", "5458.81", "5458.80")]
assert rec["readiness"][0]["met"] is False
refused(api("POST", "/migration/ytd/YTD-2026-27-1/sign", OWNER), 409, "every_total_equal")
refused(api("POST", "/migration/ytd/YTD-2026-27-1/discard", OWNER, {"reason": ""}), 422, "reason")
gone = must(api("POST", "/migration/ytd/YTD-2026-27-1/discard", OWNER, {"reason": "the provider sent a corrected FPS"}))
assert (gone["status"], gone["employees"]) == ("discarded", 0)
refused(api("POST", "/migration/ytd/YTD-2026-27-1/sign", OWNER), 409, "discarded")
must(load(ytd_reader, {**LOAD, "run_id": "YTD-2026-27-2"})) # the rows are free again
assert must(api("GET", "/migration/ytd/YTD-2026-27-2", OWNER))["employees"] == 2
def test_an_unsigned_load_changes_no_tax_and_signing_after_a_calculation_makes_it_stale(nb, ytd_reader):
setup_company()
must(load(ytd_reader))
must(api("POST", "/migration/ytd/YTD-2026-27-1/reconcile", SYSTEM))
run = open_run()["pay_run_id"]
assert slip(onboard(run), "4410-1")["to_date"]["taxable"] == "2876.00" # the opening figures count only once signed
must(api("POST", "/migration/ytd/YTD-2026-27-1/sign", OWNER))
peer_approves_own(run)
refused(api("POST", f"/pay-runs/{run}/approve", MANAGER), 409, "4410-1: changed since the calculation: year_to_date")
again = must(calculate(run))
assert slip(again, "4410-1")["to_date"]["taxable"] == "23008.00" # recalculated: now counted
refused(api("POST", "/migration/ytd/YTD-2026-27-1/discard", OWNER, {"reason": "oops"}), 409, "signed") # signed is final
def test_only_machines_load_and_reconcile_and_only_the_owner_signs(nb, ytd_reader):
refused(load(ytd_reader, who=MANAGER), 403, "prl:migration:load")
must(load(ytd_reader))
refused(api("POST", "/migration/ytd/YTD-2026-27-1/reconcile", ADMIN), 403, "prl:migration:reconcile")
assert must(api("GET", "/migration/ytd/YTD-2026-27-1", MANAGER))["status"] == "loaded"