dodil/erp-brick-payroll

DODIL ERP brick: payroll

tests/test_schema.py

"""schema.sql, the migrations and the tables in infrastructure/tables.py are one column model: they must not drift."""
import pathlib
import re

from brick_payroll.domain.values import DEFAULT_ELEMENTS, DEFAULT_RATES, PERIOD_RATES, STATUTORY_ELEMENTS, STATUTORY_RATES
from brick_payroll.infrastructure.tables import md

SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_payroll"
MONEY_COLS = ("amount", "gross_amount", "taxable_amount", "tax_amount", "ee_ni_amount", "er_ni_amount", "ee_pension_amount",
              "er_pension_amount", "student_loan_amount", "net_amount", "p45_pay_to_date", "p45_tax_to_date", "pay_to_date", "tax_to_date", "taxable_pay_amount", "pension_amount")


def expected_columns(sql: str) -> dict:
    out = {}
    for name, body in re.findall(r"CREATE TABLE IF NOT EXISTS (\w+) \(\n(.*?)\n\);", sql, re.S):
        lines = [ln.strip() for ln in body.splitlines()]
        out[name] = [ln.split()[0] for ln in lines if ln and not ln.startswith(("PRIMARY KEY", "--"))]
    return out


def test_schema_sql_is_the_tables():
    assert expected_columns((SRC / "schema.sql").read_text()) == {t.name: [c.name for c in t.columns] for t in md.sorted_tables}


def test_the_migrations_create_the_same_tables_and_are_versioned_in_order():
    dirs = sorted(p.name for p in (SRC / "migrations").iterdir())
    assert dirs == ["00001_init", "00002_seed_rows", "00003_statutory_rows", "00004_period_rows", "00005_p45", "00006_rti_submission", "00007_ytd"]
    for d in dirs:
        assert (SRC / "migrations" / d / "up.sql").exists() and (SRC / "migrations" / d / "down.sql").exists()
    up, down = {}, ""
    for d in dirs:                                                                    # every table is created by exactly one migration, and dropped by its down
        mine = expected_columns((SRC / "migrations" / d / "up.sql").read_text())
        assert not set(mine) & set(up), (d, set(mine) & set(up))
        up.update(mine)
        down += (SRC / "migrations" / d / "down.sql").read_text()
    assert {t.name: [c.name for c in t.columns] for t in md.sorted_tables}.items() <= up.items()
    assert all(f"DROP TABLE IF EXISTS {t.name};" in down for t in md.sorted_tables)


def test_the_seeded_rows_are_the_code_defaults_and_every_rate_is_flagged_to_confirm():
    sql = (SRC / "migrations" / "00002_seed_rows" / "up.sql").read_text()
    assert re.findall(r"\('\*', '(\w+)', DATE '2000-01-01'", sql) == [e[0] for e in DEFAULT_ELEMENTS]
    rows = re.findall(r"\('(\w+)', '(\w+)', '(\w+)', DATE '2026-04-06', ([0-9.]+), '\w+', '[^']+', TRUE,", sql)
    assert rows == [(k, rg, f, v) for k, rg, f, v, _, _ in DEFAULT_RATES]
    more = (SRC / "migrations" / "00003_statutory_rows" / "up.sql").read_text()
    assert re.findall(r"\('\*', '(\w+)', DATE '2000-01-01'", more) == [e[0] for e in STATUTORY_ELEMENTS]
    held = re.findall(r"\('(\w+)', '(\w+)', '(\w+)', DATE '([0-9-]+)', ([0-9.]+), '\w+', '(?:[^']|'')+', TRUE,", more)
    assert held == [(k, rg, f, vf.isoformat(), v) for k, rg, f, vf, v, _, _ in STATUTORY_RATES]
    periods = (SRC / "migrations" / "00004_period_rows" / "up.sql").read_text()
    got = re.findall(r"\('(\w+)', '(\w+)', '(\w+)', DATE '([0-9-]+)', ([0-9.]+), '\w+', '(?:[^']|'')+', TRUE,", periods)
    assert got == [(k, rg, f, vf.isoformat(), v) for k, rg, f, vf, v, _, _ in PERIOD_RATES]


def test_every_table_is_prefixed_has_a_primary_key_and_no_serial_or_update():
    for t in md.sorted_tables:
        assert t.name.startswith("prl_"), t.name
        assert t.primary_key.columns, t.name
    code = "\n".join(ln for p in [SRC / "schema.sql", *(SRC / "migrations").glob("*/*.sql")] for ln in p.read_text().splitlines() if not ln.startswith("--"))
    assert "SERIAL" not in code.upper() and "UPDATE" not in code.upper().replace("UPDATED_AT", "")


def test_money_is_numeric_18_2_and_rates_19_4():
    for t in md.sorted_tables:
        for c in t.columns:
            if c.name in MONEY_COLS:
                assert (c.type.precision, c.type.scale) == (18, 2), (t.name, c.name)
            if c.name in ("quantity", "rate_amount", "employee_pct", "employer_pct") or (t.name, c.name) == ("prl_statutory_rate", "value"):
                assert (c.type.precision, c.type.scale) == (19, 4), (t.name, c.name)


def test_the_natural_keys_are_the_specs():
    keys = {t.name: [c.name for c in t.primary_key.columns] for t in md.sorted_tables}
    assert keys["prl_pay_run"] == ["pay_run_id"] and keys["prl_payslip"] == ["pay_run_id", "payroll_id"]
    assert keys["prl_period_claim"] == ["company_code", "schedule_code", "tax_year", "period_code", "kind"]
    assert keys["prl_tax_code"] == ["payroll_id", "valid_from"] and keys["prl_statutory_rate"] == ["rate_key", "region", "frequency", "valid_from"]
    assert keys["prl_payslip_line"] == ["pay_run_id", "payroll_id", "calc_no", "line_no"]       # + calc_no: a recalculation deletes nothing