Lamont/tests/unit/test_migration_engine.py

"""Unit tests for migration discovery and history decisions."""

from contextlib import contextmanager
from pathlib import Path
from tempfile import TemporaryDirectory

import pytest

from core import migrations


@pytest.fixture
def migrationRoot():
    with TemporaryDirectory(prefix=".migration-test-", dir=Path.cwd()) as directory:
        yield Path(directory)


def _writeMigration(root, relativePath, content="SELECT 1;"):
    path = root / relativePath
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")
    return path


class FakeCursor:
    def __init__(self, applied=None, historyTable=True):
        self.applied = list(applied or [])
        self.historyTable = historyTable
        self.executed = []
        self._one = None
        self._all = []

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return False

    def execute(self, query, params=None):
        normalized = " ".join(query.split())
        self.executed.append((normalized, params))
        if "SELECT to_regclass" in normalized:
            self._one = {
                "table_name": "schema_migrations" if self.historyTable else None
            }
        elif normalized.startswith("SELECT namespace, version, checksum"):
            self._all = self.applied

    def fetchone(self):
        return self._one

    def fetchall(self):
        return self._all


class FakeConnection:
    def __init__(self, cursor):
        self.activeCursor = cursor

    def cursor(self, **_kwargs):
        return self.activeCursor


def _connectionFor(cursor):
    @contextmanager
    def fakeConnection():
        yield FakeConnection(cursor)

    return fakeConnection


def test_discovery_orders_core_before_feature_namespaces(migrationRoot):
    _writeMigration(migrationRoot, "config/migrations/0002_second.sql", "SELECT 2;")
    _writeMigration(migrationRoot, "config/migrations/0001_first.sql", "SELECT 1;")
    _writeMigration(migrationRoot, "modules/zeta/migrations/0001_zeta.sql")
    _writeMigration(migrationRoot, "modules/alpha/migrations/0002_alpha.sql")

    found = migrations.discover_migrations(migrationRoot)

    assert [(item.namespace, item.version) for item in found] == [
        ("core", 1),
        ("core", 2),
        ("alpha", 2),
        ("zeta", 1),
    ]
    assert len(found[0].checksum) == 64
    assert found[0].path == Path(
        migrationRoot, "config/migrations/0001_first.sql"
    )


def test_discovery_rejects_bad_names_and_duplicate_versions(migrationRoot):
    _writeMigration(migrationRoot, "config/migrations/not-numbered.sql")
    with pytest.raises(migrations.MigrationError, match="Invalid migration filename"):
        migrations.discover_migrations(migrationRoot)

    Path(migrationRoot, "config/migrations/not-numbered.sql").unlink()
    _writeMigration(migrationRoot, "config/migrations/0001_first.sql")
    _writeMigration(migrationRoot, "config/migrations/0001_duplicate.sql")
    with pytest.raises(migrations.MigrationError, match="Duplicate migration"):
        migrations.discover_migrations(migrationRoot)


def test_upgrade_applies_pending_and_skips_matching_history(
    migrationRoot, monkeypatch
):
    _writeMigration(migrationRoot, "config/migrations/0001_first.sql", "SELECT 42;")
    migration = migrations.discover_migrations(migrationRoot)[0]
    cursor = FakeCursor()
    monkeypatch.setattr(migrations.postgres, "get_connection", _connectionFor(cursor))

    assert migrations.upgrade(migrationRoot) == [migration]
    queries = [query for query, _params in cursor.executed]
    assert any("pg_advisory_xact_lock" in query for query in queries)
    assert "SELECT 42;" in queries
    assert any(query.startswith("INSERT INTO schema_migrations") for query in queries)

    cursor = FakeCursor(
        applied=[
            {
                "namespace": "core",
                "version": 1,
                "checksum": migration.checksum,
                "applied_at": None,
            }
        ]
    )
    monkeypatch.setattr(migrations.postgres, "get_connection", _connectionFor(cursor))
    assert migrations.upgrade(migrationRoot) == []
    assert "SELECT 42;" not in [query for query, _params in cursor.executed]


def test_upgrade_rejects_changed_applied_migration(migrationRoot, monkeypatch):
    _writeMigration(migrationRoot, "config/migrations/0001_first.sql", "SELECT 1;")
    cursor = FakeCursor(
        applied=[
            {
                "namespace": "core",
                "version": 1,
                "checksum": "0" * 64,
                "applied_at": None,
            }
        ]
    )
    monkeypatch.setattr(migrations.postgres, "get_connection", _connectionFor(cursor))

    with pytest.raises(migrations.MigrationError, match="checksum changed"):
        migrations.upgrade(migrationRoot)


def test_status_handles_new_database_and_missing_source(migrationRoot, monkeypatch):
    _writeMigration(migrationRoot, "config/migrations/0001_first.sql", "SELECT 1;")
    cursor = FakeCursor(historyTable=False)
    monkeypatch.setattr(migrations.postgres, "get_connection", _connectionFor(cursor))
    assert migrations.migration_status(migrationRoot)[0]["state"] == "pending"

    cursor = FakeCursor(
        applied=[
            {
                "namespace": "removed_feature",
                "version": 3,
                "checksum": "a" * 64,
                "applied_at": "earlier",
            }
        ]
    )
    monkeypatch.setattr(migrations.postgres, "get_connection", _connectionFor(cursor))
    status = migrations.migration_status(migrationRoot)
    assert [item["state"] for item in status] == ["pending", "missing"]
    assert status[1]["namespace"] == "removed_feature"