dodil/erp-brick-manufacturing

DODIL ERP brick: manufacturing

tests/test_schema.py

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

from brick_manufacturing.infrastructure.tables import md

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


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_together_create_the_same_tables_and_columns():
    dirs = sorted(p.name for p in (SRC / "migrations").iterdir())
    assert dirs == sorted(dirs) and dirs[0] == "00001_init"
    cols: dict = {}
    for d in dirs:
        sql = (SRC / "migrations" / d / "up.sql").read_text()
        for name, got in expected_columns(sql).items():
            cols.setdefault(name, []).extend(got)
        for table, col in re.findall(r"ALTER TABLE (\w+) ADD COLUMN (\w+)", sql):
            cols[table].append(col)
        assert (SRC / "migrations" / d / "down.sql").exists(), d
    cols.pop("mfg_schema_migrations", None)
    want = {t.name: [c.name for c in t.columns] for t in md.sorted_tables}
    assert {k: sorted(v) for k, v in cols.items() if k in want} == {k: sorted(v) for k, v in want.items()}