gabriel / musehub public
test_migrations.py python
973 lines 40.2 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 days ago
1 """Section 36 — Database Migrations / Alembic (7-layer test suite).
2
3 Covers:
4 alembic/versions/0001_consolidated_schema.py → 0019_drop_issue_comment_state_refs.py
5 19 migrations, linear chain (no branches)
6
7 Key design decisions:
8 - Tests require a real PostgreSQL instance (migrations use now(), JSON casts, etc.).
9 The DB URL is:
10 postgresql+asyncpg://musehub:musehub@localhost:5434/musehub_migration_test
11 - Each test creates a FRESH database (drop + create) so tests are fully isolated.
12 - The `migration_db_url` fixture is module-scoped: the fresh DB is created once
13 per module; individual test groups share it but leave the DB at a known
14 revision (HEAD) after each test.
15 - Tests that need the DB at a specific intermediate revision use
16 `_run_to(cfg, rev)` helpers that upgrade/downgrade as needed.
17 - **Never run the full test suite in CI** — these tests connect to the real
18 Postgres container. Run with:
19 python -m pytest tests/test_migrations_section36.py -x -q
20
21 Connection note:
22 The postgres container is exposed at localhost:5434 (mapped from 5432).
23 Credentials: musehub / musehub.
24 """
25 from __future__ import annotations
26
27 import asyncio
28 import os
29 import pathlib
30 import time
31 import uuid
32 from collections.abc import AsyncGenerator, Generator
33 from urllib.parse import urlparse as _urlparse
34
35 import pytest
36 import pytest_asyncio
37 from sqlalchemy import inspect, text
38 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, create_async_engine
39 from sqlalchemy.pool import NullPool
40
41 # Project root: works whether running locally or inside Docker (/app)
42 _PROJECT_ROOT = pathlib.Path(__file__).parent.parent
43
44 # ── constants ─────────────────────────────────────────────────────────────────
45
46 # Derive DB host from DATABASE_URL env var (set inside Docker) or fall back to
47 # the host-mapped port used for local development.
48 _raw_db_url = os.environ.get("DATABASE_URL", "")
49 if _raw_db_url:
50 _p = _urlparse(_raw_db_url)
51 _PG_BASE_URL = f"postgresql+asyncpg://{_p.username}:{_p.password}@{_p.hostname}:{_p.port}"
52 else:
53 _PG_BASE_URL = "postgresql+asyncpg://musehub:musehub@localhost:5434"
54 _TEST_DB = "musehub_migration_test_s36"
55 _TEST_URL = f"{_PG_BASE_URL}/{_TEST_DB}"
56 _ADMIN_URL = f"{_PG_BASE_URL}/musehub" # connect to main DB to create/drop the test DB
57
58 # Separate DB for performance tests so _fresh_db() there doesn't kill migrated_engine
59 _PERF_TEST_DB = "musehub_migration_perf_s36"
60 _PERF_TEST_URL = f"{_PG_BASE_URL}/{_PERF_TEST_DB}"
61
62 _ALL_REVISIONS = [
63 "0001", "0002", "0003", "0004", "0005", "0006", "0007", "0008",
64 "0009", "0010", "0011", "0012", "0013", "0014", "0015", "0016",
65 "0017", "0018", "0019", "0020", "0021", "0022", "0023", "0024",
66 "0025", "0026", "0027", "0028", "0029", "0030", "0031", "0032",
67 "0033", "0034", "0035", "0036", "0037", "0038", "0039", "0040",
68 "0041", "0042", "0043", "0044", "0045", "0046", "0047",
69 "0048", "0049", "0050",
70 ]
71 _HEAD = "0050"
72
73 # Tables that MUST exist after a full upgrade to HEAD
74 # CI tables (musehub_ci_*) are excluded — dropped by migration 0024.
75 # Milestone tables are excluded — dropped by migration 0036.
76 _REQUIRED_TABLES = {
77 "alembic_version",
78 "muse_commits", "muse_objects", "muse_snapshots", "muse_tags",
79 "musehub_auth_challenges", "musehub_auth_keys", "musehub_background_jobs",
80 "musehub_branches", "musehub_collaborators",
81 "musehub_commits", "musehub_coord_records",
82 "musehub_coord_reservations", "musehub_coord_tasks",
83 "musehub_domain_installs", "musehub_domains",
84 "musehub_identities", "musehub_issue_events", "musehub_issues", "musehub_labels",
85 "musehub_object_refs", "musehub_objects", "musehub_proposals", "musehub_releases",
86 "musehub_repos", "musehub_sessions",
87 "musehub_snapshot_entries", "musehub_snapshots", "musehub_symbol_index",
88 "musehub_wire_tags",
89 }
90
91 # Tables that MUST be absent after a full upgrade (dropped by 0016/0036)
92 _LEGACY_TABLES = {
93 "muse_users", "muse_access_tokens", "musehub_profiles",
94 "musehub_issue_milestones", "musehub_milestones",
95 }
96
97
98 # ── helpers ───────────────────────────────────────────────────────────────────
99
100
101 def _run_alembic(db_url: str, *args: str) -> None:
102 """Run an alembic command in a subprocess to avoid the settings lru_cache.
103
104 The in-process lru_cache on musehub.config.settings is populated from
105 conftest.py before DATABASE_URL is set; running in a subprocess guarantees
106 a clean settings instance that picks up our DATABASE_URL.
107 """
108 import subprocess
109 import sys
110
111 env = {
112 "DATABASE_URL": db_url,
113 "MUSE_ENV": "test",
114 "PATH": "/usr/local/bin:/usr/bin:/bin",
115 }
116 result = subprocess.run(
117 [sys.executable, "-m", "alembic"] + list(args),
118 cwd=str(_PROJECT_ROOT),
119 env=env,
120 capture_output=True,
121 text=True,
122 timeout=120,
123 )
124 if result.returncode != 0:
125 raise RuntimeError(
126 f"alembic {' '.join(args)} failed:\n{result.stderr[-2000:]}"
127 )
128
129
130 def _upgrade(db_url: str, revision: str = "head") -> None:
131 _run_alembic(db_url, "upgrade", revision)
132
133
134 def _downgrade(db_url: str, revision: str) -> None:
135 _run_alembic(db_url, "downgrade", revision)
136
137
138 def _current_rev(db_url: str) -> str | None:
139 from alembic.runtime.migration import MigrationContext
140 from sqlalchemy import create_engine
141
142 sync_url = db_url.replace("+asyncpg", "")
143 engine = create_engine(sync_url)
144 with engine.connect() as conn:
145 ctx = MigrationContext.configure(conn)
146 rev = ctx.get_current_revision()
147 engine.dispose()
148 return rev
149
150
151 async def _tables(engine: AsyncEngine) -> set[str]:
152 async with engine.connect() as conn:
153 result = await conn.execute(
154 text(
155 "SELECT tablename FROM pg_tables "
156 "WHERE schemaname='public' ORDER BY tablename"
157 )
158 )
159 return {row[0] for row in result}
160
161
162 async def _indexes_for(engine: AsyncEngine, table: str) -> set[str]:
163 async with engine.connect() as conn:
164 result = await conn.execute(
165 text(
166 "SELECT indexname FROM pg_indexes "
167 "WHERE schemaname='public' AND tablename = :t"
168 ),
169 {"t": table},
170 )
171 return {row[0] for row in result}
172
173
174 async def _columns(engine: AsyncEngine, table: str) -> set[str]:
175 async with engine.connect() as conn:
176 result = await conn.execute(
177 text(
178 "SELECT column_name FROM information_schema.columns "
179 "WHERE table_schema='public' AND table_name = :t"
180 ),
181 {"t": table},
182 )
183 return {row[0] for row in result}
184
185
186 def _fresh_db() -> None:
187 """Drop and recreate the section-36 test database synchronously."""
188 from sqlalchemy import create_engine, text as stext
189
190 sync_admin = _ADMIN_URL.replace("+asyncpg", "")
191 engine = create_engine(sync_admin, isolation_level="AUTOCOMMIT")
192 with engine.connect() as conn:
193 conn.execute(
194 stext(f"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname='{_TEST_DB}'")
195 )
196 conn.execute(stext(f"DROP DATABASE IF EXISTS {_TEST_DB}"))
197 conn.execute(stext(f"CREATE DATABASE {_TEST_DB}"))
198 engine.dispose()
199
200
201 def _perf_fresh_db() -> None:
202 """Drop and recreate the performance test database (separate from _TEST_DB)."""
203 from sqlalchemy import create_engine, text as stext
204
205 sync_admin = _ADMIN_URL.replace("+asyncpg", "")
206 engine = create_engine(sync_admin, isolation_level="AUTOCOMMIT")
207 with engine.connect() as conn:
208 conn.execute(
209 stext(f"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname='{_PERF_TEST_DB}'")
210 )
211 conn.execute(stext(f"DROP DATABASE IF EXISTS {_PERF_TEST_DB}"))
212 conn.execute(stext(f"CREATE DATABASE {_PERF_TEST_DB}"))
213 engine.dispose()
214
215
216 # ── module-scoped engine fixture ──────────────────────────────────────────────
217
218
219 @pytest.fixture(scope="module")
220 def migrated_engine() -> Generator[AsyncEngine, None, None]:
221 """Create a fresh DB, run all migrations to HEAD, yield an async engine.
222
223 Module-scoped: created once for all tests in this file.
224 The DB is left at HEAD after each test (tests that downgrade must re-upgrade).
225 """
226 _fresh_db()
227 _upgrade(_TEST_URL)
228 engine = create_async_engine(_TEST_URL, poolclass=NullPool)
229 yield engine
230 asyncio.run(engine.dispose())
231
232
233 # ══════════════════════════════════════════════════════════════════════════════
234 # 1. Unit
235 # ══════════════════════════════════════════════════════════════════════════════
236
237
238 class TestMigrationUnit:
239 """Static analysis of migration files — no DB connection needed."""
240
241 def test_revision_chain_is_linear(self) -> None:
242 """Every migration (except 0001) has a down_revision pointing to its predecessor."""
243 import importlib.util
244
245 versions_dir = _PROJECT_ROOT / "alembic" / "versions"
246 revisions = {}
247 for f in versions_dir.glob("*.py"):
248 spec = importlib.util.spec_from_file_location(f.stem, f)
249 assert spec and spec.loader
250 mod = importlib.util.module_from_spec(spec)
251 assert spec.loader is not None
252 spec.loader.exec_module(mod)
253 rev = getattr(mod, "revision", None)
254 down = getattr(mod, "down_revision", None)
255 if rev:
256 revisions[str(rev)] = str(down) if down else None
257
258 # 0001 has no predecessor
259 assert revisions.get("0001") is None
260
261 # Every other migration's down_revision must exist as a revision
262 for rev, down in revisions.items():
263 if down is not None:
264 assert down in revisions, (
265 f"Migration {rev}: down_revision '{down}' not found"
266 )
267
268 def test_all_47_migrations_present(self) -> None:
269 versions_dir = _PROJECT_ROOT / "alembic" / "versions"
270 files = {f.stem for f in versions_dir.glob("*.py") if not f.stem.startswith("__")}
271 assert len(files) == 50, f"Expected 50 migrations, found {len(files)}: {files}"
272
273 def test_head_revision_is_0047(self) -> None:
274 versions_dir = _PROJECT_ROOT / "alembic" / "versions"
275 # The head is the revision not referenced as any down_revision
276 import importlib.util
277
278 all_revs: set[str] = set()
279 down_revs: set[str] = set()
280 for f in versions_dir.glob("*.py"):
281 spec = importlib.util.spec_from_file_location(f.stem, f)
282 assert spec and spec.loader
283 mod = importlib.util.module_from_spec(spec)
284 assert spec.loader is not None
285 spec.loader.exec_module(mod)
286 rev = getattr(mod, "revision", None)
287 down = getattr(mod, "down_revision", None)
288 if rev:
289 all_revs.add(str(rev))
290 if down:
291 down_revs.add(str(down))
292 heads = all_revs - down_revs
293 assert heads == {"0050"}, f"Expected head 0050, got: {heads}"
294
295 def test_each_migration_has_upgrade_and_downgrade(self) -> None:
296 import importlib.util
297
298 versions_dir = _PROJECT_ROOT / "alembic" / "versions"
299 for f in versions_dir.glob("*.py"):
300 spec = importlib.util.spec_from_file_location(f.stem, f)
301 assert spec and spec.loader
302 mod = importlib.util.module_from_spec(spec)
303 assert spec.loader is not None
304 spec.loader.exec_module(mod)
305 assert callable(getattr(mod, "upgrade", None)), (
306 f"{f.name} missing upgrade()"
307 )
308 assert callable(getattr(mod, "downgrade", None)), (
309 f"{f.name} missing downgrade()"
310 )
311
312 def test_alembic_ini_points_to_correct_script_location(self) -> None:
313 import configparser
314
315 parser = configparser.ConfigParser()
316 parser.read(str(_PROJECT_ROOT / "alembic.ini"))
317 script_location = parser.get("alembic", "script_location", fallback="")
318 assert "alembic" in script_location
319
320 def test_env_py_imports_all_model_bases(self) -> None:
321 env_text = (_PROJECT_ROOT / "alembic" / "env.py").read_text()
322 assert "from musehub.db.database import Base" in env_text
323 assert "target_metadata = Base.metadata" in env_text
324
325
326 # ══════════════════════════════════════════════════════════════════════════════
327 # 2. Integration
328 # ══════════════════════════════════════════════════════════════════════════════
329
330
331 class TestMigrationIntegration:
332 """Real DB — run migrations and inspect schema state."""
333
334 def test_upgrade_to_head_succeeds(self) -> None:
335 """Full migration chain from empty DB to HEAD completes without error."""
336 _fresh_db()
337 _upgrade(_TEST_URL)
338 rev = _current_rev(_TEST_URL)
339 assert rev == _HEAD
340
341 def test_current_revision_tracked_in_alembic_version(
342 self, migrated_engine: AsyncEngine
343 ) -> None:
344 async def _check() -> str | None:
345 async with migrated_engine.connect() as conn:
346 result = await conn.execute(text("SELECT version_num FROM alembic_version"))
347 row = result.fetchone()
348 return row[0] if row else None
349
350 rev = asyncio.run(_check())
351 assert rev == _HEAD
352
353 def test_required_tables_exist_at_head(
354 self, migrated_engine: AsyncEngine
355 ) -> None:
356 tables = asyncio.run(_tables(migrated_engine))
357 missing = _REQUIRED_TABLES - tables
358 assert not missing, f"Tables missing after upgrade to HEAD: {missing}"
359
360 def test_legacy_tables_absent_at_head(
361 self, migrated_engine: AsyncEngine
362 ) -> None:
363 tables = asyncio.run(_tables(migrated_engine))
364 present_legacy = _LEGACY_TABLES & tables
365 assert not present_legacy, f"Legacy tables still present after HEAD: {present_legacy}"
366
367 def test_downgrade_then_upgrade_returns_to_head(self) -> None:
368 """Downgrade one step and re-upgrade — must return to HEAD cleanly."""
369 _downgrade(_TEST_URL, "0018")
370 assert _current_rev(_TEST_URL) == "0018"
371 _upgrade(_TEST_URL, "head")
372 assert _current_rev(_TEST_URL) == _HEAD
373
374 def test_full_downgrade_to_base_then_re_upgrade(self) -> None:
375 """Downgrade all the way to base then re-upgrade — round trip works."""
376 _downgrade(_TEST_URL, "base")
377 assert _current_rev(_TEST_URL) is None
378 _upgrade(_TEST_URL)
379 assert _current_rev(_TEST_URL) == _HEAD
380
381
382 # ══════════════════════════════════════════════════════════════════════════════
383 # 3. End-to-End
384 # ══════════════════════════════════════════════════════════════════════════════
385
386
387 class TestMigrationE2E:
388 """Full stack: migrate, insert data, downgrade, re-upgrade, verify data."""
389
390 def test_repo_data_survives_incremental_upgrade(
391 self, migrated_engine: AsyncEngine
392 ) -> None:
393 """Insert a repo row at HEAD, downgrade one step, re-upgrade — row survives."""
394
395 async def _run() -> None:
396 repo_id = str(uuid.uuid4())
397 async with migrated_engine.connect() as conn:
398 await conn.execute(
399 text(
400 "INSERT INTO musehub_repos (repo_id, name, slug, owner, owner_user_id, "
401 "visibility, default_branch) "
402 "VALUES (:id, :n, :s, :o, :oid, 'public', 'main')"
403 ),
404 {"id": repo_id, "n": "e2e-repo", "s": "e2e-repo", "o": "testuser", "oid": "testuser"},
405 )
406 await conn.commit()
407
408 _downgrade(_TEST_URL, "0018")
409 _upgrade(_TEST_URL)
410
411 async with migrated_engine.connect() as conn:
412 result = await conn.execute(
413 text("SELECT repo_id FROM musehub_repos WHERE repo_id = :id"),
414 {"id": repo_id},
415 )
416 row = result.fetchone()
417 assert row is not None, "Repo row lost after downgrade/re-upgrade cycle"
418
419 asyncio.run(_run())
420
421 def test_identity_row_survives_upgrade_chain(
422 self, migrated_engine: AsyncEngine
423 ) -> None:
424 """Insert an identity at HEAD; must still exist after a single-step cycle."""
425
426 async def _run() -> None:
427 identity_id = str(uuid.uuid4())
428 async with migrated_engine.connect() as conn:
429 await conn.execute(
430 text(
431 "INSERT INTO musehub_identities (identity_id, handle, display_name, identity_type) "
432 "VALUES (:id, :h, :dn, 'human')"
433 ),
434 {"id": identity_id, "h": f"e2e-user-{identity_id[:8]}", "dn": "E2E User"},
435 )
436 await conn.commit()
437
438 _downgrade(_TEST_URL, "0018")
439 _upgrade(_TEST_URL)
440
441 async with migrated_engine.connect() as conn:
442 result = await conn.execute(
443 text("SELECT identity_id FROM musehub_identities WHERE identity_id = :id"),
444 {"id": identity_id},
445 )
446 row = result.fetchone()
447 assert row is not None
448
449 asyncio.run(_run())
450
451 def test_each_migration_step_is_individually_runnable(self) -> None:
452 """Upgrade one step at a time from base to HEAD — each step must succeed."""
453 _downgrade(_TEST_URL, "base")
454 for rev in _ALL_REVISIONS:
455 _upgrade(_TEST_URL, rev)
456 actual = _current_rev(_TEST_URL)
457 assert actual == rev, f"After upgrading to {rev}, got revision {actual}"
458
459
460 # ══════════════════════════════════════════════════════════════════════════════
461 # 4. Stress
462 # ══════════════════════════════════════════════════════════════════════════════
463
464
465 class TestMigrationStress:
466 """Performance and repeated-execution scenarios."""
467
468 def test_full_upgrade_completes_under_60_seconds(self) -> None:
469 """25 migrations on an empty DB must complete in under 60 seconds."""
470 _fresh_db()
471 start = time.perf_counter()
472 _upgrade(_TEST_URL)
473 elapsed = time.perf_counter() - start
474 assert elapsed < 60, f"Full upgrade took {elapsed:.1f}s (budget: 60s)"
475
476 def test_full_downgrade_completes_under_60_seconds(self) -> None:
477 """Downgrade from HEAD to base must complete in under 60 seconds."""
478 # DB is at HEAD from previous test
479 start = time.perf_counter()
480 _downgrade(_TEST_URL, "base")
481 elapsed = time.perf_counter() - start
482 assert elapsed < 60, f"Full downgrade took {elapsed:.1f}s (budget: 60s)"
483 # Restore HEAD for subsequent tests
484 _upgrade(_TEST_URL)
485
486 def test_three_consecutive_upgrade_to_head_idempotent(self) -> None:
487 """Running upgrade head twice on an already-migrated DB is a no-op (idempotent)."""
488 assert _current_rev(_TEST_URL) == _HEAD
489 _upgrade(_TEST_URL) # Already at HEAD — must not error
490 assert _current_rev(_TEST_URL) == _HEAD
491 _upgrade(_TEST_URL)
492 assert _current_rev(_TEST_URL) == _HEAD
493
494 def test_100_identity_inserts_survive_step_cycle(
495 self, migrated_engine: AsyncEngine
496 ) -> None:
497 """Insert 100 identity rows, do a single-step down/up, verify all persist."""
498
499 async def _run() -> None:
500 ids = [str(uuid.uuid4()) for _ in range(100)]
501 async with migrated_engine.connect() as conn:
502 for i, identity_id in enumerate(ids):
503 await conn.execute(
504 text(
505 "INSERT INTO musehub_identities (identity_id, handle, display_name, identity_type) "
506 "VALUES (:id, :h, :dn, 'human')"
507 ),
508 {"id": identity_id, "h": f"stress-{i}-{identity_id[:6]}", "dn": f"Stress {i}"},
509 )
510 await conn.commit()
511
512 _downgrade(_TEST_URL, "0018")
513 _upgrade(_TEST_URL)
514
515 async with migrated_engine.connect() as conn:
516 result = await conn.execute(
517 text("SELECT COUNT(*) FROM musehub_identities WHERE identity_id = ANY(:ids)"),
518 {"ids": ids},
519 )
520 count = result.scalar()
521 assert count == 100, f"Only {count}/100 identity rows survived the cycle"
522
523 asyncio.run(_run())
524
525
526 # ══════════════════════════════════════════════════════════════════════════════
527 # 5. Data Integrity
528 # ══════════════════════════════════════════════════════════════════════════════
529
530
531 class TestMigrationDataIntegrity:
532 """Schema correctness — columns, indexes, constraints at HEAD."""
533
534 def test_musehub_repos_has_required_columns(
535 self, migrated_engine: AsyncEngine
536 ) -> None:
537 cols = asyncio.run(
538 _columns(migrated_engine, "musehub_repos")
539 )
540 for col in ("repo_id", "owner", "slug", "visibility", "default_branch"):
541 assert col in cols, f"musehub_repos missing column: {col}"
542
543 def test_musehub_identities_has_agent_columns(
544 self, migrated_engine: AsyncEngine
545 ) -> None:
546 """Migration 0017 added spawned_by, scope, expires_at."""
547 cols = asyncio.run(
548 _columns(migrated_engine, "musehub_identities")
549 )
550 for col in ("spawned_by", "scope", "expires_at"):
551 assert col in cols, f"musehub_identities missing agent column: {col}"
552
553 def test_musehub_identities_has_no_legacy_user_id(
554 self, migrated_engine: AsyncEngine
555 ) -> None:
556 """Migration 0016 removed legacy_user_id from musehub_identities."""
557 cols = asyncio.run(
558 _columns(migrated_engine, "musehub_identities")
559 )
560 assert "legacy_user_id" not in cols
561
562 def test_musehub_issue_comments_has_no_state_refs(
563 self, migrated_engine: AsyncEngine
564 ) -> None:
565 """Migration 0019 dropped state_refs from musehub_issue_comments."""
566 cols = asyncio.run(
567 _columns(migrated_engine, "musehub_issue_comments")
568 )
569 assert "state_refs" not in cols
570
571 def test_musehub_auth_keys_has_algorithm_column(
572 self, migrated_engine: AsyncEngine
573 ) -> None:
574 """Migration 0008 added the algorithm column to musehub_auth_keys."""
575 cols = asyncio.run(
576 _columns(migrated_engine, "musehub_auth_keys")
577 )
578 assert "algorithm" in cols
579
580 def test_musehub_repos_unique_owner_slug_index_exists(
581 self, migrated_engine: AsyncEngine
582 ) -> None:
583 indexes = asyncio.run(
584 _indexes_for(migrated_engine, "musehub_repos")
585 )
586 assert "uq_musehub_repos_owner_slug" in indexes
587
588 def test_musehub_auth_keys_fingerprint_unique_index(
589 self, migrated_engine: AsyncEngine
590 ) -> None:
591 indexes = asyncio.run(
592 _indexes_for(migrated_engine, "musehub_auth_keys")
593 )
594 assert "uq_musehub_auth_keys_fingerprint" in indexes
595
596 def test_musehub_collaborators_has_identity_handle_column(
597 self, migrated_engine: AsyncEngine
598 ) -> None:
599 """Migration 0018 renamed user_id → identity_handle in collaborators."""
600 cols = asyncio.run(
601 _columns(migrated_engine, "musehub_collaborators")
602 )
603 assert "identity_handle" in cols
604 assert "user_id" not in cols
605
606 def test_wire_tags_table_exists(
607 self, migrated_engine: AsyncEngine
608 ) -> None:
609 tables = asyncio.run(_tables(migrated_engine))
610 assert "musehub_wire_tags" in tables
611
612 def test_sessions_table_exists(
613 self, migrated_engine: AsyncEngine
614 ) -> None:
615 tables = asyncio.run(_tables(migrated_engine))
616 assert "musehub_sessions" in tables
617
618
619 # ══════════════════════════════════════════════════════════════════════════════
620 # 6. Security
621 # ══════════════════════════════════════════════════════════════════════════════
622
623
624 class TestMigrationSecurity:
625 """Ensure migrations don't introduce security-relevant schema regressions."""
626
627 def test_legacy_auth_tables_dropped_at_head(
628 self, migrated_engine: AsyncEngine
629 ) -> None:
630 """muse_users, muse_access_tokens, musehub_profiles must not exist at HEAD."""
631 tables = asyncio.run(_tables(migrated_engine))
632 for legacy in _LEGACY_TABLES:
633 assert legacy not in tables, f"Legacy auth table '{legacy}' still present at HEAD"
634
635 def test_musehub_auth_keys_has_fingerprint_unique_constraint(
636 self, migrated_engine: AsyncEngine
637 ) -> None:
638 """Fingerprint uniqueness prevents duplicate key registration."""
639 indexes = asyncio.run(
640 _indexes_for(migrated_engine, "musehub_auth_keys")
641 )
642 assert "uq_musehub_auth_keys_fingerprint" in indexes
643
644 def test_musehub_identities_handle_unique(
645 self, migrated_engine: AsyncEngine
646 ) -> None:
647 """Identity handles must be unique — prevents impersonation via duplicate handle."""
648 indexes = asyncio.run(
649 _indexes_for(migrated_engine, "musehub_identities")
650 )
651 assert "uq_musehub_identities_handle" in indexes
652
653 def test_downgrade_does_not_expose_dropped_columns(self) -> None:
654 """After 0019 downgrade and re-upgrade, state_refs stays dropped.
655
656 A faulty downgrade might resurrect dropped columns — verify it doesn't.
657 """
658 _downgrade(_TEST_URL, "0018")
659 _upgrade(_TEST_URL) # Re-apply 0019
660 engine = create_async_engine(_TEST_URL, poolclass=NullPool)
661 try:
662 cols = asyncio.run(
663 _columns(engine, "musehub_issue_comments")
664 )
665 finally:
666 asyncio.run(engine.dispose())
667 assert "state_refs" not in cols
668
669 def test_repos_owner_slug_uniqueness_enforced(
670 self, migrated_engine: AsyncEngine
671 ) -> None:
672 """Inserting two repos with the same owner/slug must fail with an integrity error."""
673 from sqlalchemy.exc import IntegrityError
674
675 async def _run() -> None:
676 rid1 = str(uuid.uuid4())
677 rid2 = str(uuid.uuid4())
678 async with migrated_engine.connect() as conn:
679 await conn.execute(
680 text(
681 "INSERT INTO musehub_repos (repo_id, name, slug, owner, owner_user_id, visibility, default_branch) "
682 "VALUES (:id, 'dup', 'dup-slug', 'sec-owner', 'sec-owner', 'public', 'main')"
683 ),
684 {"id": rid1},
685 )
686 await conn.commit()
687 async with migrated_engine.connect() as conn:
688 with pytest.raises(IntegrityError):
689 await conn.execute(
690 text(
691 "INSERT INTO musehub_repos (repo_id, name, slug, owner, owner_user_id, visibility, default_branch) "
692 "VALUES (:id, 'dup', 'dup-slug', 'sec-owner', 'sec-owner', 'public', 'main')"
693 ),
694 {"id": rid2},
695 )
696 await conn.commit()
697
698 asyncio.run(_run())
699
700
701 # ══════════════════════════════════════════════════════════════════════════════
702 # 7. Performance
703 # ══════════════════════════════════════════════════════════════════════════════
704
705
706 class TestMigrationPerformance:
707 """Latency budgets for migration operations."""
708
709 def test_single_step_upgrade_under_5_seconds(self) -> None:
710 """Each individual migration step must complete in under 5 seconds."""
711 _fresh_db() # terminate open connections before DDL-heavy downgrade
712 for rev in _ALL_REVISIONS:
713 start = time.perf_counter()
714 _upgrade(_TEST_URL, rev)
715 elapsed = time.perf_counter() - start
716 assert elapsed < 5, (
717 f"Migration {rev} upgrade took {elapsed:.2f}s (budget: 5s per step)"
718 )
719
720 def test_single_step_downgrade_under_5_seconds(self) -> None:
721 """Each individual migration downgrade must complete in under 5 seconds."""
722 _fresh_db()
723 _upgrade(_TEST_URL) # start from a guaranteed-clean HEAD
724 for rev in reversed(_ALL_REVISIONS[:-1]):
725 start = time.perf_counter()
726 _downgrade(_TEST_URL, rev)
727 elapsed = time.perf_counter() - start
728 assert elapsed < 5, (
729 f"Migration {rev} downgrade took {elapsed:.2f}s (budget: 5s per step)"
730 )
731 # Downgrade 0001 to base
732 start = time.perf_counter()
733 _downgrade(_TEST_URL, "base")
734 elapsed = time.perf_counter() - start
735 assert elapsed < 5, f"Migration 0001 downgrade took {elapsed:.2f}s"
736 # Restore HEAD for any remaining tests
737 _upgrade(_TEST_URL)
738
739 def test_schema_introspection_under_500ms(
740 self, migrated_engine: AsyncEngine
741 ) -> None:
742 """Listing all tables in the public schema must complete in under 500ms."""
743 start = time.perf_counter()
744 asyncio.run(_tables(migrated_engine))
745 elapsed_ms = (time.perf_counter() - start) * 1000
746 assert elapsed_ms < 500, f"Table introspection took {elapsed_ms:.0f}ms (budget: 500ms)"
747
748 def test_index_introspection_under_200ms(
749 self, migrated_engine: AsyncEngine
750 ) -> None:
751 """Listing indexes for a single table must complete in under 200ms."""
752 start = time.perf_counter()
753 asyncio.run(
754 _indexes_for(migrated_engine, "musehub_repos")
755 )
756 elapsed_ms = (time.perf_counter() - start) * 1000
757 assert elapsed_ms < 200, f"Index introspection took {elapsed_ms:.0f}ms (budget: 200ms)"
758
759
760 # ══════════════════════════════════════════════════════════════════════════════
761 # Alembic chain tests (structural, no DB connection required)
762 # ══════════════════════════════════════════════════════════════════════════════
763
764
765 import inspect
766 import stat
767 import types
768
769 from alembic.config import Config
770 from alembic.script import ScriptDirectory
771
772 _REPO_ROOT = pathlib.Path(__file__).parent.parent
773 _EXPECTED_HEAD_PREFIX = "0050"
774 _EXPECTED_MIGRATION_COUNT = 50
775
776
777 # ---------------------------------------------------------------------------
778 # Helpers
779 # ---------------------------------------------------------------------------
780
781 def _script_dir() -> ScriptDirectory:
782 cfg = Config(str(_REPO_ROOT / "alembic.ini"))
783 cfg.set_main_option("script_location", str(_REPO_ROOT / "alembic"))
784 return ScriptDirectory.from_config(cfg)
785
786
787 def _is_stub_downgrade(mod: types.ModuleType) -> bool:
788 """Return True if downgrade() is a pass-only or single-ellipsis stub."""
789 import ast
790 try:
791 src = inspect.getsource(getattr(mod, "downgrade"))
792 except (OSError, AttributeError):
793 return True
794
795 # Strip the 'def downgrade...:' line and check what's left
796 lines = [l.strip() for l in src.splitlines() if l.strip() and not l.strip().startswith("def ")]
797 # A stub body is just 'pass', '...', or a docstring with nothing else
798 non_comment = [l for l in lines if not l.startswith("#") and not l.startswith('"""') and not l.startswith("'''")]
799 if not non_comment:
800 return True
801 if len(non_comment) == 1 and non_comment[0] in ("pass", "..."):
802 return True
803 return False
804
805
806 # ---------------------------------------------------------------------------
807 # Linear chain / versioning
808 # ---------------------------------------------------------------------------
809
810 def test_migration_chain_is_linear() -> None:
811 """Migration graph must have exactly one head (no branches)."""
812 heads = _script_dir().get_heads()
813 assert len(heads) == 1, (
814 f"Expected single-head chain, got {len(heads)} heads: {heads}. "
815 "Resolve the branch before merging."
816 )
817
818
819 def test_migration_count_matches_expected() -> None:
820 """Migration count must equal _EXPECTED_MIGRATION_COUNT.
821
822 Update _EXPECTED_MIGRATION_COUNT here when adding a new migration.
823 """
824 revisions = list(_script_dir().walk_revisions())
825 assert len(revisions) == _EXPECTED_MIGRATION_COUNT, (
826 f"Expected {_EXPECTED_MIGRATION_COUNT} migrations, found {len(revisions)}. "
827 "Update _EXPECTED_MIGRATION_COUNT in this file."
828 )
829
830
831 def test_head_revision_prefix() -> None:
832 """Head must start with the expected revision prefix."""
833 heads = _script_dir().get_heads()
834 assert len(heads) == 1
835 assert heads[0].startswith(_EXPECTED_HEAD_PREFIX), (
836 f"Expected head starting with '{_EXPECTED_HEAD_PREFIX}', got '{heads[0]}'. "
837 "Update _EXPECTED_HEAD_PREFIX when a new migration is added."
838 )
839
840
841 def test_all_migrations_importable() -> None:
842 """Every migration module must be importable without errors."""
843 for rev in _script_dir().walk_revisions():
844 assert rev.module is not None, (
845 f"Revision {rev.revision} has no module — check for missing file."
846 )
847
848
849 def test_revision_ids_are_sequential_integers() -> None:
850 """Revision IDs must be zero-padded 4-digit integers with no gaps."""
851 revisions = sorted(_script_dir().walk_revisions(), key=lambda r: r.revision)
852 ids = sorted(int(r.revision[:4]) for r in revisions if r.revision[:4].isdigit())
853 expected = list(range(1, len(ids) + 1))
854 assert ids == expected, (
855 f"Revision IDs are not sequential (no gaps): found {ids}, expected {expected}."
856 )
857
858
859 # ---------------------------------------------------------------------------
860 # Downgrade coverage — every forward migration has a real downgrade
861 # ---------------------------------------------------------------------------
862
863 def test_all_migrations_have_downgrade_function() -> None:
864 """Every migration module must define a downgrade() function."""
865 for rev in _script_dir().walk_revisions():
866 mod = rev.module
867 assert mod is not None
868 assert hasattr(mod, "downgrade"), (
869 f"Revision {rev.revision} is missing a downgrade() function."
870 )
871
872
873 def test_no_stub_downgrade_implementations() -> None:
874 """No migration may have a pass-only or ellipsis-only downgrade().
875
876 A stub downgrade makes rollback a silent no-op — forbidden.
877 """
878 stubs = []
879 for rev in _script_dir().walk_revisions():
880 mod = rev.module
881 if mod is not None and _is_stub_downgrade(mod):
882 stubs.append(rev.revision)
883
884 assert not stubs, (
885 f"Migrations with stub (non-functional) downgrade(): {stubs}. "
886 "Implement the actual rollback DDL."
887 )
888
889
890 def test_every_migration_references_tables_in_downgrade() -> None:
891 """Migrations that add a column or table must also reference it in downgrade().
892
893 Heuristic: if upgrade() calls op.add_column / op.create_table for table X,
894 downgrade() must reference X (via drop_column / drop_table).
895 Checked by source inspection — not exhaustive, but catches obvious omissions.
896 """
897 import re
898
899 _ADD_RE = re.compile(r'op\.(add_column|create_table)\(\s*["\'](\w+)["\']')
900 _DROP_RE = re.compile(r'op\.(drop_column|drop_table)\(\s*["\'](\w+)["\']')
901
902 violations = []
903 for rev in _script_dir().walk_revisions():
904 mod = rev.module
905 if mod is None:
906 continue
907 try:
908 up_src = inspect.getsource(getattr(mod, "upgrade"))
909 down_src = inspect.getsource(getattr(mod, "downgrade"))
910 except (OSError, AttributeError):
911 continue
912
913 tables_added = {m.group(2) for m in _ADD_RE.finditer(up_src)}
914 tables_dropped = {m.group(2) for m in _DROP_RE.finditer(down_src)}
915 missing = tables_added - tables_dropped
916 if missing:
917 violations.append(f"{rev.revision}: added {missing} but downgrade() doesn't drop them")
918
919 assert not violations, (
920 "Migrations with incomplete downgrade():\n" + "\n".join(violations)
921 )
922
923
924 # ---------------------------------------------------------------------------
925 # Transaction safety — env.py wraps migrations in a transaction
926 # ---------------------------------------------------------------------------
927
928 def test_env_py_uses_begin_transaction() -> None:
929 """alembic/env.py must call context.begin_transaction() for both offline and online runs.
930
931 This ensures that a failed migration rolls back cleanly instead of leaving
932 the schema in a partially-applied state.
933 """
934 env_path = _REPO_ROOT / "alembic" / "env.py"
935 src = env_path.read_text()
936 count = src.count("context.begin_transaction()")
937 assert count >= 2, (
938 f"Expected at least 2 calls to context.begin_transaction() in env.py "
939 f"(one for offline, one for online), found {count}. "
940 "Wrap both run_migrations_offline() and do_run_migrations() in a transaction."
941 )
942
943
944 # ---------------------------------------------------------------------------
945 # Prod-snapshot migration test script
946 # ---------------------------------------------------------------------------
947
948 def test_migrate_test_script_exists_and_is_executable() -> None:
949 """deploy/migrate-test.sh must exist and be executable.
950
951 This script is the mechanism for testing migrations against a production
952 data snapshot before applying to production (checklist item 5.3.2).
953 """
954 script = _REPO_ROOT / "deploy" / "migrate-test.sh"
955 assert script.exists(), (
956 "deploy/migrate-test.sh is missing. "
957 "This script is required to validate migrations against a prod snapshot."
958 )
959 mode = script.stat().st_mode
960 assert mode & stat.S_IXUSR, (
961 "deploy/migrate-test.sh must be executable (chmod +x)."
962 )
963
964
965 def test_migrate_test_script_contains_round_trip() -> None:
966 """deploy/migrate-test.sh must perform an upgrade→downgrade→upgrade round-trip."""
967 script = (_REPO_ROOT / "deploy" / "migrate-test.sh").read_text()
968 assert "upgrade head" in script, "script must run 'alembic upgrade head'"
969 assert "downgrade" in script, "script must run 'alembic downgrade' step"
970 # Must do upgrade twice (initial + after downgrade round-trip)
971 assert script.count("upgrade head") >= 2, (
972 "script must upgrade to HEAD twice (initial apply + round-trip after downgrade)"
973 )
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago