dodil/erp-brick-purchasing

DODIL ERP brick: purchasing

tests/test_schema.py

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

from brick_purchasing.infrastructure.tables import md

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


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} or declared == {t.name: [c.name for c in t.columns] for t in md.tables.values()}


def test_the_migration_creates_the_same_tables_and_drops_them():
    up = expected_columns((SRC / "migrations" / "00001_init" / "up.sql").read_text())
    assert set(up) == {t.name for t in md.tables.values()}
    down = (SRC / "migrations" / "00001_init" / "down.sql").read_text()
    assert all(f"DROP TABLE IF EXISTS {n};" in down for n in up)


def test_every_table_is_prefixed_and_has_a_natural_primary_key_never_serial():
    for t in md.tables.values():
        assert t.name.startswith("pur_"), t.name
        assert len(t.primary_key.columns) >= 1, t.name
        assert not any(c.autoincrement is True for c in t.primary_key.columns), t.name
    sql = (SRC / "schema.sql").read_text()
    assert "SERIAL" not in re.sub(r"--.*", "", sql).upper()


def test_money_is_numeric_18_2_rates_19_4_and_quantities_19_6():
    sql = (SRC / "schema.sql").read_text()
    body = dict(re.findall(r"CREATE TABLE IF NOT EXISTS (\w+) \(\n(.*?)\n\);", sql, re.S))
    assert "unit_price NUMERIC(19, 4)" in body["pur_order_line"] and "net_amount NUMERIC(18, 2)" in body["pur_order_line"]
    assert "quantity_base NUMERIC(19, 6)" in body["pur_order_line"] and "total NUMERIC(18, 2)" in body["pur_bill"]


def test_the_catch_weight_columns_are_reserved_and_null():
    t = md.tables["pur_receipt_line"]
    assert "catch_quantity" in t.c and "catch_unit_code" in t.c


def test_the_dataclasses_are_the_tables():
    """An aggregate's dataclass and its table carry the same scalar fields (the table is generated from the dataclass)."""
    from brick_purchasing.infrastructure import tables
    for spec in (tables.REQUEST, tables.ORDER, tables.RECEIPT, tables.BILL, tables.RUN, tables.PAYMENT, tables.RFQ, tables.AGREEMENT, tables.DOWN_PAYMENT):
        scalars = {f.name for f in dataclasses.fields(spec.cls) if f.name not in ("events",) and not isinstance(getattr(spec.cls, f.name, None), list)}
        cols = {c.name for c in spec.root.c} - {"version", "created_at", "updated_at"}
        assert cols <= scalars, (spec.cls.__name__, cols - scalars)