dodil/erp-brick-payroll

DODIL ERP brick: payroll

tests/test_invariants.py

"""The rules that must not be undone (module.yaml tests.invariants, in_code_not_rows), each tested by attempting to break it."""
import pytest
from conftest import (
    ADMIN,
    EMP4410,
    HR_MANAGER,
    MANAGER,
    PAYMENTS_CLIENT,
    PEER,
    SYSTEM,
    api,
    approve,
    calculate,
    must,
    on_day,
    onboard,
    open_run,
    peer_approves_own,
    person,
    refused,
    setup_company,
)

from brick_payroll.application import use_cases


def calculated():
    setup_company()
    run = open_run()["pay_run_id"]
    onboard(run)
    return run


# ── one claim key per pay run or person operation ─────────────────────────────────────────────

def test_a_period_is_paid_by_one_regular_run_and_a_cancelled_period_is_opened_again(nb):
    run = calculated()
    refused(api("POST", "/pay-runs", ADMIN, {"company_code": "UK01", "schedule_code": "M", "period": "2026-11"}), 409, "already pays")
    refused(api("POST", f"/pay-runs/{run}/cancel", MANAGER, {"reason": "wrong pension rows"}), 200)
    again = open_run()["pay_run_id"]
    assert again == "UK01-M-2026-27-M08-R2"
    refused(api("POST", f"/pay-runs/{run}/cancel", MANAGER, {"reason": "again"}), 409, "not possible from cancelled")


def test_two_approvals_of_one_run_leave_one_approval_one_journal_and_one_set_of_payouts(nb):
    run = calculated()
    peer_approves_own(run)
    must(approve(run))
    refused(approve(run, PEER), 409, "not possible from approved")
    refused(api("POST", f"/pay-runs/{run}/payslips/2101-1/approve", PEER), 409)
    with use_cases.repository() as repo:                                   # a racer who loaded the run before the first commit
        with pytest.raises(Exception, match="approve"):
            repo.claims.once("approve", run, use_cases._now(), "racer")
    assert len(nb.journals) == 1 and len(nb.payouts) == 4 and nb.sent_calls == 1


# ── side effects only after the commit, from pending rows ─────────────────────────────────────

def test_a_rolled_back_approval_sends_nothing_and_a_crash_after_commit_is_sent_by_the_next_drain(nb, monkeypatch):
    run = calculated()
    peer_approves_own(run)
    real = use_cases._approval_effects

    def boom(repo, r, today, *rest):
        real(repo, r, today, *rest)
        raise RuntimeError("the bucket went away mid-transaction")
    monkeypatch.setattr(use_cases, "_approval_effects", boom)
    with pytest.raises(RuntimeError):
        approve(run)
    monkeypatch.setattr(use_cases, "_approval_effects", real)
    assert nb.journals == {} and nb.payouts == {} and nb.sent_calls == 0
    assert must(api("GET", f"/pay-runs/{run}", MANAGER))["status"] == "calculated"
    drain = use_cases._drain
    monkeypatch.setattr(use_cases, "_drain", lambda *a, **k: [])          # the process dies right after the commit
    assert must(approve(run))["status"] == "approved"
    assert nb.journals == {} and nb.payouts == {}
    monkeypatch.setattr(use_cases, "_drain", drain)
    must(api("POST", "/outbox/drain", SYSTEM))
    assert len(nb.journals) == 1 and len(nb.payouts) == 4 and nb.sent_calls == 1


# ── states never go backwards ─────────────────────────────────────────────────────────────────

def test_an_approved_run_is_never_recalculated_cancelled_or_its_payslips_reapproved(nb):
    run = calculated()
    peer_approves_own(run)
    must(approve(run))
    refused(calculate(run), 409, "not possible from approved")
    refused(api("POST", f"/pay-runs/{run}/cancel", MANAGER, {"reason": "oops"}), 409, "not possible from approved")
    refused(api("POST", "/payroll-employees/4410-1/details", ADMIN, {"tax_code": "BR", "ni_category": "A", "source": "manual"}), 409,
            "not possible from active")


# ── no self-approval, under either login ──────────────────────────────────────────────────────

def test_the_preparer_cannot_approve_under_a_second_email_with_the_same_sub(nb):
    run = calculated()
    peer_approves_own(run)
    lead = person("a.second.login@uk.example", ["prl:pay_run:approve", "prl:pay_run:read"], "a-1")      # admin A's sub
    alias = person("admin.a@uk.example", ["prl:pay_run:approve", "prl:pay_run:read"], "other-sub")      # admin A's email
    for who in (lead, alias):
        body = refused(approve(run, who), 409)
        assert "second_person" in [g["gate"] for g in body["gates"] if not g["met"]], body


