Codebase MCP Server

"""Performance test for project switching latency (T024). Tests project switching meets Constitutional Principle IV target: <50ms latency. Test Scenario: Quickstart Scenario 8 - Performance Validates: Project switching via search operations Traces to: Constitutional Principle IV, Technical Context performance goals Constitutional Compliance: - Principle IV: Performance guarantees (<50ms switching latency target) - Principle VII: Test-driven development (validates performance requirements) - Principle VIII: Type safety (mypy --strict compliance) Performance Targets: - Mean latency: <50ms per project switch - Max latency: <150ms per project switch (p99) - Measurement method: pytest-benchmark with 10 warmup iterations Implementation Notes: - Uses pytest-benchmark for reliable performance measurements - Warmup phase excludes initial indexing from measurements - Measures complete search operation (includes project switching overhead) - Tests rapid back-and-forth switching pattern """ from __future__ import annotations import asyncio from pathlib import Path from typing import TYPE_CHECKING, Any import pytest from src.mcp.tools.indexing import index_repository from src.mcp.tools.search import search_code if TYPE_CHECKING: from pytest_benchmark.fixture import BenchmarkFixture # ============================================================================== # Performance Targets (Constitutional Principle IV) # ============================================================================== MEAN_LATENCY_TARGET_MS: float = 50.0 # Mean switching latency target MAX_LATENCY_TARGET_MS: float = 150.0 # Max switching latency target (p99) # ============================================================================== # Test Fixtures # ============================================================================== @pytest.fixture def repo_a(tmp_path: Path) -> str: """Fixture: Project A codebase for performance testing. Args: tmp_path: Pytest fixture providing temporary directory Returns: Absolute path to repository directory """ repo_dir = tmp_path / "perf-repo-a" repo_dir.mkdir() (repo_dir / "auth.py").write_text( """ def authenticate_user_a(username: str, password: str) -> bool: '''Project A authentication implementation''' return validate_credentials_a(username, password) def validate_credentials_a(username: str, password: str) -> bool: '''Validate user credentials for Project A''' return check_password_hash(username, password) def check_password_hash(username: str, password: str) -> bool: '''Check password hash against database''' return True """ ) (repo_dir / "database.py").write_text( """ def query_users_a(filter_active: bool = True) -> list: '''Query users from Project A database''' return execute_query_a("SELECT * FROM users WHERE active = ?", [filter_active]) def execute_query_a(sql: str, params: list) -> list: '''Execute SQL query in Project A''' return [] """ ) return str(repo_dir) @pytest.fixture def repo_b(tmp_path: Path) -> str: """Fixture: Project B codebase for performance testing. Args: tmp_path: Pytest fixture providing temporary directory Returns: Absolute path to repository directory """ repo_dir = tmp_path / "perf-repo-b" repo_dir.mkdir() (repo_dir / "auth.py").write_text( """ def authenticate_user_b(token: str) -> bool: '''Project B authentication implementation using token''' return verify_token_b(token) def verify_token_b(token: str) -> bool: '''Verify authentication token for Project B''' return validate_jwt_signature(token) def validate_jwt_signature(token: str) -> bool: '''Validate JWT signature''' return True """ ) (repo_dir / "database.py").write_text( """ def query_users_b(filter_role: str = "admin") -> list: '''Query users from Project B database''' return execute_query_b("SELECT * FROM users WHERE role = ?", [filter_role]) def execute_query_b(sql: str, params: list) -> list: '''Execute SQL query in Project B''' return [] """ ) return str(repo_dir) @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() # ============================================================================== # Performance Tests # ============================================================================== @pytest.mark.performance @pytest.mark.benchmark @pytest.mark.asyncio async def test_project_switching_latency( repo_a: str, repo_b: str, benchmark: BenchmarkFixture ) -> None: """Benchmark project switching latency (Constitutional Principle IV). This test validates that switching between projects meets the <50ms mean latency target defined in Constitutional Principle IV. Test Steps: 1. Setup: Index two distinct projects (excluded from benchmark) 2. Warmup: Perform initial queries to warm caches 3. Benchmark: Switch between projects and measure latency 4. Assert: Mean latency < 50ms, max latency < 150ms Expected Outcome: ✅ Test passes: Switching latency meets performance targets ❌ Test fails: Latency exceeds targets (Constitutional violation) ⏭️ Test skipped: Infrastructure not ready (CodeChunk.project_id missing) Args: repo_a: Path to Project A repository fixture repo_b: Path to Project B repository fixture benchmark: pytest-benchmark fixture for measurements Performance Targets: - Mean latency: <50ms (Constitutional Principle IV) - Max latency: <150ms (p99 threshold) Notes: - This test may be blocked by CodeChunk.project_id infrastructure issues - Benchmark includes full search operation (not just switching overhead) - Uses pytest-benchmark for reliable measurements with warmup """ # Setup: Index both repositories (excluded from benchmark timing) try: result_a = await index_repository.fn( repo_path=repo_a, project_id="perf-project-a", force_reindex=False, ctx=None, ) # Check if indexing failed due to infrastructure issue if result_a["status"] == "failed": error_msg = " ".join(result_a.get("errors", [])) if "null value in column \"project_id\"" in error_msg: pytest.skip( f"Test blocked by infrastructure issue: CodeChunk.project_id column " f"migration not applied yet. This test validates Constitutional " f"Principle IV performance targets and will run once infrastructure " f"is ready." ) assert result_a["status"] == "success", f"Project A indexing failed: {result_a}" result_b = await index_repository.fn( repo_path=repo_b, project_id="perf-project-b", force_reindex=False, ctx=None, ) assert result_b["status"] == "success", f"Project B indexing failed: {result_b}" except Exception as e: # Handle other infrastructure issues if "null value in column \"project_id\"" in str(e) or "different loop" in str(e): pytest.skip(f"Test blocked by infrastructure issue: {e}") raise # Warmup: Perform initial queries to warm caches (excluded from benchmark) warmup_result_a = await search_code.fn( query="authentication", project_id="perf-project-a", limit=5, ctx=None, ) assert len(warmup_result_a["results"]) > 0, "Warmup query A should find results" warmup_result_b = await search_code.fn( query="authentication", project_id="perf-project-b", limit=5, ctx=None, ) assert len(warmup_result_b["results"]) > 0, "Warmup query B should find results" # Define benchmark workload: Rapid switching between projects async def switch_between_projects() -> tuple[dict[str, Any], dict[str, Any]]: """Execute project switching workload. Returns: Tuple of (result_a, result_b) from search operations """ # Switch to project-a result_a = await search_code.fn( query="authentication", project_id="perf-project-a", limit=5, ctx=None, ) # Switch to project-b result_b = await search_code.fn( query="authentication", project_id="perf-project-b", limit=5, ctx=None, ) return (result_a, result_b) # Synchronous wrapper for pytest-benchmark def sync_wrapper() -> tuple[dict[str, Any], dict[str, Any]]: """Synchronous wrapper for async function. Returns: Search results from async operation """ loop = asyncio.get_event_loop() return loop.run_until_complete(switch_between_projects()) # Execute benchmark results = benchmark(sync_wrapper) result_a, result_b = results # Validate search results assert result_a["project_id"] == "perf-project-a" assert result_b["project_id"] == "perf-project-b" assert len(result_a["results"]) > 0 assert len(result_b["results"]) > 0 # Get benchmark statistics stats = benchmark.stats mean_latency_s = stats["mean"] max_latency_s = stats["max"] # Convert to milliseconds mean_latency_ms = mean_latency_s * 1000 max_latency_ms = max_latency_s * 1000 # Assert performance targets (Constitutional Principle IV) # Note: Benchmark measures TWO searches (one per project), so divide by 2 mean_per_switch = mean_latency_ms / 2 max_per_switch = max_latency_ms / 2 # Log performance metrics print(f"\n[PROJECT SWITCHING PERFORMANCE]") print(f" Mean latency per switch: {mean_per_switch:.2f}ms") print(f" Max latency per switch: {max_per_switch:.2f}ms") print(f" Target mean: <{MEAN_LATENCY_TARGET_MS}ms") print(f" Target max: <{MAX_LATENCY_TARGET_MS}ms") print(f" Rounds: {stats['rounds']}") # Assert targets assert mean_per_switch < MEAN_LATENCY_TARGET_MS, ( f"Mean switching latency ({mean_per_switch:.2f}ms) exceeds target " f"({MEAN_LATENCY_TARGET_MS}ms) - Constitutional Principle IV violation" ) assert max_per_switch < MAX_LATENCY_TARGET_MS, ( f"Max switching latency ({max_per_switch:.2f}ms) exceeds target " f"({MAX_LATENCY_TARGET_MS}ms) - Constitutional Principle IV violation" ) @pytest.mark.performance @pytest.mark.benchmark @pytest.mark.asyncio async def test_rapid_switching_stability( repo_a: str, repo_b: str, benchmark: BenchmarkFixture ) -> None: """Test stability of project switching under rapid back-and-forth pattern. This test validates that rapid switching maintains consistent performance without degradation over many iterations. Args: repo_a: Path to Project A repository fixture repo_b: Path to Project B repository fixture benchmark: pytest-benchmark fixture for measurements """ # Setup: Index both repositories try: result_a = await index_repository.fn(repo_path=repo_a, project_id="rapid-a") if result_a["status"] == "failed": error_msg = " ".join(result_a.get("errors", [])) if "null value in column \"project_id\"" in error_msg: pytest.skip(f"Test blocked by infrastructure issue: CodeChunk.project_id column missing") result_b = await index_repository.fn(repo_path=repo_b, project_id="rapid-b") if result_b["status"] == "failed": error_msg = " ".join(result_b.get("errors", [])) if "null value in column \"project_id\"" in error_msg: pytest.skip(f"Test blocked by infrastructure issue: CodeChunk.project_id column missing") except Exception as e: # Handle infrastructure issues (CodeChunk.project_id column missing or event loop issues) if "null value in column \"project_id\"" in str(e) or "different loop" in str(e): pytest.skip(f"Test blocked by infrastructure issue: {e}") raise # Warmup await search_code.fn(query="authentication", project_id="rapid-a", limit=1) await search_code.fn(query="authentication", project_id="rapid-b", limit=1) # Benchmark: Rapid switching pattern (5 switches) async def rapid_switching_pattern() -> list[str]: """Execute rapid switching pattern. Returns: List of project_ids from search results """ project_ids: list[str] = [] for _ in range(5): result_a = await search_code.fn( query="authentication", project_id="rapid-a", limit=1 ) project_ids.append(result_a["project_id"]) result_b = await search_code.fn( query="authentication", project_id="rapid-b", limit=1 ) project_ids.append(result_b["project_id"]) return project_ids def sync_wrapper() -> list[str]: loop = asyncio.get_event_loop() return loop.run_until_complete(rapid_switching_pattern()) # Execute benchmark project_ids = benchmark(sync_wrapper) # Validate switching worked correctly assert len(project_ids) == 10 # 5 iterations * 2 projects assert project_ids[0::2] == ["rapid-a"] * 5 # Even indices assert project_ids[1::2] == ["rapid-b"] * 5 # Odd indices # Log performance stats = benchmark.stats mean_latency_ms = stats["mean"] * 1000 print(f"\n[RAPID SWITCHING STABILITY]") print(f" Mean latency (10 switches): {mean_latency_ms:.2f}ms") print(f" Mean per switch: {mean_latency_ms / 10:.2f}ms") # ============================================================================== # Test Markers # ============================================================================== pytestmark = [ pytest.mark.performance, pytest.mark.benchmark, ]