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