dodil/erp-brick-pos

DODIL ERP brick: point of sale

tests/test_schema.py

"""schema.sql, the first migration and the tables in infrastructure/tables.py are one column model: they must not drift. And the
DataK3 rules a table lives by: every table prefixed and keyed on a natural primary key (never SERIAL), no UPDATE ... FROM, money
as NUMERIC(18,2)."""
import pathlib
import re

from sqlalchemy import Numeric, create_engine, inspect

from brick_pos.infrastructure.tables import md

SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_pos"
MIGRATIONS = sorted((SRC / "migrations").glob("*/up.sql"))


def code(path: pathlib.Path) -> str:
    """The SQL without its comment lines."""
    return "\n".join(ln for ln in path.read_text().splitlines() if not ln.lstrip().startswith("--"))


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():
    up = {}
    for m in MIGRATIONS:                                                               # in order: a later release appends columns
        text = m.read_text()
        up.update(expected_columns(text))
        for table, column in re.findall(r"ALTER TABLE (\w+) ADD COLUMN (\w+)", text):
            up[table] = [*up[table], column]
    up.pop("pos_schema_migrations")                                                    # the brick's journal, not a table of the model
    assert up == {t.name: [c.name for c in t.columns] for t in md.sorted_tables}


def test_the_migrations_apply_and_roll_back_on_a_real_engine(tmp_path):
    e = create_engine(f"sqlite:///{tmp_path / 'm.db'}")
    with e.begin() as conn:
        for m in MIGRATIONS:
            for stmt in [s for s in code(m).split(";") if s.strip()]:
                conn.exec_driver_sql(stmt)
    assert {t.name for t in md.sorted_tables} | {"pos_schema_migrations"} == set(inspect(e).get_table_names())
    with e.begin() as conn:
        for m in reversed(MIGRATIONS):
            for stmt in [s for s in code(m.with_name("down.sql")).split(";") if s.strip()]:
                conn.exec_driver_sql(stmt)
    assert inspect(e).get_table_names() == []


def test_every_table_is_prefixed_has_a_natural_key_and_exact_money():
    for t in md.sorted_tables:
        assert t.name.startswith("pos_"), t.name
        assert t.primary_key.columns, t.name
        assert not any(c.autoincrement is True for c in t.columns), t.name
        for c in t.columns:
            if isinstance(c.type, Numeric):
                assert (c.type.precision, c.type.scale) in {(18, 2), (19, 4), (18, 3), (9, 4), (19, 6)}  # (19,6): a recipe component, quantity x quantity, (t.name, c.name)
    for sql in [SRC / "schema.sql", *MIGRATIONS]:
        body = code(sql)
        assert not re.search(r"\b(SERIAL|BIGSERIAL|AUTOINCREMENT|GENERATED)\b", body, re.I), sql
        assert body.count("PRIMARY KEY") == body.count("CREATE TABLE"), sql


def test_no_statement_updates_with_a_join():
    """DataK3 has no UPDATE ... FROM: a keyed write is an upsert, a state move is a version claim."""
    for path in SRC.rglob("*.py"):
        assert not re.search(r"UPDATE\s+\w+\s+SET[^;]*\bFROM\b", path.read_text(), re.I | re.S), path
    for path in SRC.rglob("*.sql"):
        assert not re.search(r"UPDATE\s+\w+\s+SET[^;]*\bFROM\b", path.read_text(), re.I | re.S), path