dodil/erp-brick-timesheets

DODIL ERP brick: timesheets

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_timesheets.domain.values import DEFAULT_ACTIVITIES, DEFAULT_RATES
from brick_timesheets.infrastructure.tables import md

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


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 code_of(sql: str) -> str:
    return "\n".join(ln for ln in sql.splitlines() if not ln.startswith("--"))


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 dirs == ["00001_init", "00002_default_activities_and_rates", "00003_billing_run", "00004_odoo_migration"]
    for d in dirs:
        assert (SRC / "migrations" / d / "up.sql").exists() and (SRC / "migrations" / d / "down.sql").exists()
    up = expected_columns((SRC / "migrations" / "00001_init" / "up.sql").read_text())
    up.update(expected_columns((SRC / "migrations" / "00003_billing_run" / "up.sql").read_text()))
    up.update(expected_columns((SRC / "migrations" / "00004_odoo_migration" / "up.sql").read_text()))
    later = (SRC / "migrations" / "00003_billing_run" / "up.sql").read_text()
    for table, col in re.findall(r"ALTER TABLE (\w+) ADD COLUMN (\w+) ", later):       # an added column is an append: the model keeps it last
        up[table] = [*up[table], col]
    up.pop("ts_schema_migrations")                                                           # the migration journal is the runner's, not a model table
    assert {t.name: [c.name for c in t.columns] for t in md.sorted_tables} == 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 ("ts_billing_run", "ts_migration_run", "ts_migration_map"))     # the later migrations' tables have their own down


def test_the_seeded_defaults_are_the_code_defaults():
    sql = (SRC / "migrations" / "00002_default_activities_and_rates" / "up.sql").read_text()
    acts = re.findall(r"\('\*', '(\w+)', DATE '2000-01-01', '([^']+)', (TRUE|FALSE), (TRUE|FALSE)", sql)
    assert tuple((c, n, b == "TRUE", i == "TRUE") for c, n, b, i in acts) == DEFAULT_ACTIVITIES
    rates = re.findall(r"\('\*', '(\w+)', DATE '2026-01-01', ([0-9.]+), 'GBP'", sql)
    assert tuple(rates) == 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("ts_"), t.name
        assert t.primary_key.columns, t.name
    code = "\n".join(code_of(p.read_text()) for p in [SRC / "schema.sql", *(SRC / "migrations").glob("*/*.sql")])
    assert "SERIAL" not in code.upper() and "UPDATE" not in code.upper().replace("UPDATED_AT", "")
    imports = "\n".join(ln for p in SRC.rglob("*.py") for ln in p.read_text().splitlines() if ln.strip() in ("update,", "from sqlalchemy import update"))
    assert not imports and not re.search(r"\bUPDATE\s+\w+\s+SET", "\n".join(p.read_text() for p in SRC.rglob("*.py")))   # rows move by keyed upsert


def test_money_is_numeric_18_2_and_hours_and_rates_19_4():
    for t in md.sorted_tables:
        for c in t.columns:
            if c.name in ("hours", "unit_cost", "rate_amount", "quantity", "unit_price"):
                assert (c.type.precision, c.type.scale) == (19, 4), (t.name, c.name)


def test_a_week_is_keyed_from_its_root_by_revision():
    keys = {t.name: [c.name for c in t.primary_key.columns] for t in md.sorted_tables}
    assert keys["ts_sheet"] == ["sheet_id", "revision"]
    assert keys["ts_line"] == ["sheet_id", "revision", "line_no"] and keys["ts_entry"] == ["sheet_id", "revision", "line_no", "work_date"]
    assert keys["ts_line_approval"] == ["sheet_id", "revision", "line_no"]
    assert keys["ts_billing_claim"] == ["sheet_id", "revision", "line_no", "purpose"]
    assert keys["ts_cost_rate"] == ["company_code", "rate_group", "valid_from"] and keys["ts_person_rate_group"] == ["person_id", "valid_from"]