102 lines
3.0 KiB
Python
102 lines
3.0 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Check that migration files up to and including 0092 have not been modified.
|
||
|
|
|
||
|
|
On first run (or with --generate), creates a hash file.
|
||
|
|
On subsequent runs, verifies that hashes match.
|
||
|
|
|
||
|
|
Usage:
|
||
|
|
python scripts/check_migration_hashes.py # Verify
|
||
|
|
python scripts/check_migration_hashes.py --generate # Generate/refresh hashes
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import hashlib
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
BASE = Path(__file__).resolve().parent.parent
|
||
|
|
MIGRATIONS_DIR = BASE / "alembic" / "versions"
|
||
|
|
HASH_FILE = BASE / "alembic" / "migration_hashes.txt"
|
||
|
|
MAX_REVISION = 92 # Migrations 0001-0092 must not change
|
||
|
|
|
||
|
|
|
||
|
|
def get_migration_files() -> list[Path]:
|
||
|
|
"""Get all migration files with revision number <= MAX_REVISION."""
|
||
|
|
files = []
|
||
|
|
for f in sorted(MIGRATIONS_DIR.glob("*.py")):
|
||
|
|
# Extract revision number from filename like 0093_fix_...
|
||
|
|
name = f.stem
|
||
|
|
if not name[:4].isdigit():
|
||
|
|
continue
|
||
|
|
rev = int(name[:4])
|
||
|
|
if rev <= MAX_REVISION:
|
||
|
|
files.append(f)
|
||
|
|
return files
|
||
|
|
|
||
|
|
|
||
|
|
def compute_hash(path: Path) -> str:
|
||
|
|
"""Compute SHA256 hash of a file."""
|
||
|
|
h = hashlib.sha256()
|
||
|
|
with open(path, "rb") as f:
|
||
|
|
h.update(f.read())
|
||
|
|
return h.hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def generate_hashes() -> None:
|
||
|
|
"""Generate hash file from current migration files."""
|
||
|
|
files = get_migration_files()
|
||
|
|
lines = []
|
||
|
|
for f in files:
|
||
|
|
h = compute_hash(f)
|
||
|
|
lines.append(f"{h} {f.name}")
|
||
|
|
HASH_FILE.write_text("\n".join(lines) + "\n")
|
||
|
|
print(f"Generated {len(lines)} hashes in {HASH_FILE}")
|
||
|
|
|
||
|
|
|
||
|
|
def verify_hashes() -> int:
|
||
|
|
"""Verify that migration hashes match the stored hashes. Returns 0 on success, 1 on failure."""
|
||
|
|
if not HASH_FILE.exists():
|
||
|
|
print(f"SKIP: No hash file at {HASH_FILE}. Run with --generate first.")
|
||
|
|
return 0 # Don't fail CI if no hash file exists yet
|
||
|
|
|
||
|
|
stored = {}
|
||
|
|
for line in HASH_FILE.read_text().strip().split("\n"):
|
||
|
|
parts = line.split(" ", 1)
|
||
|
|
if len(parts) == 2:
|
||
|
|
stored[parts[1]] = parts[0]
|
||
|
|
|
||
|
|
files = get_migration_files()
|
||
|
|
errors = 0
|
||
|
|
|
||
|
|
for f in files:
|
||
|
|
current_hash = compute_hash(f)
|
||
|
|
if f.name not in stored:
|
||
|
|
print(f"NEW: {f.name} (not in hash file)")
|
||
|
|
errors += 1
|
||
|
|
elif stored[f.name] != current_hash:
|
||
|
|
print(f"CHANGED: {f.name}")
|
||
|
|
errors += 1
|
||
|
|
else:
|
||
|
|
print(f"OK: {f.name}")
|
||
|
|
|
||
|
|
# Check for missing files (in hash file but not on disk)
|
||
|
|
current_names = {f.name for f in files}
|
||
|
|
for name in stored:
|
||
|
|
if name not in current_names:
|
||
|
|
print(f"MISSING: {name}")
|
||
|
|
errors += 1
|
||
|
|
|
||
|
|
if errors > 0:
|
||
|
|
print(f"\nFAILED: {errors} migration(s) changed or missing")
|
||
|
|
return 1
|
||
|
|
else:
|
||
|
|
print(f"\nOK: All {len(files)} migration hashes verified")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
if "--generate" in sys.argv:
|
||
|
|
generate_hashes()
|
||
|
|
else:
|
||
|
|
sys.exit(verify_hashes())
|