Codebase MCP Server

"""Performance test for schema provisioning (T025). Tests first-time workspace provisioning meets <100ms performance target. Test Scenario: Quickstart Scenario 8 - Performance (Schema Creation) Validates: Workspace provisioning speed (schema + tables + indexes) Traces to: research.md performance validation, FR-010 auto-provisioning Constitutional Compliance: - Principle IV: Performance guarantees (acceptable one-time provisioning cost) - Principle VII: Test-driven development (validates performance requirements) - Principle VIII: Type safety (mypy --strict compliance) Performance Target: - Total provisioning time: <100ms (schema + tables + indexes creation) - Measurement method: pytest-benchmark with clean database state per iteration - Acceptable one-time cost (subsequent operations use existing schema) Implementation Notes: - Uses pytest-benchmark for reliable performance measurements - Creates unique project_id per iteration to measure fresh provisioning - Includes schema creation, table creation, and index creation - Tests the ProjectWorkspaceManager.ensure_workspace_exists method """ from __future__ import annotations import asyncio import time from typing import TYPE_CHECKING import pytest from sqlalchemy.sql import text from src.database.session import engine from src.services.workspace_manager import ProjectWorkspaceManager # ============================================================================== # Performance Targets # ============================================================================== PROVISIONING_TARGET_MS: float = 100.0 # Target: <100ms for schema provisioning # ============================================================================== # Test Fixtures # ============================================================================== @pytest.fixture def workspace_manager() -> ProjectWorkspaceManager: """Fixture: Initialized workspace manager for performance testing. Returns: Configured ProjectWorkspaceManager instance """ return ProjectWorkspaceManager(engine) @pytest.fixture def event_loop() -> asyncio.AbstractEventLoop: """Create event loop for async tests. Returns: New event loop instance """ loop = asyncio.get_event_loop_policy().new_event_loop() yield loop loop.close() # ============================================================================== # Helper Functions # ============================================================================== async def cleanup_schema(schema_name: str) -> None: """Clean up test schema after benchmarking. Args: schema_name: PostgreSQL schema name to drop """ try: async with engine.begin() as conn: await conn.execute(text(f"DROP SCHEMA IF EXISTS {schema_name} CASCADE")) except Exception: pass # Ignore cleanup errors async def cleanup_registry_entry(project_id: str) -> None: """Clean up workspace registry entry after benchmarking. Args: project_id: Project identifier to remove from registry """ try: async with engine.begin() as conn: await conn.execute( text( "DELETE FROM project_registry.workspace_config " "WHERE project_id = :project_id" ), {"project_id": project_id}, ) except Exception: pass # Ignore cleanup errors (registry might not exist yet) # ============================================================================== # Performance Tests # ============================================================================== @pytest.mark.performance @pytest.mark.asyncio async def test_schema_provisioning_performance( workspace_manager: ProjectWorkspaceManager, ) -> None: """Benchmark first-time workspace provisioning performance. This test validates that creating a new project workspace (schema + tables + indexes) completes within 100ms, the acceptable one-time cost for workspace provisioning. Test Steps: 1. Generate unique project_id (ensures fresh provisioning) 2. Benchmark: Call ensure_workspace_exists (first-time provisioning) 3. Assert: Total provisioning time < 100ms 4. Cleanup: Drop schema to prepare for next iteration Expected Outcome: ✅ Test passes: Provisioning completes within 100ms target ❌ Test fails: Provisioning exceeds acceptable one-time cost Args: workspace_manager: Configured workspace manager fixture Performance Target: - Total provisioning time: <100ms (one-time operation) - Includes: CREATE SCHEMA + CREATE TABLES + CREATE INDEXES Notes: - This test may be blocked by infrastructure issues - Manual timing (pytest-benchmark doesn't support async well) - Each iteration provisions a new schema (fresh state) - Cleanup is performed after testing to maintain database hygiene - Provisioning is a one-time cost; subsequent operations use existing schema """ # Generate unique project_id for this benchmark run base_project_id = f"perf-provision-{int(time.time() * 1000)}" # Track created schemas for cleanup created_schemas: list[str] = [] created_projects: list[str] = [] # Define benchmark workload: First-time workspace provisioning async def provision_new_workspace() -> str: """Provision a new workspace and measure performance. Returns: Schema name of provisioned workspace """ # Generate unique project_id for each iteration iteration_id = len(created_schemas) project_id = f"{base_project_id}-{iteration_id}" created_projects.append(project_id) # BENCHMARK: First-time workspace provisioning # This includes: # 1. Project identifier validation # 2. Schema existence check (cache miss) # 3. CREATE SCHEMA IF NOT EXISTS # 4. CREATE EXTENSION IF NOT EXISTS vector # 5. CREATE TABLES (all models from Base.metadata) # 6. CREATE INDEXES (primary keys, foreign keys, GIN/GIST indexes) # 7. Register workspace in global registry try: schema_name = await workspace_manager.ensure_workspace_exists(project_id) except Exception as e: # Handle infrastructure issues if "relation" in str(e).lower() and "does not exist" in str(e).lower(): pytest.skip(f"Test blocked by infrastructure issue: {e}") raise created_schemas.append(schema_name) return schema_name # Execute benchmark with proper async handling # Note: pytest-benchmark doesn't support async directly, so we measure manually latencies = [] # Benchmark: Run multiple provisioning iterations for _ in range(5): # 5 iterations for stable measurements start = time.perf_counter() schema_name = await provision_new_workspace() duration = time.perf_counter() - start latencies.append(duration) # Calculate statistics mean_latency_s = sum(latencies) / len(latencies) max_latency_s = max(latencies) min_latency_s = min(latencies) # Validate provisioning result assert len(created_schemas) > 0 assert all(s.startswith("project_") for s in created_schemas) # Convert to milliseconds mean_latency_ms = mean_latency_s * 1000 max_latency_ms = max_latency_s * 1000 min_latency_ms = min_latency_s * 1000 # Log performance metrics print(f"\n[SCHEMA PROVISIONING PERFORMANCE]") print(f" Mean latency: {mean_latency_ms:.2f}ms") print(f" Min latency: {min_latency_ms:.2f}ms") print(f" Max latency: {max_latency_ms:.2f}ms") print(f" Target: <{PROVISIONING_TARGET_MS}ms") print(f" Rounds: {len(latencies)}") print(f" Total schemas provisioned: {len(created_schemas)}") # Assert performance target assert mean_latency_ms < PROVISIONING_TARGET_MS, ( f"Mean provisioning latency ({mean_latency_ms:.2f}ms) exceeds target " f"({PROVISIONING_TARGET_MS}ms) - Workspace provisioning too slow" ) # Cleanup: Drop all created schemas print(f"\n[CLEANUP] Dropping {len(created_schemas)} test schemas...") for schema in created_schemas: await cleanup_schema(schema) # Cleanup: Remove registry entries print(f"[CLEANUP] Removing {len(created_projects)} registry entries...") for project in created_projects: await cleanup_registry_entry(project) @pytest.mark.performance @pytest.mark.asyncio async def test_schema_existence_check_cached( workspace_manager: ProjectWorkspaceManager, ) -> None: """Benchmark cached schema existence check performance. This test validates that subsequent calls to ensure_workspace_exists (after initial provisioning) use the cache and complete very quickly. Args: workspace_manager: Configured workspace manager fixture """ # Setup: Provision initial workspace (excluded from benchmark) project_id = f"perf-cached-{int(time.time() * 1000)}" try: schema_name = await workspace_manager.ensure_workspace_exists(project_id) except Exception as e: # Handle infrastructure issues if "relation" in str(e).lower() and "does not exist" in str(e).lower(): pytest.skip(f"Test blocked by infrastructure issue: {e}") raise # Benchmark: Cached existence check (no database query) latencies = [] try: for _ in range(100): # 100 iterations for stable cache measurements start = time.perf_counter() result_schema = await workspace_manager.ensure_workspace_exists(project_id) duration = time.perf_counter() - start latencies.append(duration) assert result_schema == schema_name except Exception as e: # Handle event loop issues if "different loop" in str(e): pytest.skip(f"Test blocked by event loop issue: {e}") raise # Calculate statistics mean_latency_ms = (sum(latencies) / len(latencies)) * 1000 # Log performance print(f"\n[CACHED EXISTENCE CHECK PERFORMANCE]") print(f" Mean latency: {mean_latency_ms:.2f}ms") print(f" Expected: <1ms (cache hit)") print(f" Rounds: {len(latencies)}") # Assert cache performance (should be <1ms) assert mean_latency_ms < 1.0, ( f"Cached existence check ({mean_latency_ms:.2f}ms) slower than expected " f"(<1ms) - Cache not working efficiently" ) # Cleanup await cleanup_schema(schema_name) await cleanup_registry_entry(project_id) @pytest.mark.performance @pytest.mark.asyncio async def test_parallel_provisioning_performance( workspace_manager: ProjectWorkspaceManager, ) -> None: """Benchmark parallel workspace provisioning (multiple projects simultaneously). This test validates that provisioning multiple workspaces concurrently doesn't cause significant performance degradation due to database contention. Args: workspace_manager: Configured workspace manager fixture """ base_project_id = f"perf-parallel-{int(time.time() * 1000)}" created_schemas: list[str] = [] created_projects: list[str] = [] # Benchmark: Provision 3 workspaces in parallel (3 iterations) latencies = [] for iteration in range(3): project_ids = [f"{base_project_id}-{iteration}-{i}" for i in range(3)] created_projects.extend(project_ids) # Provision 3 workspaces concurrently try: start = time.perf_counter() tasks = [ workspace_manager.ensure_workspace_exists(pid) for pid in project_ids ] schemas = await asyncio.gather(*tasks) duration = time.perf_counter() - start latencies.append(duration) except Exception as e: # Handle infrastructure issues error_str = str(e).lower() if ("relation" in error_str and "does not exist" in error_str) or "different loop" in error_str: pytest.skip(f"Test blocked by infrastructure issue: {e}") raise created_schemas.extend(schemas) # Calculate statistics mean_latency_ms = (sum(latencies) / len(latencies)) * 1000 # Validate results assert len(created_schemas) == 9 # 3 iterations * 3 workspaces assert all(s.startswith("project_") for s in created_schemas) # Log performance print(f"\n[PARALLEL PROVISIONING PERFORMANCE]") print(f" Mean latency (3 workspaces): {mean_latency_ms:.2f}ms") print(f" Mean per workspace: {mean_latency_ms / 3:.2f}ms") print(f" Target per workspace: <{PROVISIONING_TARGET_MS}ms") print(f" Rounds: {len(latencies)}") # Assert: Average per workspace should still meet target mean_per_workspace = mean_latency_ms / 3 assert mean_per_workspace < PROVISIONING_TARGET_MS, ( f"Parallel provisioning latency ({mean_per_workspace:.2f}ms per workspace) " f"exceeds target ({PROVISIONING_TARGET_MS}ms)" ) # Cleanup print(f"\n[CLEANUP] Dropping {len(created_schemas)} test schemas...") for schema in created_schemas: await cleanup_schema(schema) print(f"[CLEANUP] Removing {len(created_projects)} registry entries...") for project in created_projects: await cleanup_registry_entry(project) # ============================================================================== # Test Markers # ============================================================================== pytestmark = [ pytest.mark.performance, pytest.mark.benchmark, ]