dodil/erp-brick-payments

DODIL ERP brick: payments

tests/test_schema.py

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

from brick_payments.infrastructure.tables import md

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


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 = expected_columns((SRC / "migrations" / "00001_init" / "up.sql").read_text())
    up.update(expected_columns((SRC / "migrations" / "00002_audit1" / "up.sql").read_text()))
    up.update(expected_columns((SRC / "migrations" / "00003_statements_dd" / "up.sql").read_text()))
    up.update(expected_columns((SRC / "migrations" / "00004_slice_b" / "up.sql").read_text()))
    up.update(expected_columns((SRC / "migrations" / "00005_slice_c" / "up.sql").read_text()))
    up.update(expected_columns((SRC / "migrations" / "00006_v6" / "up.sql").read_text()))
    assert {t.name for t in md.sorted_tables} <= set(up)


def test_every_table_is_prefixed_has_a_primary_key_and_no_serial():
    for t in md.sorted_tables:
        assert t.name.startswith("pay_"), t.name
        assert t.primary_key.columns, t.name
    code = "\n".join(ln for ln in (SRC / "schema.sql").read_text().splitlines() if not ln.startswith("--"))
    assert "SERIAL" not in code.upper()


def test_money_is_numeric_18_2_and_a_rate_numeric_19_4():
    for t in md.sorted_tables:
        for c in t.columns:
            if c.name in ("amount", "closing_balance", "credit_limit_amount", "source_amount"):
                assert (c.type.precision, c.type.scale) == (18, 2), (t.name, c.name)
            if c.name == "fx_rate":
                assert (c.type.precision, c.type.scale) == (19, 4)