gabriel / musehub public
test_musehub_alembic.py python
137 lines 4.8 KB
Raw
sha256:94ef169c149a452bff7c604ded8b280b19bd477c2dabcb56972780b0b784c7aa Merge 'fix/assignee-sigil-inline' into 'dev' — proposal: As… Human 1 day 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 = "0069"
28 _EXPECTED_MIGRATION_COUNT = 69
29
30 # Core tables that the consolidated schema migration (0001) must create
31 _CORE_TABLES = {
32 "musehub_repos",
33 "musehub_identities",
34 }
35
36
37 # ---------------------------------------------------------------------------
38 # Helpers
39 # ---------------------------------------------------------------------------
40
41 def _script_dir() -> ScriptDirectory:
42 """Return a ScriptDirectory for the project migrations."""
43 cfg = Config(str(_REPO_ROOT / "alembic.ini"))
44 cfg.set_main_option("script_location", str(_REPO_ROOT / "alembic"))
45 return ScriptDirectory.from_config(cfg)
46
47
48 # ---------------------------------------------------------------------------
49 # Tests
50 # ---------------------------------------------------------------------------
51
52
53 def test_alembic_migration_chain_has_single_head() -> None:
54 """The migration graph must have exactly one head revision (linear chain)."""
55 heads = _script_dir().get_heads()
56 assert len(heads) == 1, (
57 f"Expected a single-head migration chain, found {len(heads)} heads: {heads}. "
58 "Branching migrations require manual resolution before merging."
59 )
60
61
62 def test_alembic_migration_count() -> None:
63 """The migration chain must contain exactly the expected number of revisions."""
64 revisions = list(_script_dir().walk_revisions())
65 assert len(revisions) == _EXPECTED_MIGRATION_COUNT, (
66 f"Expected {_EXPECTED_MIGRATION_COUNT} migrations, "
67 f"found {len(revisions)}: {[r.revision for r in revisions]}. "
68 "Update _EXPECTED_MIGRATION_COUNT when adding a new migration."
69 )
70
71
72 def test_alembic_head_revision_is_latest() -> None:
73 """The current head revision must match the expected head."""
74 heads = _script_dir().get_heads()
75 assert len(heads) == 1
76 assert heads[0].startswith(_EXPECTED_HEAD), (
77 f"Expected head to start with '{_EXPECTED_HEAD}', got '{heads[0]}'. "
78 "Update _EXPECTED_HEAD when a new migration is added."
79 )
80
81
82 def test_alembic_all_revisions_importable() -> None:
83 """Every migration module can be imported; no NameError or ImportError."""
84 script_dir = _script_dir()
85 revisions = list(script_dir.walk_revisions())
86 assert len(revisions) > 0, "No revisions found — check alembic script_location."
87 for rev in revisions:
88 assert rev.revision is not None
89 assert rev.module is not None, (
90 f"Revision {rev.revision} has no module — the migration file may be missing."
91 )
92
93
94 def test_alembic_schema_migration_defines_core_tables() -> None:
95 """Migration 0001 upgrade() must reference core tables."""
96 script_dir = _script_dir()
97 revisions = list(script_dir.walk_revisions())
98
99 rev = next(
100 (r for r in revisions if r.revision.startswith("0001")),
101 None,
102 )
103 assert rev is not None, "Revision 0001 not found — consolidated schema migration is missing."
104 assert rev.module is not None
105
106 mod: types.ModuleType = rev.module
107 import inspect
108 upgrade_src = inspect.getsource(getattr(mod, "upgrade"))
109
110 missing = {t for t in _CORE_TABLES if t not in upgrade_src}
111 assert not missing, (
112 f"Schema migration 0001 upgrade() does not reference: {missing}. "
113 "Ensure all core tables are created in the migration."
114 )
115
116
117 def test_alembic_schema_migration_downgrade_drops_core_tables() -> None:
118 """Migration 0001 downgrade() must drop core tables."""
119 script_dir = _script_dir()
120 revisions = list(script_dir.walk_revisions())
121
122 rev = next(
123 (r for r in revisions if r.revision.startswith("0001")),
124 None,
125 )
126 assert rev is not None, "Revision 0001 not found."
127 assert rev.module is not None
128
129 mod: types.ModuleType = rev.module
130 import inspect
131 downgrade_src = inspect.getsource(getattr(mod, "downgrade"))
132
133 missing = {t for t in _CORE_TABLES if t not in downgrade_src}
134 assert not missing, (
135 f"Schema migration 0001 downgrade() does not reference: {missing}. "
136 "Ensure all core tables are dropped in the downgrade."
137 )
File History 3 commits
sha256:94ef169c149a452bff7c604ded8b280b19bd477c2dabcb56972780b0b784c7aa Merge 'fix/assignee-sigil-inline' into 'dev' — proposal: As… Human 1 day ago
sha256:6b1949fc2797ca4c1936a637a4cbfec828ef56cf52398a2e74ca3c4f494e728f fix: use wire_bytes not mpack_bytes_raw in compute_object_b… Sonnet 4.6 patch 10 days ago
sha256:4aed3d8601c8dd3ed37074de35f11f4a9699a0a4b99d43727048fd3f8e6fd13d chore: doc sweep, ignore wrangler build state, misc fixes Sonnet 4.6 minor ⚠ 12 days ago