Phase 8.1+8.2: CI Pipeline und Migrations-Release-Gate
8.1 Merge-CI: - Backend Tests und Frontend Tests zu ci_pipeline.sh hinzugefuegt - Migration Hash Check (<=0092) mit check_migration_hashes.py - npm ci --legacy-peer-deps in Forgejo Workflow und ci_pipeline.sh - 93 Migration-Hashes generiert und verifiziert 8.2 Migrations-Release-Gate: - migration_release_gate.sh: Fresh Install, Schema Snapshot, RLS/Grants Check, Cross-Tenant Test, Data Integrity - Prueft leere DB Installation mit Alembic Head + Plugin-Migrationen - Verifiziert RLS >= 100 Tabellen, 4 DB-Rollen, kein BYPASSRLS auf crm_api - Cross-Tenant: 0 rows ohne/fake tenant context
This commit is contained in:
Executable
+101
@@ -0,0 +1,101 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user