gabriel / musehub public
test_musehub_alembic.py python
139 lines 4.9 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 days ago
1 """Alembic migration structural tests.
2
3 Validates the migration chain without executing SQL against a live database.
4 Uses ``ScriptDirectory`` inspection to check the revision graph, head revision,
5 module imports, and CI migration table definitions — all without connecting to
6 a database engine.
7
8 Structural inspection is used (not live SQL) because some migrations use
9 PostgreSQL-specific SQL (``now()``, JSON casts, etc.) and editing applied
10 migrations is forbidden.
11 """
12 from __future__ import annotations
13
14 import importlib
15 import pathlib
16 import types
17
18 from alembic.config import Config
19 from alembic.script import ScriptDirectory
20
21
22 # ---------------------------------------------------------------------------
23 # Constants
24 # ---------------------------------------------------------------------------
25
26 _REPO_ROOT = pathlib.Path(__file__).parent.parent
27 _EXPECTED_HEAD = "0050"
28 _EXPECTED_MIGRATION_COUNT = 50
29
30 # Tables that migration 0006 must create (CI system)
31 _CI_TABLES = {
32 "musehub_ci_runs",
33 "musehub_ci_jobs",
34 "musehub_ci_steps",
35 "musehub_ci_secrets",
36 }
37
38
39 # ---------------------------------------------------------------------------
40 # Helpers
41 # ---------------------------------------------------------------------------
42
43 def _script_dir() -> ScriptDirectory:
44 """Return a ScriptDirectory for the project migrations."""
45 cfg = Config(str(_REPO_ROOT / "alembic.ini"))
46 cfg.set_main_option("script_location", str(_REPO_ROOT / "alembic"))
47 return ScriptDirectory.from_config(cfg)
48
49
50 # ---------------------------------------------------------------------------
51 # Tests
52 # ---------------------------------------------------------------------------
53
54
55 def test_alembic_migration_chain_has_single_head() -> None:
56 """The migration graph must have exactly one head revision (linear chain)."""
57 heads = _script_dir().get_heads()
58 assert len(heads) == 1, (
59 f"Expected a single-head migration chain, found {len(heads)} heads: {heads}. "
60 "Branching migrations require manual resolution before merging."
61 )
62
63
64 def test_alembic_migration_count() -> None:
65 """The migration chain must contain exactly the expected number of revisions."""
66 revisions = list(_script_dir().walk_revisions())
67 assert len(revisions) == _EXPECTED_MIGRATION_COUNT, (
68 f"Expected {_EXPECTED_MIGRATION_COUNT} migrations, "
69 f"found {len(revisions)}: {[r.revision for r in revisions]}. "
70 "Update _EXPECTED_MIGRATION_COUNT when adding a new migration."
71 )
72
73
74 def test_alembic_head_revision_is_drop_social_layer() -> None:
75 """The current head revision must be the drop-social-layer migration (0025)."""
76 heads = _script_dir().get_heads()
77 assert len(heads) == 1
78 assert heads[0].startswith(_EXPECTED_HEAD), (
79 f"Expected head to start with '{_EXPECTED_HEAD}', got '{heads[0]}'. "
80 "Update _EXPECTED_HEAD when a new migration is added."
81 )
82
83
84 def test_alembic_all_revisions_importable() -> None:
85 """Every migration module can be imported; no NameError or ImportError."""
86 script_dir = _script_dir()
87 revisions = list(script_dir.walk_revisions())
88 assert len(revisions) > 0, "No revisions found — check alembic script_location."
89 for rev in revisions:
90 assert rev.revision is not None
91 assert rev.module is not None, (
92 f"Revision {rev.revision} has no module — the migration file may be missing."
93 )
94
95
96 def test_alembic_ci_migration_defines_expected_tables() -> None:
97 """Migration 0006 upgrade() must reference all four CI tables."""
98 script_dir = _script_dir()
99 revisions = list(script_dir.walk_revisions())
100
101 ci_rev = next(
102 (r for r in revisions if r.revision.startswith("0006")),
103 None,
104 )
105 assert ci_rev is not None, "Revision 0006 not found — CI migration is missing."
106 assert ci_rev.module is not None
107
108 mod: types.ModuleType = ci_rev.module
109 import inspect
110 upgrade_src = inspect.getsource(getattr(mod, "upgrade"))
111
112 missing = {t for t in _CI_TABLES if t not in upgrade_src}
113 assert not missing, (
114 f"CI migration 0006 upgrade() does not reference: {missing}. "
115 "Ensure all four CI tables are created in the migration."
116 )
117
118
119 def test_alembic_ci_migration_downgrade_drops_all_tables() -> None:
120 """Migration 0006 downgrade() must drop all four CI tables."""
121 script_dir = _script_dir()
122 revisions = list(script_dir.walk_revisions())
123
124 ci_rev = next(
125 (r for r in revisions if r.revision.startswith("0006")),
126 None,
127 )
128 assert ci_rev is not None, "Revision 0006 not found."
129 assert ci_rev.module is not None
130
131 mod: types.ModuleType = ci_rev.module
132 import inspect
133 downgrade_src = inspect.getsource(getattr(mod, "downgrade"))
134
135 missing = {t for t in _CI_TABLES if t not in downgrade_src}
136 assert not missing, (
137 f"CI migration 0006 downgrade() does not reference: {missing}. "
138 "Ensure all four CI tables are dropped in the downgrade."
139 )
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago