dodil/erp-brick-ar

DODIL ERP brick: accounts receivable

tests/test_schema.py

"""schema.sql and the tables in infrastructure/tables.py are one column model: they must not drift (scripts/gen_schema.py writes both)."""
import pathlib
import re
import subprocess
import sys

from brick_ar.infrastructure.tables import ADDED, md

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


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():
    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_first_migration_creates_the_same_tables_and_down_drops_them():
    up = expected_columns((SRC / "migrations" / "00001_init" / "up.sql").read_text())
    later = {n + sfx for a in ADDED.values() for n, spec in a.items() if spec == "*" for sfx in ("", "_versions")}
    assert {t.name for t in md.sorted_tables} - later == set(up)
    down = (SRC / "migrations" / "00001_init" / "down.sql").read_text()
    assert all(f"DROP TABLE IF EXISTS {t.name};" in down for t in md.sorted_tables if t.name not in later)


def test_the_second_migration_adds_exactly_what_the_model_added_after_the_first():
    up = (SRC / "migrations" / "00002_audit1" / "up.sql").read_text()
    for table, spec in ADDED[2].items():
        assert (f"CREATE TABLE IF NOT EXISTS {table} (" in up) if spec == "*" else all(f"ALTER TABLE {table} ADD COLUMN {c} " in up for c in spec)


def test_the_third_migration_adds_exactly_what_the_model_added_in_v4_slice_a():
    up = (SRC / "migrations" / "00003_v4a" / "up.sql").read_text()
    for table, spec in ADDED[3].items():
        assert (f"CREATE TABLE IF NOT EXISTS {table} (" in up) if spec == "*" else all(f"ALTER TABLE {table} ADD COLUMN {c} " in up for c in spec)


def test_the_generator_has_nothing_to_change():
    r = subprocess.run([sys.executable, str(ROOT / "scripts" / "gen_schema.py"), "--check"], capture_output=True, text=True)
    assert r.returncode == 0, r.stdout


def test_every_table_is_prefixed_has_a_natural_key_and_no_serial_or_float():
    for t in md.sorted_tables:
        assert t.name.startswith("ar_")
        assert t.primary_key.columns.keys(), t.name
        for c in t.columns:
            assert c.autoincrement in (False, "auto") and "FLOAT" not in str(c.type).upper() and "SERIAL" not in str(c.type).upper(), (t.name, c.name)
    money = {c.name for t in md.sorted_tables for c in t.columns if c.name in ("subtotal", "tax_total", "total", "net_amount", "tax_amount")}
    assert all(str(c.type) == "NUMERIC(18, 2)" for t in md.sorted_tables for c in t.columns if c.name in money)
    assert str(md.tables["ar_invoice_lines"].c.unit_price.type) == "NUMERIC(19, 4)"


def test_the_bigint_bp_id_not_a_bp_key():
    cols = {c.name: str(c.type) for t in md.sorted_tables for c in t.columns}
    assert "bp_key" not in cols and str(md.tables["ar_invoices"].c.bp_id.type) == "BIGINT"