def test_nobody_approves_their_own_payslip_under_a_second_email_with_the_same_sub(nb):
    run = calculated()
    second = person("m.private@uk.example", ["prl:pay_run:approve", "prl:pay_run:read"], "m-1")         # M's sub, another email
    refused(api("POST", f"/pay-runs/{run}/payslips/2101-1/approve", second), 409, "own_payslip_by_peer")
    body = refused(approve(run, second), 409, "own_payslip_by_peer")
    assert body["gate"] == "own_payslip_by_peer"


# ── the business date is in business_timezone ─────────────────────────────────────────────────

def test_the_bacs_lead_is_counted_from_the_london_date_not_the_utc_date(nb):
    setup_company()
    with on_day("2026-09-20T10:00:00+01:00"):
        run = open_run("2026-09")["pay_run_id"]                            # paid Wednesday 2026-09-30
        onboard(run)
        peer_approves_own(run)
    with on_day("2026-09-27T23:30:00+00:00"):                              # 00:30 BST on Monday 28th; UTC still says Sunday 27th
        refused(approve(run), 409, "bacs_lead")                            # Tue, Wed, Thu: past the 30th
    with on_day("2026-09-27T22:30:00+00:00"):                              # 23:30 BST on Sunday 27th
        assert must(approve(run))["approved_on"] == "2026-09-27"


# ── bounds are 422, never 500 ─────────────────────────────────────────────────────────────────

def test_out_of_bounds_values_are_422_with_the_gate_never_a_500(nb):
    run = calculated()
    for amount in ("1.234", "abc", "99999999999999999", "-5.00"):
        refused(api("POST", f"/pay-runs/{run}/payouts/prl:{run}:4410:net/paid", PAYMENTS_CLIENT,
                    {"status": "paid", "paid_on": "2026-11-30", "amount": amount}), 422, "bounds")
    refused(api("POST", f"/pay-runs/{run}/payouts/x/paid", PAYMENTS_CLIENT, {"status": "paid", "paid_on": "30/11/2026", "amount": "1.00"}), 422, "bounds")
    refused(api("POST", "/pay-runs", ADMIN, {"company_code": "UK01", "schedule_code": "M", "period": "2026-13"}), 422, "bounds")
    refused(api("POST", "/pension-schemes", MANAGER, {"company_code": "UK01", "scheme_code": "X", "employee_pct": "101", "employer_pct": "3",
                                                      "valid_from": "2027-04-06"}), 422, "bounds")
    refused(api("POST", "/pension-schemes", MANAGER, {"company_code": "UK01", "scheme_code": "X", "employee_pct": "4", "employer_pct": "2",
                                                      "valid_from": "2027-04-06"}), 422, "minimum_contribution")
    refused(api("POST", "/payroll-employees/4410-1/details", ADMIN, {"tax_code": "12X57", "ni_category": "A", "source": "manual"}), 422, "bounds")
    refused(api("POST", "/paye-schemes", MANAGER, {"company_code": "UK01", "paye_ref": "AB45678", "accounts_office_ref": "123PA00045678"}), 422, "bounds")


def test_a_negative_net_blocks_approval(nb, monkeypatch):
    run = calculated()
    peer_approves_own(run)
    from brick_payroll.domain import pay_run as pr
    gates = pr.approval_gates

    def with_a_negative_net(r, *a):                     # a deduction larger than the pay (e.g. a later slice's recovery line)
        r.payslips[0].net = -r.payslips[0].net
        return gates(r, *a)
    monkeypatch.setattr(pr, "approval_gates", with_a_negative_net)
    body = refused(approve(run), 409, "net_not_negative")
    assert "2001-1: net pay is below 0.00" in body["detail"]


# ── payroll data is visible to payroll roles only ─────────────────────────────────────────────

def test_a_person_without_a_payroll_role_sees_nothing(nb):
    run = calculated()
    peer_approves_own(run)
    must(approve(run))
    with on_day("2026-11-30T09:00:00+00:00"):
        for who in (HR_MANAGER, EMP4410):
            refused(api("GET", "/pay-runs", who), 403)
            refused(api("GET", f"/pay-runs/{run}", who), 404)
            refused(api("GET", f"/pay-runs/{run}/payslips/2101-1", who), 404)
            refused(api("GET", f"/pay-runs/{run}/payslips/2101-1/document", who), 404)
            refused(api("GET", "/payroll-employees", who), 403)
            refused(api("GET", "/payroll-employees/2101-1", who), 403)
        refused(api("GET", "/my/payslips", HR_MANAGER), 403)
        mine = must(api("GET", "/my/payslips", EMP4410))["payslips"]
        assert [p["payroll_id"] for p in mine] == ["4410-1"]                    # only their own


def test_machine_permissions_are_refused_to_every_person_role_and_a_wildcard(nb):
    run = calculated()
    star = person("root@uk.example", ["*:*:*"], "root")
    for who in (ADMIN, MANAGER, star):
        refused(api("POST", "/outbox/drain", who), 403)
        refused(api("POST", f"/pay-runs/{run}/payouts/prl:{run}:4410:net/paid", who, {"status": "paid", "paid_on": "2026-11-30", "amount": "1.00"}), 403)