scripts/check_migrations.py

#!/usr/bin/env python3
"""Apply all migrations in an isolated database and validate schema invariants."""
from __future__ import annotations

import os
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "hermes_overlay"))

from trading.domain.schema import apply_migrations, connect, list_migration_files, migration_plan, validate_schema  # noqa: E402


def main() -> int:
    with tempfile.TemporaryDirectory(prefix="hermes-migrations-") as tmp:
        os.environ["HERMES_HOME"] = tmp
        ok, messages = apply_migrations()
        if not ok:
            print("migration apply failed:", messages)
            return 1
        ok, messages = validate_schema()
        if not ok:
            print("schema validation failed:", messages)
            return 1
        conn = connect()
        assert conn is not None
        try:
            applied = {row[0] for row in conn.execute("SELECT version FROM schema_migrations")}
        finally:
            conn.close()
        expected = {path.stem for path in list_migration_files()}
        if applied != expected:
            print(f"migration version mismatch: applied={sorted(applied)} expected={sorted(expected)}")
            return 1
        postgres = migration_plan("postgresql")
        forbidden = ("AUTOINCREMENT", "PRAGMA", "RANDOMBLOB", "BEGIN IMMEDIATE", "INSERT OR")
        for migration in postgres:
            sql = migration.path.read_text(encoding="utf-8").upper()
            found = [token for token in forbidden if token in sql]
            if found:
                print(f"postgres migration portability failure {migration.version}: {found}")
                return 1
        print(
            f"migration_check=PASS sqlite_versions={sorted(applied)} "
            f"postgres_versions={[migration.version for migration in postgres]}"
        )
        return 0


if __name__ == "__main__":
    raise SystemExit(main())