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