"""Section 36 — Database Migrations / Alembic (7-layer test suite). Covers: alembic/versions/0001_consolidated_schema.py → 0019_drop_issue_comment_state_refs.py 19 migrations, linear chain (no branches) Key design decisions: - Tests require a real PostgreSQL instance (migrations use now(), JSON casts, etc.). The DB URL is: postgresql+asyncpg://musehub:musehub@localhost:5434/musehub_migration_test - Each test creates a FRESH database (drop + create) so tests are fully isolated. - The `migration_db_url` fixture is module-scoped: the fresh DB is created once per module; individual test groups share it but leave the DB at a known revision (HEAD) after each test. - Tests that need the DB at a specific intermediate revision use `_run_to(cfg, rev)` helpers that upgrade/downgrade as needed. - **Never run the full test suite in CI** — these tests connect to the real Postgres container. Run with: python -m pytest tests/test_migrations_section36.py -x -q Connection note: The postgres container is exposed at localhost:5434 (mapped from 5432). Credentials: musehub / musehub. """ from __future__ import annotations import asyncio import time import uuid from collections.abc import AsyncGenerator, Generator from typing import Any import pytest import pytest_asyncio from sqlalchemy import inspect, text from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, create_async_engine from sqlalchemy.pool import NullPool # ── constants ───────────────────────────────────────────────────────────────── _PG_BASE_URL = "postgresql+asyncpg://musehub:musehub@localhost:5434" _TEST_DB = "musehub_migration_test_s36" _TEST_URL = f"{_PG_BASE_URL}/{_TEST_DB}" _ADMIN_URL = f"{_PG_BASE_URL}/musehub" # connect to main DB to create/drop the test DB _ALL_REVISIONS = [ "0001", "0002", "0003", "0004", "0005", "0006", "0007", "0008", "0009", "0010", "0011", "0012", "0013", "0014", "0015", "0016", "0017", "0018", "0019", "0020", "0021", "0022", "0023", "0024", "0025", "0026", "0027", "0028", ] _HEAD = "0028" # Tables that MUST exist after a full upgrade to HEAD # CI tables (musehub_ci_*) are excluded — dropped by migration 0024. _REQUIRED_TABLES = { "alembic_version", "muse_commits", "muse_objects", "muse_snapshots", "muse_tags", "musehub_auth_keys", "musehub_branches", "musehub_collaborators", "musehub_commits", "musehub_coord_records", "musehub_coord_reservations", "musehub_coord_tasks", "musehub_domain_installs", "musehub_domains", "musehub_identities", "musehub_issues", "musehub_labels", "musehub_objects", "musehub_proposals", "musehub_releases", "musehub_repos", "musehub_sessions", "musehub_snapshot_entries", "musehub_snapshots", "musehub_symbol_index", "musehub_wire_tags", } # Tables that MUST be absent after a full upgrade (dropped by 0016) _LEGACY_TABLES = {"muse_users", "muse_access_tokens", "musehub_profiles"} # ── helpers ─────────────────────────────────────────────────────────────────── def _run_alembic(db_url: str, *args: str) -> None: """Run an alembic command in a subprocess to avoid the settings lru_cache. The in-process lru_cache on musehub.config.settings is populated from conftest.py before DATABASE_URL is set; running in a subprocess guarantees a clean settings instance that picks up our DATABASE_URL. """ import subprocess import sys env = { "DATABASE_URL": db_url, "MUSE_ENV": "test", "PATH": "/usr/bin:/bin:/usr/local/bin", } result = subprocess.run( [sys.executable, "-m", "alembic"] + list(args), cwd="/Users/gabriel/musehub", env=env, capture_output=True, text=True, timeout=120, ) if result.returncode != 0: raise RuntimeError( f"alembic {' '.join(args)} failed:\n{result.stderr[-2000:]}" ) def _upgrade(db_url: str, revision: str = "head") -> None: _run_alembic(db_url, "upgrade", revision) def _downgrade(db_url: str, revision: str) -> None: _run_alembic(db_url, "downgrade", revision) def _current_rev(db_url: str) -> str | None: from alembic.runtime.migration import MigrationContext from sqlalchemy import create_engine sync_url = db_url.replace("+asyncpg", "") engine = create_engine(sync_url) with engine.connect() as conn: ctx = MigrationContext.configure(conn) rev = ctx.get_current_revision() engine.dispose() return rev async def _tables(engine: AsyncEngine) -> set[str]: async with engine.connect() as conn: result = await conn.execute( text( "SELECT tablename FROM pg_tables " "WHERE schemaname='public' ORDER BY tablename" ) ) return {row[0] for row in result} async def _indexes_for(engine: AsyncEngine, table: str) -> set[str]: async with engine.connect() as conn: result = await conn.execute( text( "SELECT indexname FROM pg_indexes " "WHERE schemaname='public' AND tablename = :t" ), {"t": table}, ) return {row[0] for row in result} async def _columns(engine: AsyncEngine, table: str) -> set[str]: async with engine.connect() as conn: result = await conn.execute( text( "SELECT column_name FROM information_schema.columns " "WHERE table_schema='public' AND table_name = :t" ), {"t": table}, ) return {row[0] for row in result} def _fresh_db() -> None: """Drop and recreate the section-36 test database synchronously.""" from sqlalchemy import create_engine, text as stext sync_admin = _ADMIN_URL.replace("+asyncpg", "") engine = create_engine(sync_admin, isolation_level="AUTOCOMMIT") with engine.connect() as conn: conn.execute( stext(f"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname='{_TEST_DB}'") ) conn.execute(stext(f"DROP DATABASE IF EXISTS {_TEST_DB}")) conn.execute(stext(f"CREATE DATABASE {_TEST_DB}")) engine.dispose() # ── module-scoped engine fixture ────────────────────────────────────────────── @pytest.fixture(scope="module") def migrated_engine() -> Generator[AsyncEngine, None, None]: """Create a fresh DB, run all migrations to HEAD, yield an async engine. Module-scoped: created once for all tests in this file. The DB is left at HEAD after each test (tests that downgrade must re-upgrade). """ _fresh_db() _upgrade(_TEST_URL) engine = create_async_engine(_TEST_URL, poolclass=NullPool) yield engine asyncio.run(engine.dispose()) # ══════════════════════════════════════════════════════════════════════════════ # 1. Unit # ══════════════════════════════════════════════════════════════════════════════ class TestMigrationUnit: """Static analysis of migration files — no DB connection needed.""" def test_revision_chain_is_linear(self) -> None: """Every migration (except 0001) has a down_revision pointing to its predecessor.""" import importlib.util import pathlib versions_dir = pathlib.Path("/Users/gabriel/musehub/alembic/versions") revisions = {} for f in versions_dir.glob("*.py"): spec = importlib.util.spec_from_file_location(f.stem, f) assert spec and spec.loader mod = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(mod) rev = getattr(mod, "revision", None) down = getattr(mod, "down_revision", None) if rev: revisions[str(rev)] = str(down) if down else None # 0001 has no predecessor assert revisions.get("0001") is None # Every other migration's down_revision must exist as a revision for rev, down in revisions.items(): if down is not None: assert down in revisions, ( f"Migration {rev}: down_revision '{down}' not found" ) def test_all_25_migrations_present(self) -> None: import pathlib versions_dir = pathlib.Path("/Users/gabriel/musehub/alembic/versions") files = {f.stem for f in versions_dir.glob("*.py") if not f.stem.startswith("__")} assert len(files) == 28, f"Expected 28 migrations, found {len(files)}: {files}" def test_head_revision_is_0028(self) -> None: import pathlib versions_dir = pathlib.Path("/Users/gabriel/musehub/alembic/versions") # The head is the revision not referenced as any down_revision import importlib.util all_revs: set[str] = set() down_revs: set[str] = set() for f in versions_dir.glob("*.py"): spec = importlib.util.spec_from_file_location(f.stem, f) assert spec and spec.loader mod = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(mod) rev = getattr(mod, "revision", None) down = getattr(mod, "down_revision", None) if rev: all_revs.add(str(rev)) if down: down_revs.add(str(down)) heads = all_revs - down_revs assert heads == {"0028"}, f"Expected head 0028, got: {heads}" def test_each_migration_has_upgrade_and_downgrade(self) -> None: import pathlib, importlib.util versions_dir = pathlib.Path("/Users/gabriel/musehub/alembic/versions") for f in versions_dir.glob("*.py"): spec = importlib.util.spec_from_file_location(f.stem, f) assert spec and spec.loader mod = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(mod) assert callable(getattr(mod, "upgrade", None)), ( f"{f.name} missing upgrade()" ) assert callable(getattr(mod, "downgrade", None)), ( f"{f.name} missing downgrade()" ) def test_alembic_ini_points_to_correct_script_location(self) -> None: import configparser parser = configparser.ConfigParser() parser.read("/Users/gabriel/musehub/alembic.ini") script_location = parser.get("alembic", "script_location", fallback="") assert "alembic" in script_location def test_env_py_imports_all_model_bases(self) -> None: env_text = open("/Users/gabriel/musehub/alembic/env.py").read() assert "from musehub.db.database import Base" in env_text assert "target_metadata = Base.metadata" in env_text # ══════════════════════════════════════════════════════════════════════════════ # 2. Integration # ══════════════════════════════════════════════════════════════════════════════ class TestMigrationIntegration: """Real DB — run migrations and inspect schema state.""" def test_upgrade_to_head_succeeds(self) -> None: """Full migration chain from empty DB to HEAD completes without error.""" _fresh_db() _upgrade(_TEST_URL) rev = _current_rev(_TEST_URL) assert rev == _HEAD def test_current_revision_tracked_in_alembic_version( self, migrated_engine: AsyncEngine ) -> None: async def _check() -> str | None: async with migrated_engine.connect() as conn: result = await conn.execute(text("SELECT version_num FROM alembic_version")) row = result.fetchone() return row[0] if row else None rev = asyncio.run(_check()) assert rev == _HEAD def test_required_tables_exist_at_head( self, migrated_engine: AsyncEngine ) -> None: tables = asyncio.run(_tables(migrated_engine)) missing = _REQUIRED_TABLES - tables assert not missing, f"Tables missing after upgrade to HEAD: {missing}" def test_legacy_tables_absent_at_head( self, migrated_engine: AsyncEngine ) -> None: tables = asyncio.run(_tables(migrated_engine)) present_legacy = _LEGACY_TABLES & tables assert not present_legacy, f"Legacy tables still present after HEAD: {present_legacy}" def test_downgrade_then_upgrade_returns_to_head(self) -> None: """Downgrade one step and re-upgrade — must return to HEAD cleanly.""" _downgrade(_TEST_URL, "0018") assert _current_rev(_TEST_URL) == "0018" _upgrade(_TEST_URL, "head") assert _current_rev(_TEST_URL) == _HEAD def test_full_downgrade_to_base_then_re_upgrade(self) -> None: """Downgrade all the way to base then re-upgrade — round trip works.""" _downgrade(_TEST_URL, "base") assert _current_rev(_TEST_URL) is None _upgrade(_TEST_URL) assert _current_rev(_TEST_URL) == _HEAD # ══════════════════════════════════════════════════════════════════════════════ # 3. End-to-End # ══════════════════════════════════════════════════════════════════════════════ class TestMigrationE2E: """Full stack: migrate, insert data, downgrade, re-upgrade, verify data.""" def test_repo_data_survives_incremental_upgrade( self, migrated_engine: AsyncEngine ) -> None: """Insert a repo row at HEAD, downgrade one step, re-upgrade — row survives.""" async def _run() -> None: repo_id = str(uuid.uuid4()) async with migrated_engine.connect() as conn: await conn.execute( text( "INSERT INTO musehub_repos (repo_id, name, slug, owner, owner_user_id, " "visibility, default_branch) " "VALUES (:id, :n, :s, :o, :oid, 'public', 'main')" ), {"id": repo_id, "n": "e2e-repo", "s": "e2e-repo", "o": "testuser", "oid": "testuser"}, ) await conn.commit() _downgrade(_TEST_URL, "0018") _upgrade(_TEST_URL) async with migrated_engine.connect() as conn: result = await conn.execute( text("SELECT repo_id FROM musehub_repos WHERE repo_id = :id"), {"id": repo_id}, ) row = result.fetchone() assert row is not None, "Repo row lost after downgrade/re-upgrade cycle" asyncio.run(_run()) def test_identity_row_survives_upgrade_chain( self, migrated_engine: AsyncEngine ) -> None: """Insert an identity at HEAD; must still exist after a single-step cycle.""" async def _run() -> None: identity_id = str(uuid.uuid4()) async with migrated_engine.connect() as conn: await conn.execute( text( "INSERT INTO musehub_identities (id, handle, display_name, identity_type) " "VALUES (:id, :h, :dn, 'human')" ), {"id": identity_id, "h": f"e2e-user-{identity_id[:8]}", "dn": "E2E User"}, ) await conn.commit() _downgrade(_TEST_URL, "0018") _upgrade(_TEST_URL) async with migrated_engine.connect() as conn: result = await conn.execute( text("SELECT id FROM musehub_identities WHERE id = :id"), {"id": identity_id}, ) row = result.fetchone() assert row is not None asyncio.run(_run()) def test_each_migration_step_is_individually_runnable(self) -> None: """Upgrade one step at a time from base to HEAD — each step must succeed.""" _downgrade(_TEST_URL, "base") for rev in _ALL_REVISIONS: _upgrade(_TEST_URL, rev) actual = _current_rev(_TEST_URL) assert actual == rev, f"After upgrading to {rev}, got revision {actual}" # ══════════════════════════════════════════════════════════════════════════════ # 4. Stress # ══════════════════════════════════════════════════════════════════════════════ class TestMigrationStress: """Performance and repeated-execution scenarios.""" def test_full_upgrade_completes_under_60_seconds(self) -> None: """25 migrations on an empty DB must complete in under 60 seconds.""" _fresh_db() start = time.perf_counter() _upgrade(_TEST_URL) elapsed = time.perf_counter() - start assert elapsed < 60, f"Full upgrade took {elapsed:.1f}s (budget: 60s)" def test_full_downgrade_completes_under_60_seconds(self) -> None: """Downgrade from HEAD to base must complete in under 60 seconds.""" # DB is at HEAD from previous test start = time.perf_counter() _downgrade(_TEST_URL, "base") elapsed = time.perf_counter() - start assert elapsed < 60, f"Full downgrade took {elapsed:.1f}s (budget: 60s)" # Restore HEAD for subsequent tests _upgrade(_TEST_URL) def test_three_consecutive_upgrade_to_head_idempotent(self) -> None: """Running upgrade head twice on an already-migrated DB is a no-op (idempotent).""" assert _current_rev(_TEST_URL) == _HEAD _upgrade(_TEST_URL) # Already at HEAD — must not error assert _current_rev(_TEST_URL) == _HEAD _upgrade(_TEST_URL) assert _current_rev(_TEST_URL) == _HEAD def test_100_identity_inserts_survive_step_cycle( self, migrated_engine: AsyncEngine ) -> None: """Insert 100 identity rows, do a single-step down/up, verify all persist.""" async def _run() -> None: ids = [str(uuid.uuid4()) for _ in range(100)] async with migrated_engine.connect() as conn: for i, identity_id in enumerate(ids): await conn.execute( text( "INSERT INTO musehub_identities (id, handle, display_name, identity_type) " "VALUES (:id, :h, :dn, 'human')" ), {"id": identity_id, "h": f"stress-{i}-{identity_id[:6]}", "dn": f"Stress {i}"}, ) await conn.commit() _downgrade(_TEST_URL, "0018") _upgrade(_TEST_URL) async with migrated_engine.connect() as conn: result = await conn.execute( text("SELECT COUNT(*) FROM musehub_identities WHERE id = ANY(:ids)"), {"ids": ids}, ) count = result.scalar() assert count == 100, f"Only {count}/100 identity rows survived the cycle" asyncio.run(_run()) # ══════════════════════════════════════════════════════════════════════════════ # 5. Data Integrity # ══════════════════════════════════════════════════════════════════════════════ class TestMigrationDataIntegrity: """Schema correctness — columns, indexes, constraints at HEAD.""" def test_musehub_repos_has_required_columns( self, migrated_engine: AsyncEngine ) -> None: cols = asyncio.run( _columns(migrated_engine, "musehub_repos") ) for col in ("repo_id", "owner", "slug", "visibility", "default_branch"): assert col in cols, f"musehub_repos missing column: {col}" def test_musehub_identities_has_agent_columns( self, migrated_engine: AsyncEngine ) -> None: """Migration 0017 added spawned_by, scope, expires_at.""" cols = asyncio.run( _columns(migrated_engine, "musehub_identities") ) for col in ("spawned_by", "scope", "expires_at"): assert col in cols, f"musehub_identities missing agent column: {col}" def test_musehub_identities_has_no_legacy_user_id( self, migrated_engine: AsyncEngine ) -> None: """Migration 0016 removed legacy_user_id from musehub_identities.""" cols = asyncio.run( _columns(migrated_engine, "musehub_identities") ) assert "legacy_user_id" not in cols def test_musehub_issue_comments_has_no_state_refs( self, migrated_engine: AsyncEngine ) -> None: """Migration 0019 dropped state_refs from musehub_issue_comments.""" cols = asyncio.run( _columns(migrated_engine, "musehub_issue_comments") ) assert "state_refs" not in cols def test_musehub_auth_keys_has_algorithm_column( self, migrated_engine: AsyncEngine ) -> None: """Migration 0008 added the algorithm column to musehub_auth_keys.""" cols = asyncio.run( _columns(migrated_engine, "musehub_auth_keys") ) assert "algorithm" in cols def test_musehub_repos_unique_owner_slug_index_exists( self, migrated_engine: AsyncEngine ) -> None: indexes = asyncio.run( _indexes_for(migrated_engine, "musehub_repos") ) assert "uq_musehub_repos_owner_slug" in indexes def test_musehub_auth_keys_fingerprint_unique_index( self, migrated_engine: AsyncEngine ) -> None: indexes = asyncio.run( _indexes_for(migrated_engine, "musehub_auth_keys") ) assert "uq_musehub_auth_keys_fingerprint" in indexes def test_musehub_collaborators_has_identity_handle_column( self, migrated_engine: AsyncEngine ) -> None: """Migration 0018 renamed user_id → identity_handle in collaborators.""" cols = asyncio.run( _columns(migrated_engine, "musehub_collaborators") ) assert "identity_handle" in cols assert "user_id" not in cols def test_musehub_snapshot_entries_exists_and_has_key_columns( self, migrated_engine: AsyncEngine ) -> None: """Migration 0013 replaced manifest blob with musehub_snapshot_entries table.""" tables = asyncio.run(_tables(migrated_engine)) assert "musehub_snapshot_entries" in tables assert "musehub_snapshots" in tables # still exists, manifest column removed def test_wire_tags_table_exists( self, migrated_engine: AsyncEngine ) -> None: tables = asyncio.run(_tables(migrated_engine)) assert "musehub_wire_tags" in tables def test_sessions_table_exists( self, migrated_engine: AsyncEngine ) -> None: tables = asyncio.run(_tables(migrated_engine)) assert "musehub_sessions" in tables # ══════════════════════════════════════════════════════════════════════════════ # 6. Security # ══════════════════════════════════════════════════════════════════════════════ class TestMigrationSecurity: """Ensure migrations don't introduce security-relevant schema regressions.""" def test_legacy_auth_tables_dropped_at_head( self, migrated_engine: AsyncEngine ) -> None: """muse_users, muse_access_tokens, musehub_profiles must not exist at HEAD.""" tables = asyncio.run(_tables(migrated_engine)) for legacy in _LEGACY_TABLES: assert legacy not in tables, f"Legacy auth table '{legacy}' still present at HEAD" def test_musehub_auth_keys_has_fingerprint_unique_constraint( self, migrated_engine: AsyncEngine ) -> None: """Fingerprint uniqueness prevents duplicate key registration.""" indexes = asyncio.run( _indexes_for(migrated_engine, "musehub_auth_keys") ) assert "uq_musehub_auth_keys_fingerprint" in indexes def test_musehub_identities_handle_unique( self, migrated_engine: AsyncEngine ) -> None: """Identity handles must be unique — prevents impersonation via duplicate handle.""" indexes = asyncio.run( _indexes_for(migrated_engine, "musehub_identities") ) assert "uq_musehub_identities_handle" in indexes def test_downgrade_does_not_expose_dropped_columns(self) -> None: """After 0019 downgrade and re-upgrade, state_refs stays dropped. A faulty downgrade might resurrect dropped columns — verify it doesn't. """ _downgrade(_TEST_URL, "0018") _upgrade(_TEST_URL) # Re-apply 0019 engine = create_async_engine(_TEST_URL, poolclass=NullPool) try: cols = asyncio.run( _columns(engine, "musehub_issue_comments") ) finally: asyncio.run(engine.dispose()) assert "state_refs" not in cols def test_repos_owner_slug_uniqueness_enforced( self, migrated_engine: AsyncEngine ) -> None: """Inserting two repos with the same owner/slug must fail with an integrity error.""" from sqlalchemy.exc import IntegrityError async def _run() -> None: rid1 = str(uuid.uuid4()) rid2 = str(uuid.uuid4()) async with migrated_engine.connect() as conn: await conn.execute( text( "INSERT INTO musehub_repos (repo_id, name, slug, owner, owner_user_id, visibility, default_branch) " "VALUES (:id, 'dup', 'dup-slug', 'sec-owner', 'sec-owner', 'public', 'main')" ), {"id": rid1}, ) await conn.commit() async with migrated_engine.connect() as conn: with pytest.raises(IntegrityError): await conn.execute( text( "INSERT INTO musehub_repos (repo_id, name, slug, owner, owner_user_id, visibility, default_branch) " "VALUES (:id, 'dup', 'dup-slug', 'sec-owner', 'sec-owner', 'public', 'main')" ), {"id": rid2}, ) await conn.commit() asyncio.run(_run()) # ══════════════════════════════════════════════════════════════════════════════ # 7. Performance # ══════════════════════════════════════════════════════════════════════════════ class TestMigrationPerformance: """Latency budgets for migration operations.""" def test_single_step_upgrade_under_5_seconds(self) -> None: """Each individual migration step must complete in under 5 seconds.""" _downgrade(_TEST_URL, "base") for rev in _ALL_REVISIONS: start = time.perf_counter() _upgrade(_TEST_URL, rev) elapsed = time.perf_counter() - start assert elapsed < 5, ( f"Migration {rev} upgrade took {elapsed:.2f}s (budget: 5s per step)" ) def test_single_step_downgrade_under_5_seconds(self) -> None: """Each individual migration downgrade must complete in under 5 seconds.""" # Currently at HEAD for rev in reversed(_ALL_REVISIONS[:-1]): # downgrade from 0019 to 0001 start = time.perf_counter() _downgrade(_TEST_URL, rev) elapsed = time.perf_counter() - start assert elapsed < 5, ( f"Migration {rev} downgrade took {elapsed:.2f}s (budget: 5s per step)" ) # Downgrade 0001 to base start = time.perf_counter() _downgrade(_TEST_URL, "base") elapsed = time.perf_counter() - start assert elapsed < 5, f"Migration 0001 downgrade took {elapsed:.2f}s" # Restore HEAD for any remaining tests _upgrade(_TEST_URL) def test_schema_introspection_under_500ms( self, migrated_engine: AsyncEngine ) -> None: """Listing all tables in the public schema must complete in under 500ms.""" start = time.perf_counter() asyncio.run(_tables(migrated_engine)) elapsed_ms = (time.perf_counter() - start) * 1000 assert elapsed_ms < 500, f"Table introspection took {elapsed_ms:.0f}ms (budget: 500ms)" def test_index_introspection_under_200ms( self, migrated_engine: AsyncEngine ) -> None: """Listing indexes for a single table must complete in under 200ms.""" start = time.perf_counter() asyncio.run( _indexes_for(migrated_engine, "musehub_repos") ) elapsed_ms = (time.perf_counter() - start) * 1000 assert elapsed_ms < 200, f"Index introspection took {elapsed_ms:.0f}ms (budget: 200ms)"