dodil/erp-platform

tests/test_migrate.py

"""migrate.py's bookkeeping: which schema it reads, and the versions it records.

`record_versions` runs here against SQLite, which supports the same INSERT … ON CONFLICT form: what
is tested is the rule (record, and never let an older version overwrite a newer one), not DataK3.
Every row here is written with all its columns in VALUES, so DataK3's SET-list bug (fixed in 0.2.2)
does not change the outcome.
"""
import sys

import pytest
from sqlalchemy import create_engine


@pytest.fixture
def mig(monkeypatch):
    monkeypatch.setenv("BUCKET", "not-used")
    from erp_platform import db, migrate
    monkeypatch.setattr(db, "engine", create_engine("sqlite://"))
    return migrate


def meta(m) -> dict:
    with m.db.engine.connect() as c:
        return dict(c.exec_driver_sql("SELECT key, value FROM platform_meta").fetchall())


def test_records_the_platform_and_each_brick(mig, monkeypatch):
    monkeypatch.setattr(mig, "PLATFORM_VERSION", "0.1.0")
    mig.record_versions({"quote": "0.3.0", "party": "0.2.0"})
    assert meta(mig) == {"platform_version": "0.1.0", "brick:quote": "0.3.0", "brick:party": "0.2.0"}


def test_a_newer_version_replaces_the_record(mig, monkeypatch):
    monkeypatch.setattr(mig, "PLATFORM_VERSION", "0.1.0")
    mig.record_versions({"quote": "0.3.0"})
    monkeypatch.setattr(mig, "PLATFORM_VERSION", "0.2.0")
    mig.record_versions({"quote": "0.10.0"})            # 0.10 > 0.3: compared as numbers
    assert meta(mig) == {"platform_version": "0.2.0", "brick:quote": "0.10.0"}


def test_a_rollback_never_overwrites_a_newer_record(mig, monkeypatch, capsys):
    monkeypatch.setattr(mig, "PLATFORM_VERSION", "0.2.0")
    mig.record_versions({"quote": "0.4.0"})
    monkeypatch.setattr(mig, "PLATFORM_VERSION", "0.1.0")
    mig.record_versions({"quote": "0.3.0"})
    assert meta(mig) == {"platform_version": "0.2.0", "brick:quote": "0.4.0"}
    assert "a rollback" in capsys.readouterr().out


def test_bricks_sharing_a_bucket_keep_their_own_rows(mig):
    mig.record_versions({"quote": "0.3.0"})
    mig.record_versions({"party": "0.2.0"})
    assert {"brick:quote", "brick:party"} <= set(meta(mig))


def test_a_missing_schema_is_an_error_not_an_empty_migration(mig, monkeypatch, tmp_path, capsys):
    monkeypatch.chdir(tmp_path)
    monkeypatch.setattr(sys, "argv", ["erp-migrate"])
    assert mig.main() == 1
    assert "no schema at" in capsys.readouterr().err


def test_a_malformed_brick_flag_is_refused(mig, monkeypatch, tmp_path, capsys):
    (tmp_path / "schema.sql").write_text("")
    monkeypatch.chdir(tmp_path)
    monkeypatch.setattr(sys, "argv", ["erp-migrate", "--brick", "quote"])
    assert mig.main() == 1
    assert "name=version" in capsys.readouterr().err


def test_comments_are_not_columns(monkeypatch):
    monkeypatch.setenv("BUCKET", "not-used")
    from erp_platform.migrate import expected_columns
    sql = """CREATE TABLE IF NOT EXISTS t (
    id      BIGINT NOT NULL,      -- the key
                                  -- a comment on a line of its own
    name    VARCHAR,
    PRIMARY KEY (id)
);"""
    assert expected_columns(sql) == {"t": ["id", "name"]}


# ── 0.9.11: a bucket's migrate inserts each brick's starting rows, re-run safe ──────────────

def _seeding_brick(monkeypatch, name="seedpkg"):
    import types
    from sqlalchemy import Column, MetaData, String, Table, select
    md = MetaData()
    t = Table(f"{name}_units", md, Column("code", String, primary_key=True))

    class Repository:
        def __init__(self, conn):
            self.conn = conn

        def seed(self):
            have = {r[0] for r in self.conn.execute(select(t.c.code))}
            for code in ("EA", "KG"):
                if code not in have:
                    self.conn.execute(t.insert().values(code=code))
    tables = types.ModuleType(f"{name}.infrastructure.tables")
    Repository.__module__ = tables.__name__
    tables.md, tables.Repository = md, Repository
    for mod in (name, f"{name}.infrastructure"):
        monkeypatch.setitem(sys.modules, mod, types.ModuleType(mod))
    monkeypatch.setitem(sys.modules, tables.__name__, tables)
    return types.SimpleNamespace(name=name, package=name), md, t


def test_seed_inserts_the_starting_rows_once(mig, monkeypatch, capsys):
    from sqlalchemy import select
    member, md, t = _seeding_brick(monkeypatch)
    md.create_all(mig.db.engine)
    assert mig.seed(member) == 0
    assert mig.seed(member) == 0                                    # a re-run adds nothing
    with mig.db.engine.connect() as c:
        assert sorted(r[0] for r in c.execute(select(t.c.code))) == ["EA", "KG"]
    assert "seeded seedpkg's starting rows" in capsys.readouterr().out


def test_a_brick_without_tables_or_seed_is_not_an_error(mig, monkeypatch):
    import types
    assert mig.seed(types.SimpleNamespace(name="nope", package="no_such_brick_pkg")) == 0


def test_compose_migrate_runs_each_bricks_seed_after_its_schema(monkeypatch, tmp_path):
    import shutil
    from erp_platform import compose, gate, migrate
    here = __import__("pathlib").Path(__file__).parent
    sys.path.insert(0, str(here / "scenario_bricks"))
    shutil.copy(here / "scenarios" / "compose.yaml", tmp_path / "compose.yaml")
    calls = []
    monkeypatch.setattr(gate, "enforce", lambda *a, **k: True)
    monkeypatch.setattr(migrate, "migrate_platform", lambda **k: 0)
    monkeypatch.setattr(migrate, "migrate", lambda schema, bricks, **k: calls.append(("schema", *bricks)) or 0)
    monkeypatch.setattr(migrate, "seed", lambda m: calls.append(("seed", m.name)) or 0)
    assert compose.main(["migrate", str(tmp_path / "compose.yaml")]) == 0
    assert calls == [("schema", "orders"), ("seed", "orders"), ("schema", "stock"), ("seed", "stock"),
                     ("schema", "mailer"), ("seed", "mailer")]