dodil/erp-brick-expenses

DODIL ERP brick: expenses

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_expenses.domain.values import DEFAULT_CATEGORIES, DEFAULT_RATES
from brick_expenses.infrastructure.tables import md

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


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}


MIGRATIONS = ["00001_init", "00002_default_categories", "00003_gates", "00004_cards", "00005_fx_vat", "00006_receipts", "00007_rates", "00008_ai", "00009_reminders", "00010_rebill", "00011_migration"]          # appended with every slice; never edited, renamed or dropped


def migrated_columns() -> dict:
    """table -> {column}: every migration's CREATEs and ALTER ... ADD COLUMNs, in order (what a bucket migrated step by step ends with)."""
    out: dict = {}
    for d in MIGRATIONS:
        sql = (SRC / "migrations" / d / "up.sql").read_text()
        for table, cols in expected_columns(sql).items():
            out[table] = set(cols)
        for table, col in re.findall(r"ALTER TABLE (\w+) ADD COLUMN (\w+)", sql):
            out.setdefault(table, set()).add(col)
    return out


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 == MIGRATIONS
    for d in dirs:
        assert (SRC / "migrations" / d / "up.sql").exists() and (SRC / "migrations" / d / "down.sql").exists()
    ours = {t.name: {c.name for c in t.columns} for t in md.sorted_tables}
    got = migrated_columns()
    assert {t: got.get(t) for t in ours} == ours                                   # step by step ends where schema.sql starts
    for d in dirs:                                                                 # every table a migration creates, its down drops
        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 if t != "exp_schema_migrations" or d == "00001_init"), d


def test_the_seeded_categories_are_the_code_defaults():
    rows = []
    for d in MIGRATIONS:
        sql = (SRC / "migrations" / d / "up.sql").read_text()
        for m in re.finditer(r"\('\*', '(\w+)', DATE '([\d-]+)', '([^']+)', '(\w+)', '(\w+)', (TRUE|FALSE), '(\w+)'", sql):
            code, since, name, purpose, tax, receipt, kind = m.groups()
            rows.append((code, name, purpose, tax, receipt == "TRUE", kind, since))
    assert tuple(rows) == tuple(DEFAULT_CATEGORIES)


def test_the_seeded_rates_are_the_code_defaults():
    sql = (SRC / "migrations" / "00007_rates" / "up.sql").read_text()
    rows = []
    for m in re.finditer(r"\('(\w+)', '(\w+)', '(\w+)', (\d+), DATE '([\d-]+)', ([\d.]+), '(\w+)', ([\d.]+|NULL), (NULL|'\w+'), ([\d.]+|NULL), "
                         r"([\d.]+), ([\d.]+), ([\d.]+), '([\d-]+)', '([^']+)', 'migration'", sql):
        kind, country, variant, band, since, amount, cur, upto, unit, partial, b, lu, di, year, source = m.groups()
        n = lambda v: None if v == "NULL" else v      # noqa: E731
        rows.append((kind, country, variant, int(band), since, amount, cur, n(upto), "" if unit == "NULL" else unit.strip("'"), n(partial), b, lu, di, year, source))
    assert tuple(rows) == tuple(DEFAULT_RATES)


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("exp_"), 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():
    for t in md.sorted_tables:
        for c in t.columns:
            if c.name in ("amount", "vat_amount", "receipt_amount", "total_amount", "reimbursable_amount"):
                assert (c.type.precision, c.type.scale) == (18, 2), (t.name, c.name)


def test_reports_and_lines_are_keyed_from_the_root():
    keys = {t.name: [c.name for c in t.primary_key.columns] for t in md.sorted_tables}
    assert keys["exp_report"] == ["report_id"]
    assert keys["exp_line"] == ["report_id", "revision", "line_no"] and keys["exp_approval_step"] == ["report_id", "revision", "level"]