dodil/erp-brick-time-off

DODIL ERP brick: time off

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_time_off.domain.values import DEFAULT_LEAVE_TYPES, DEFAULT_POLICIES
from brick_time_off.infrastructure.tables import md

SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_time_off"


def expected_columns(sql: str) -> dict:
    """table -> [column], as a schema file declares it (erp_platform.migrate's reading, without a BUCKET)."""
    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():
    declared = expected_columns((SRC / "schema.sql").read_text())
    assert declared == {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 [d[:5] for d in dirs] == [f"{n:05d}" for n in range(1, len(dirs) + 1)]                 # appended in order, none skipped
    assert dirs[:2] == ["00001_init", "00002_default_leave_types_and_policies"]
    made: dict = {}
    for d in dirs:
        assert (SRC / "migrations" / d / "up.sql").exists() and (SRC / "migrations" / d / "down.sql").exists()
        up = expected_columns((SRC / "migrations" / d / "up.sql").read_text())
        down = (SRC / "migrations" / d / "down.sql").read_text()
        assert all(f"DROP TABLE IF EXISTS {t};" in down for t in up), d                           # every migration takes back what it made
        assert not made.keys() & up.keys(), d                                                     # a table is made once; a column is a new migration
        made |= up
    made.pop("tof_schema_migrations")                                                           # the migration journal is the runner's, not a model table
    assert made == {t.name: [c.name for c in t.columns] for t in md.sorted_tables}


def test_the_seeded_leave_types_and_policies_are_the_code_defaults():
    sql = (SRC / "migrations" / "00002_default_leave_types_and_policies" / "up.sql").read_text()
    types = re.findall(r"\('\*', '(\w+)', DATE '2000-01-01', '([^']+)', '(\w+)', '(\w+)', '(\w+)', '(\w+)', (\w+),", sql)
    assert tuple((c, n, p, r, a, e, None if m == "NULL" else int(m)) for c, n, p, r, a, e, m in types) == DEFAULT_LEAVE_TYPES
    pols = re.findall(r"\('\*', '([\w-]+)', DATE '2000-01-01', '(\w+)', '(\{[^}]*\})', '(\w+)', (\d+), '(\w+)', 'front_loaded', NULL, '01-01', '(\w+)'", sql)
    assert tuple((c, lt, u, d, bh, rd, ap) for c, lt, ap, u, d, bh, rd in pols) == DEFAULT_POLICIES


def test_every_table_is_prefixed_has_a_primary_key_and_no_serial_or_update_from():
    for t in md.sorted_tables:
        assert t.name.startswith("tof_"), 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_quantities_are_numeric_19_4():
    for t in md.sorted_tables:
        for c in t.columns:
            if c.name in ("cost", "quantity", "stale_cost"):
                assert (c.type.precision, c.type.scale) == (19, 4), (t.name, c.name)


def test_the_day_claim_and_the_ledger_are_keyed_as_the_spec_says():
    keys = {t.name: [c.name for c in t.primary_key.columns] for t in md.sorted_tables}
    assert keys["tof_day_claim"] == ["person_id", "absence_date", "part"]                 # the row a half day is raced for
    assert keys["tof_balance_entry"] == ["entry_ref"] and keys["tof_request"] == ["request_id"]
    assert keys["tof_leaver"] == ["source_ref"] and keys["tof_claim"] == ["claim_scope", "claim_key", "n"]