"""Tests for checklist section 5.3 — Migrations. Covers (structural, no DB connection required): - All schema changes are versioned Alembic migrations (linear chain, no gaps) - Every migration has a real (non-empty) downgrade() implementation - Migration env.py wraps execution in a transaction (begin_transaction) - Migration script is executable (deploy/migrate-test.sh exists and is +x) - Head revision matches expected constant """ from __future__ import annotations import inspect import os import pathlib import stat import types import pytest from alembic.config import Config from alembic.script import ScriptDirectory _REPO_ROOT = pathlib.Path(__file__).parent.parent _EXPECTED_HEAD_PREFIX = "0028" _EXPECTED_MIGRATION_COUNT = 28 # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _script_dir() -> ScriptDirectory: cfg = Config(str(_REPO_ROOT / "alembic.ini")) cfg.set_main_option("script_location", str(_REPO_ROOT / "alembic")) return ScriptDirectory.from_config(cfg) def _is_stub_downgrade(mod: types.ModuleType) -> bool: """Return True if downgrade() is a pass-only or single-ellipsis stub.""" import ast try: src = inspect.getsource(getattr(mod, "downgrade")) except (OSError, AttributeError): return True # Strip the 'def downgrade...:' line and check what's left lines = [l.strip() for l in src.splitlines() if l.strip() and not l.strip().startswith("def ")] # A stub body is just 'pass', '...', or a docstring with nothing else non_comment = [l for l in lines if not l.startswith("#") and not l.startswith('"""') and not l.startswith("'''")] if not non_comment: return True if len(non_comment) == 1 and non_comment[0] in ("pass", "..."): return True return False # --------------------------------------------------------------------------- # Linear chain / versioning # --------------------------------------------------------------------------- def test_migration_chain_is_linear() -> None: """Migration graph must have exactly one head (no branches).""" heads = _script_dir().get_heads() assert len(heads) == 1, ( f"Expected single-head chain, got {len(heads)} heads: {heads}. " "Resolve the branch before merging." ) def test_migration_count_matches_expected() -> None: """Migration count must equal _EXPECTED_MIGRATION_COUNT. Update _EXPECTED_MIGRATION_COUNT here when adding a new migration. """ revisions = list(_script_dir().walk_revisions()) assert len(revisions) == _EXPECTED_MIGRATION_COUNT, ( f"Expected {_EXPECTED_MIGRATION_COUNT} migrations, found {len(revisions)}. " "Update _EXPECTED_MIGRATION_COUNT in this file." ) def test_head_revision_prefix() -> None: """Head must start with the expected revision prefix.""" heads = _script_dir().get_heads() assert len(heads) == 1 assert heads[0].startswith(_EXPECTED_HEAD_PREFIX), ( f"Expected head starting with '{_EXPECTED_HEAD_PREFIX}', got '{heads[0]}'. " "Update _EXPECTED_HEAD_PREFIX when a new migration is added." ) def test_all_migrations_importable() -> None: """Every migration module must be importable without errors.""" for rev in _script_dir().walk_revisions(): assert rev.module is not None, ( f"Revision {rev.revision} has no module — check for missing file." ) def test_revision_ids_are_sequential_integers() -> None: """Revision IDs must be zero-padded 4-digit integers with no gaps.""" revisions = sorted(_script_dir().walk_revisions(), key=lambda r: r.revision) ids = sorted(int(r.revision[:4]) for r in revisions if r.revision[:4].isdigit()) expected = list(range(1, len(ids) + 1)) assert ids == expected, ( f"Revision IDs are not sequential (no gaps): found {ids}, expected {expected}." ) # --------------------------------------------------------------------------- # Downgrade coverage — every forward migration has a real downgrade # --------------------------------------------------------------------------- def test_all_migrations_have_downgrade_function() -> None: """Every migration module must define a downgrade() function.""" for rev in _script_dir().walk_revisions(): mod = rev.module assert mod is not None assert hasattr(mod, "downgrade"), ( f"Revision {rev.revision} is missing a downgrade() function." ) def test_no_stub_downgrade_implementations() -> None: """No migration may have a pass-only or ellipsis-only downgrade(). A stub downgrade makes rollback a silent no-op — forbidden. """ stubs = [] for rev in _script_dir().walk_revisions(): mod = rev.module if mod is not None and _is_stub_downgrade(mod): stubs.append(rev.revision) assert not stubs, ( f"Migrations with stub (non-functional) downgrade(): {stubs}. " "Implement the actual rollback DDL." ) def test_every_migration_references_tables_in_downgrade() -> None: """Migrations that add a column or table must also reference it in downgrade(). Heuristic: if upgrade() calls op.add_column / op.create_table for table X, downgrade() must reference X (via drop_column / drop_table). Checked by source inspection — not exhaustive, but catches obvious omissions. """ import re _ADD_RE = re.compile(r'op\.(add_column|create_table)\(\s*["\'](\w+)["\']') _DROP_RE = re.compile(r'op\.(drop_column|drop_table)\(\s*["\'](\w+)["\']') violations = [] for rev in _script_dir().walk_revisions(): mod = rev.module if mod is None: continue try: up_src = inspect.getsource(getattr(mod, "upgrade")) down_src = inspect.getsource(getattr(mod, "downgrade")) except (OSError, AttributeError): continue tables_added = {m.group(2) for m in _ADD_RE.finditer(up_src)} tables_dropped = {m.group(2) for m in _DROP_RE.finditer(down_src)} missing = tables_added - tables_dropped if missing: violations.append(f"{rev.revision}: added {missing} but downgrade() doesn't drop them") assert not violations, ( "Migrations with incomplete downgrade():\n" + "\n".join(violations) ) # --------------------------------------------------------------------------- # Transaction safety — env.py wraps migrations in a transaction # --------------------------------------------------------------------------- def test_env_py_uses_begin_transaction() -> None: """alembic/env.py must call context.begin_transaction() for both offline and online runs. This ensures that a failed migration rolls back cleanly instead of leaving the schema in a partially-applied state. """ env_path = _REPO_ROOT / "alembic" / "env.py" src = env_path.read_text() count = src.count("context.begin_transaction()") assert count >= 2, ( f"Expected at least 2 calls to context.begin_transaction() in env.py " f"(one for offline, one for online), found {count}. " "Wrap both run_migrations_offline() and do_run_migrations() in a transaction." ) # --------------------------------------------------------------------------- # Prod-snapshot migration test script # --------------------------------------------------------------------------- def test_migrate_test_script_exists_and_is_executable() -> None: """deploy/migrate-test.sh must exist and be executable. This script is the mechanism for testing migrations against a production data snapshot before applying to production (checklist item 5.3.2). """ script = _REPO_ROOT / "deploy" / "migrate-test.sh" assert script.exists(), ( "deploy/migrate-test.sh is missing. " "This script is required to validate migrations against a prod snapshot." ) mode = script.stat().st_mode assert mode & stat.S_IXUSR, ( "deploy/migrate-test.sh must be executable (chmod +x)." ) def test_migrate_test_script_contains_round_trip() -> None: """deploy/migrate-test.sh must perform an upgrade→downgrade→upgrade round-trip.""" script = (_REPO_ROOT / "deploy" / "migrate-test.sh").read_text() assert "upgrade head" in script, "script must run 'alembic upgrade head'" assert "downgrade" in script, "script must run 'alembic downgrade' step" # Must do upgrade twice (initial + after downgrade round-trip) assert script.count("upgrade head") >= 2, ( "script must upgrade to HEAD twice (initial apply + round-trip after downgrade)" )