scripts/validate_postgresql_migrations.py

#!/usr/bin/env python3
"""Offline contract validation for committed PostgreSQL migrations."""
from __future__ import annotations

import re
import sys
from pathlib import Path

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

from trading.domain.schema import REQUIRED_TABLES, migration_plan  # noqa: E402

FORBIDDEN = (
    "AUTOINCREMENT",
    "PRAGMA ",
    "BEGIN IMMEDIATE",
    "SQLITE_MASTER",
    "RANDOMBLOB(",
    "INSERT OR IGNORE",
)


def validate(root: Path = ROOT) -> list[str]:
    errors: list[str] = []
    try:
        plan = migration_plan("postgresql")
    except Exception as exc:
        return [str(exc)]
    combined = "\n".join(item.path.read_text(encoding="utf-8") for item in plan)
    upper = combined.upper()
    for token in FORBIDDEN:
        if token in upper:
            errors.append(f"PostgreSQL migration contains SQLite-only token: {token}")
    created = {
        match.group(1).lower()
        for match in re.finditer(
            r"\bCREATE\s+TABLE(?:\s+IF\s+NOT\s+EXISTS)?\s+[\"']?([a-zA-Z_][a-zA-Z0-9_]*)",
            combined,
            flags=re.IGNORECASE,
        )
    }
    # schema_migrations is created by the runner before applying migration files.
    missing = sorted((REQUIRED_TABLES - {"schema_migrations"}) - created)
    if missing:
        errors.append(f"PostgreSQL baseline missing required tables: {missing}")
    if "BIGSERIAL" not in upper:
        errors.append("PostgreSQL baseline must use a native generated-key type")
    if "REFERENCES ORDERS(ORDER_ID)" not in upper:
        errors.append("PostgreSQL order child entities must enforce foreign keys")
    for migration in plan:
        text = migration.path.read_text(encoding="utf-8").strip()
        if not text.endswith(";"):
            errors.append(f"migration does not end with a statement terminator: {migration.version}")
    return errors


def main() -> int:
    errors = validate()
    for error in errors:
        print(f"ERROR: {error}")
    print(
        f"postgresql_migrations={'PASS' if not errors else 'FAIL'} "
        f"migrations={len(migration_plan('postgresql')) if not errors else 'unknown'}"
    )
    return 1 if errors else 0


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