Spaces:
Sleeping
Sleeping
File size: 2,282 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 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | #!/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())
|