diff --git a/tests/integration/persistence/test_migration_branch_convergence.py b/tests/integration/persistence/test_migration_branch_convergence.py index c48f29702..903b2794a 100644 --- a/tests/integration/persistence/test_migration_branch_convergence.py +++ b/tests/integration/persistence/test_migration_branch_convergence.py @@ -108,6 +108,18 @@ async def _seed_common(engine): base_url='https://fixture.invalid', api_keys=['fixture-key'], ) + await _insert( + conn, + 'llm_models', + uuid='model-1', + workspace_uuid=WORKSPACE, + name='Fixture model', + provider_uuid='provider-1', + abilities=['vision'], + extra_args={'sentinel': 'preserve-model'}, + prefered_ranking=7, + reasoning_config={'level': 'provider_default'}, + ) await _insert( conn, 'legacy_pipelines', @@ -261,6 +273,31 @@ async def _assert_merged_schema_and_data(engine, source): assert await conn.scalar(sa.text('SELECT password FROM users')) == 'hash' assert await conn.scalar(sa.text('SELECT COUNT(*) FROM workspace_memberships')) == 1 assert await conn.scalar(sa.text('SELECT name FROM model_providers')) == 'Fixture provider' + models = await conn.run_sync(lambda sync: sa.Table('llm_models', sa.MetaData(), autoload_with=sync)) + model = ( + await conn.execute( + sa.select( + models.c.name, + models.c.abilities, + models.c.extra_args, + models.c.prefered_ranking, + models.c.reasoning_config, + ) + ) + ).one() + assert tuple(model) == ( + 'Fixture model', + ['vision'], + {'sentinel': 'preserve-model'}, + 7, + {'level': 'provider_default'}, + ) + reasoning_column = await conn.run_sync( + lambda sync: next( + column for column in sa.inspect(sync).get_columns('llm_models') if column['name'] == 'reasoning_config' + ) + ) + assert reasoning_column['nullable'] is False assert await conn.scalar(sa.text('SELECT description FROM legacy_pipelines')) == 'preserve pipeline' assert ( await conn.execute(sa.text('SELECT message_content FROM monitoring_messages ORDER BY id')) @@ -306,8 +343,18 @@ async def _assert_merged_schema_and_data(engine, source): @pytest.mark.asyncio -@pytest.mark.parametrize('source', ['baseline', 'master', 'beta']) -async def test_historical_schema_and_populated_branch_upgrade(convergence_engine, source): +@pytest.mark.parametrize( + ('source', 'checkpoint'), + [ + ('baseline', None), + ('baseline', '0016_space_launch_replay'), + ('baseline', '0018_merge_workspace_heads'), + ('baseline', '0018_llm_reasoning_config'), + ('master', None), + ('beta', None), + ], +) +async def test_historical_schema_and_populated_branch_upgrade(convergence_engine, source, checkpoint): engine = convergence_engine fixture = json.loads((FIXTURES / f'{source}_schema.json').read_text()) async with engine.begin() as conn: @@ -338,6 +385,9 @@ async def test_historical_schema_and_populated_branch_upgrade(convergence_engine await run_alembic_upgrade(engine, '0010_scope_resources') async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) + if checkpoint is not None: + await run_alembic_upgrade(engine, checkpoint) + assert await get_alembic_current(engine) == checkpoint await run_alembic_upgrade(engine, 'head') await _assert_merged_schema_and_data(engine, source) await run_alembic_upgrade(engine, 'head') @@ -351,8 +401,29 @@ async def test_empty_database_startup_schema_then_real_migrations(convergence_en assert await conn.run_sync(lambda sync: sa.inspect(sync).get_table_names()) == [] # This is the documented fresh-install contract: create_all precedes Alembic. await conn.run_sync(Base.metadata.create_all) + await conn.execute( + Base.metadata.tables['metadata'].insert().values(key='instance_uuid', value='migration-convergence-fresh') + ) + await conn.execute( + Base.metadata.tables['users'] + .insert() + .values( + uuid=ACCOUNT, user='fresh@example.com', normalized_email='fresh@example.com', password='fresh-password' + ) + ) await run_alembic_upgrade(engine, 'head') assert await get_alembic_current(engine) == get_alembic_head() + async with engine.connect() as conn: + tables = await conn.run_sync(lambda sync: set(sa.inspect(sync).get_table_names())) + assert { + 'users', + 'bots', + 'agents', + 'runner_state', + 'directory_projection_inbox', + 'passkey_credentials', + } <= tables + assert await conn.scalar(sa.text('SELECT password FROM users')) == 'fresh-password' await run_alembic_upgrade(engine, 'head') assert await get_alembic_current(engine) == get_alembic_head() diff --git a/tests/integration/persistence/test_migrations.py b/tests/integration/persistence/test_migrations.py index 84c216dea..602899195 100644 --- a/tests/integration/persistence/test_migrations.py +++ b/tests/integration/persistence/test_migrations.py @@ -122,39 +122,6 @@ class TestSQLiteMigrationBaseline: class TestSQLiteMigrationUpgrade: """Tests for upgrade to head workflow.""" - @pytest.mark.asyncio - async def test_upgrade_from_published_space_launch_head_to_merged_head(self, sqlite_engine): - """A database released at the production-only 0016 head must remain upgradable.""" - async with sqlite_engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - - await run_alembic_stamp(sqlite_engine, '0016_space_launch_replay') - await run_alembic_upgrade(sqlite_engine, 'head') - - assert await get_alembic_current(sqlite_engine) == _get_script_head() - - @pytest.mark.asyncio - async def test_upgrade_from_development_workspace_head_to_merged_head(self, sqlite_engine): - """A database at the 4.11 development head must absorb later master migrations.""" - async with sqlite_engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - - await run_alembic_stamp(sqlite_engine, '0018_merge_workspace_heads') - await run_alembic_upgrade(sqlite_engine, 'head') - - assert await get_alembic_current(sqlite_engine) == _get_script_head() - - @pytest.mark.asyncio - async def test_upgrade_from_reasoning_config_head_to_merged_head(self, sqlite_engine): - """A database that already ran the feature migration must remain upgradable.""" - async with sqlite_engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - - await run_alembic_stamp(sqlite_engine, '0018_llm_reasoning_config') - await run_alembic_upgrade(sqlite_engine, 'head') - - assert await get_alembic_current(sqlite_engine) == _get_script_head() - @pytest.mark.asyncio async def test_upgrade_removes_agent_enabled_column(self, sqlite_engine): async with sqlite_engine.begin() as conn: @@ -171,62 +138,6 @@ class TestSQLiteMigrationUpgrade: assert 'enabled' not in columns - @pytest.mark.asyncio - async def test_upgrade_from_baseline_to_head(self, sqlite_engine): - """ - Upgrade from baseline to head applies all migrations. - - Workflow: - 1. Create tables - 2. Stamp baseline - 3. Upgrade to head - 4. Verify current revision is head - """ - # Create tables - async with sqlite_engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - - # Stamp baseline - await run_alembic_stamp(sqlite_engine, '0001_baseline') - - # Upgrade to head - await run_alembic_upgrade(sqlite_engine, 'head') - - # Verify revision - rev = await get_alembic_current(sqlite_engine) - assert rev is not None, 'Expected a revision after upgrade' - # Head should be the latest migration. Resolve the actual head from the - # Alembic script directory instead of hardcoding a revision number, so - # adding a new migration doesn't require editing this assertion. - assert rev == _get_script_head(), f'Expected head {_get_script_head()}, got {rev}' - - @pytest.mark.asyncio - async def test_upgrade_idempotent(self, sqlite_engine): - """ - Running upgrade to head multiple times is idempotent. - - Workflow: - 1. Upgrade to head - 2. Get revision - 3. Upgrade to head again - 4. Verify same revision - """ - # Create tables - async with sqlite_engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - - # Stamp and upgrade - await run_alembic_stamp(sqlite_engine, '0001_baseline') - await run_alembic_upgrade(sqlite_engine, 'head') - - rev1 = await get_alembic_current(sqlite_engine) - - # Upgrade again - should be idempotent - await run_alembic_upgrade(sqlite_engine, 'head') - - rev2 = await get_alembic_current(sqlite_engine) - assert rev2 == rev1, f'Expected {rev1}, got {rev2}' - @pytest.mark.asyncio async def test_upgrade_from_0012_adds_knowledge_base_embedding_dimension(self, sqlite_engine): """The PostgreSQL pgvector revision also evolves the OSS ORM schema.""" @@ -340,87 +251,14 @@ class TestSQLiteMigrationFreshDatabase: assert await get_alembic_current(sqlite_engine) == '0022_codex_credentials' @pytest.mark.asyncio - async def test_fresh_db_upgrade_from_scratch(self, tmp_path): - """ - Fresh database (no tables) can be upgraded directly to head. - - Workflow: - 1. Create fresh engine with new DB file - 2. Create tables - 3. Upgrade to head - 4. Verify revision - """ - # Use different DB file for fresh test - fresh_db_file = tmp_path / 'test_migrations_fresh.db' - fresh_url = f'sqlite+aiosqlite:///{fresh_db_file}' - fresh_engine = create_async_engine(fresh_url) - - # Create tables on fresh DB - async with fresh_engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - - # Upgrade to head directly (no baseline stamp) - await run_alembic_upgrade(fresh_engine, 'head') - - # Verify revision - rev = await get_alembic_current(fresh_engine) - assert rev is not None, 'Expected a revision on fresh DB' - - await fresh_engine.dispose() - - @pytest.mark.asyncio - async def test_fresh_db_without_create_all_behavior(self, tmp_path): - """ - Fresh database without create_all - test actual behavior. - - This tests what happens when migrations run on truly empty DB. - The behavior is determined by Alembic and migration scripts. - - EXPECTED: Either: - 1. Migration succeeds (if scripts handle empty DB) - 2. Migration fails with specific error (if scripts require tables) - - IMPORTANT: This test verifies the ACTUAL behavior, not accepting - any arbitrary failure with try-except pass. - """ - fresh_db_file = tmp_path / 'test_empty_migrations.db' - fresh_url = f'sqlite+aiosqlite:///{fresh_db_file}' - fresh_engine = create_async_engine(fresh_url) - - # Capture the actual behavior - actual_result = None - actual_error = None - - try: - await run_alembic_upgrade(fresh_engine, 'head') - rev = await get_alembic_current(fresh_engine) - actual_result = rev - except Exception as e: - actual_error = e - - await fresh_engine.dispose() - - # Verify specific behavior - one of two outcomes is expected - if actual_result is not None: - # Migration succeeded - verify revision exists - assert actual_result is not None, 'Revision should exist after successful migration' - else: - # Migration failed - verify the error type is known - # Alembic typically raises specific errors for missing tables - assert actual_error is not None, 'Error should be captured if migration failed' - # Log the error type for documentation (don't silently pass) - error_type = type(actual_error).__name__ - # Acceptable error types for empty DB scenarios - acceptable_errors = [ - 'OperationalError', # SQLite table not found - 'ProgrammingError', # SQLAlchemy errors - 'CommandError', # Alembic command errors - ] - assert error_type in acceptable_errors, ( - f'Unexpected error type: {error_type}. ' - f'This may indicate a regression in migration behavior. ' - f'Error: {actual_error}' - ) + async def test_fresh_db_without_create_all_behavior(self, sqlite_engine): + """Bare Alembic reaches head but relies on ORM bootstrap for core tables.""" + await run_alembic_upgrade(sqlite_engine, 'head') + assert await get_alembic_current(sqlite_engine) == _get_script_head() + async with sqlite_engine.connect() as conn: + tables = await conn.run_sync(lambda sync: set(sqlalchemy.inspect(sync).get_table_names())) + assert {'agents', 'runner_state', 'directory_projection_states', 'directory_projection_inbox'} <= tables + assert not {'users', 'bots', 'workspaces'} & tables class TestSQLiteMigrationGetCurrent: diff --git a/tests/integration/persistence/test_migrations_postgres.py b/tests/integration/persistence/test_migrations_postgres.py index ea64d8e1a..cb1bd807a 100644 --- a/tests/integration/persistence/test_migrations_postgres.py +++ b/tests/integration/persistence/test_migrations_postgres.py @@ -424,62 +424,6 @@ class TestPostgreSQLMigrationBaseline: class TestPostgreSQLMigrationUpgrade: """Tests for upgrade to head workflow on PostgreSQL.""" - @pytest.mark.asyncio - async def test_postgres_upgrade_from_baseline_to_head(self, postgres_engine, clean_tables, clean_alembic_version): - """ - Upgrade from baseline to head applies all migrations. - - Workflow: - 1. Create tables - 2. Stamp baseline - 3. Upgrade to head - 4. Verify current revision is head - """ - # Create tables - async with postgres_engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - - # Stamp baseline - await run_alembic_stamp(postgres_engine, '0001_baseline') - - # Upgrade to head - await run_alembic_upgrade(postgres_engine, 'head') - - # Verify revision - rev = await get_alembic_current(postgres_engine) - assert rev is not None, 'Expected a revision after upgrade' - # Head should be the latest migration. Resolve the actual head from the - # Alembic script directory instead of hardcoding a revision number, so - # adding a new migration doesn't require editing this assertion. - assert rev == _get_script_head(), f'Expected head {_get_script_head()}, got {rev}' - - @pytest.mark.asyncio - async def test_postgres_upgrade_idempotent(self, postgres_engine, clean_tables, clean_alembic_version): - """ - Running upgrade to head multiple times is idempotent. - - Workflow: - 1. Upgrade to head - 2. Get revision - 3. Upgrade to head again - 4. Verify same revision - """ - # Create tables - async with postgres_engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - - # Stamp and upgrade - await run_alembic_stamp(postgres_engine, '0001_baseline') - await run_alembic_upgrade(postgres_engine, 'head') - - rev1 = await get_alembic_current(postgres_engine) - - # Upgrade again - should be idempotent - await run_alembic_upgrade(postgres_engine, 'head') - - rev2 = await get_alembic_current(postgres_engine) - assert rev2 == rev1, f'Expected {rev1}, got {rev2}' - @pytest.mark.asyncio @pytest.mark.parametrize('settings_type', ['TEXT', 'JSON']) async def test_legacy_knowledge_restore_statement_accepts_historical_and_fresh_settings_types( @@ -1055,7 +999,27 @@ class TestPostgreSQLTenantRuntime: await rag_manager.load_knowledge_bases_from_db() mcp_loader = MCPLoader(runtime_application) - mcp_loader.host_mcp_server = AsyncMock() + active_hosts = 0 + peak_active_hosts = 0 + peak_host_tasks = 0 + track_hosted_task = mcp_loader.track_hosted_task + + def observe_host_task(task, context): + nonlocal peak_host_tasks + track_hosted_task(task, context) + peak_host_tasks = max(peak_host_tasks, len(mcp_loader._hosted_mcp_tasks)) + + async def host_probe(_context, _config): + nonlocal active_hosts, peak_active_hosts + active_hosts += 1 + peak_active_hosts = max(peak_active_hosts, active_hosts) + try: + await asyncio.sleep(0) + finally: + active_hosts -= 1 + + mcp_loader.track_hosted_task = observe_host_task + mcp_loader.host_mcp_server = AsyncMock(side_effect=host_probe) await mcp_loader.load_mcp_servers_from_db() dispatch_tasks = tuple(mcp_loader._host_dispatch_tasks) if dispatch_tasks: @@ -1096,6 +1060,12 @@ class TestPostgreSQLTenantRuntime: assert len(pipeline_manager.pipelines) == workspace_count assert len(rag_manager.knowledge_bases) == workspace_count assert mcp_loader.host_mcp_server.await_count == workspace_count + assert 0 < peak_active_hosts <= min(workspace_count, 8) + assert 0 < peak_host_tasks <= min(workspace_count, 8) + assert active_hosts == 0 + logging.getLogger('postgres-capacity-runtime-test').info( + 'MCP startup peaks: active_hosts=%d hosted_tasks=%d', peak_active_hosts, peak_host_tasks + ) assert not mcp_loader._host_dispatch_tasks assert not mcp_loader._hosted_mcp_tasks assert len(plugin_handler.reconciled) == workspace_count