Spaces:
Sleeping
Sleeping
File size: 1,974 Bytes
2e658e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | #!/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())
|