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