"""
Integration tests for performance validation (Scenario 7).
Tests verify (from quickstart.md):
- Performance requirements (FR-034, FR-035)
- Indexing: 10,000 files in <60 seconds
- Search: p95 latency <500ms
- Load testing with concurrent operations
- Performance regression detection
TDD Compliance: These tests MUST FAIL initially since services are not implemented yet.
"""
from __future__ import annotations
import asyncio
import random
import time
from pathlib import Path
from typing import TYPE_CHECKING
import pytest
from sqlalchemy.ext.asyncio import AsyncSession
if TYPE_CHECKING:
from src.services.indexer import IndexResult
from src.services.searcher import SearchResult
@pytest.fixture
async def large_test_repository(tmp_path: Path) -> Path:
"""
Generate a large test repository with 10,000 Python files.
Each file contains realistic Python code with functions, classes,
and docstrings for comprehensive indexing and search testing.
"""
repo_path = tmp_path / "large-repo"
repo_path.mkdir()
# Create directory structure
for i in range(100):
module_dir = repo_path / f"module_{i:03d}"
module_dir.mkdir()
# Create 100 files per module (100 * 100 = 10,000 files)
for j in range(100):
file_path = module_dir / f"file_{j:03d}.py"
# Generate realistic Python code
code = f'''"""Module {i} File {j} - Generated test code."""
import os
import sys
from typing import List, Dict, Optional
class DataProcessor:
"""Process data with various transformations."""
def __init__(self, name: str = "processor_{i}_{j}") -> None:
"""Initialize processor with name."""
self.name = name
self.data: List[int] = []
def add_data(self, value: int) -> None:
"""Add a value to the data list."""
self.data.append(value)
def process(self) -> int:
"""Process the data and return sum."""
return sum(self.data)
def transform(self, multiplier: int = 2) -> List[int]:
"""Transform data by multiplying each value."""
return [x * multiplier for x in self.data]
def calculate_average(numbers: List[int]) -> float:
"""Calculate the average of a list of numbers."""
if not numbers:
return 0.0
return sum(numbers) / len(numbers)
def find_maximum(numbers: List[int]) -> Optional[int]:
"""Find the maximum value in a list."""
if not numbers:
return None
return max(numbers)
def filter_positive(numbers: List[int]) -> List[int]:
"""Filter out negative numbers from the list."""
return [n for n in numbers if n > 0]
def main() -> None:
"""Main entry point for module {i} file {j}."""
processor = DataProcessor()
processor.add_data(10)
processor.add_data(20)
result = processor.process()
print(f"Result: {{result}}")
if __name__ == "__main__":
main()
'''
file_path.write_text(code)
return repo_path
@pytest.fixture
async def db_session() -> AsyncSession:
"""Create async database session for tests."""
pytest.skip("Database session fixture not implemented yet (requires T019-T027)")
@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.slow # Mark as slow test (can be skipped in CI)
async def test_indexing_10k_files_under_60_seconds_not_implemented(
large_test_repository: Path,
db_session: AsyncSession,
) -> None:
"""
Test indexing 10,000 files in <60 seconds - NOT YET IMPLEMENTED.
Performance requirement from spec:
- 10,000 files indexed in <60 seconds
- This is a critical performance target
Expected workflow:
1. Index repository with 10,000 files
2. Measure total indexing time
3. Verify time <60 seconds
4. Verify all files indexed
5. Verify all chunks created
This test MUST FAIL until T031 (indexer optimization) is implemented.
"""
pytest.skip("Indexer service not implemented yet (T031)")
# Future implementation:
# from src.services.indexer import index_repository
#
# start_time = time.perf_counter()
#
# result = await index_repository(
# path=large_test_repository,
# name="Large Test Repository",
# force_reindex=True,
# )
#
# duration = time.perf_counter() - start_time
#
# # Verify performance target
# assert duration < 60, f"Indexing took {duration:.2f}s, exceeds 60s target"
#
# # Verify indexing success
# assert result.status == "success"
# assert result.files_indexed == 10_000
# assert result.chunks_created > 50_000 # ~5+ chunks per file
# assert result.duration_seconds < 60
@pytest.mark.integration
@pytest.mark.asyncio
async def test_search_p95_latency_under_500ms_not_implemented(
large_test_repository: Path,
db_session: AsyncSession,
) -> None:
"""
Test search p95 latency <500ms - NOT YET IMPLEMENTED.
Performance requirement from spec:
- Search p95 latency <500ms
- This ensures responsive user experience
Expected workflow:
1. Index large repository (prerequisite)
2. Run 100 diverse search queries
3. Measure latency for each
4. Calculate p50, p95, p99 percentiles
5. Verify p95 <500ms
This test MUST FAIL until T032 (searcher optimization) is implemented.
"""
pytest.skip("Search service not implemented yet (T032)")
# Future implementation:
# from src.services.indexer import index_repository
# from src.services.searcher import search_code
#
# # Index repository first
# await index_repository(
# path=large_test_repository,
# name="Large Test Repository",
# force_reindex=True,
# )
#
# # Generate diverse search queries
# queries = [
# "calculate average of numbers",
# "find maximum value in list",
# "filter positive numbers",
# "data processor class",
# "transform data with multiplier",
# "process data and return sum",
# "initialize processor with name",
# "add value to data list",
# "main entry point",
# "import statements",
# ] * 10 # 100 total queries
#
# latencies: list[float] = []
#
# for query in queries:
# start_time = time.perf_counter()
# await search_code(query=query, limit=10)
# latency_ms = (time.perf_counter() - start_time) * 1000
# latencies.append(latency_ms)
#
# # Calculate percentiles
# latencies.sort()
# p50 = latencies[len(latencies) // 2]
# p95 = latencies[int(len(latencies) * 0.95)]
# p99 = latencies[int(len(latencies) * 0.99)]
#
# # Verify performance targets
# assert p95 < 500, f"P95 latency {p95:.0f}ms exceeds 500ms target"
# assert p99 < 1000, f"P99 latency {p99:.0f}ms exceeds 1000ms target"
#
# # Log performance metrics
# print(f"Search performance: P50={p50:.0f}ms, P95={p95:.0f}ms, P99={p99:.0f}ms")
@pytest.mark.integration
@pytest.mark.asyncio
async def test_concurrent_search_operations(
large_test_repository: Path,
db_session: AsyncSession,
) -> None:
"""
Test concurrent search operations maintain performance.
Expected behavior:
- Run 10 concurrent search queries
- Measure latency for each
- Verify all complete successfully
- Verify average latency acceptable
This test MUST FAIL until T032 (searcher) is implemented.
"""
pytest.skip("Search service not implemented yet (T032)")
# Future implementation:
# from src.services.indexer import index_repository
# from src.services.searcher import search_code
#
# # Index repository
# await index_repository(
# path=large_test_repository,
# name="Large Test Repository",
# force_reindex=True,
# )
#
# async def search_task(query: str) -> float:
# """Perform search and return latency in ms."""
# start_time = time.perf_counter()
# await search_code(query=query, limit=10)
# return (time.perf_counter() - start_time) * 1000
#
# # Run 10 concurrent searches
# queries = [f"search query {i}" for i in range(10)]
# tasks = [search_task(q) for q in queries]
#
# latencies = await asyncio.gather(*tasks)
#
# # Verify all completed
# assert len(latencies) == 10
#
# # Verify average latency reasonable
# avg_latency = sum(latencies) / len(latencies)
# assert avg_latency < 1000, f"Average concurrent latency {avg_latency:.0f}ms too high"
@pytest.mark.integration
@pytest.mark.asyncio
async def test_incremental_update_performance(
large_test_repository: Path,
db_session: AsyncSession,
) -> None:
"""
Test incremental update performance on large repository.
Expected behavior:
- Index 10,000 files
- Modify 10 files
- Incremental update should be much faster than full index
- Should complete in <10 seconds
This test MUST FAIL until T031 (indexer with incremental logic) is implemented.
"""
pytest.skip("Incremental indexing not implemented yet (T031)")
# Future implementation:
# from src.services.indexer import index_repository
#
# # Initial full index
# await index_repository(
# path=large_test_repository,
# name="Large Test Repository",
# force_reindex=True,
# )
#
# # Modify 10 files
# for i in range(10):
# file_path = large_test_repository / f"module_000" / f"file_{i:03d}.py"
# content = file_path.read_text()
# file_path.write_text(content + "\n# Modified\n")
#
# # Incremental update
# start_time = time.perf_counter()
# result = await index_repository(
# path=large_test_repository,
# name="Large Test Repository",
# force_reindex=False,
# )
# duration = time.perf_counter() - start_time
#
# # Verify performance
# assert duration < 10, f"Incremental update took {duration:.2f}s, should be <10s"
# assert result.files_indexed == 10 # Only modified files
@pytest.mark.integration
@pytest.mark.asyncio
async def test_memory_usage_during_large_indexing(
large_test_repository: Path,
db_session: AsyncSession,
) -> None:
"""
Test memory usage stays reasonable during large repository indexing.
Expected behavior:
- Index 10,000 files
- Memory usage should not grow unbounded
- Verify batching and streaming work correctly
This test MUST FAIL until T031 (indexer with batching) is implemented.
"""
pytest.skip("Indexer batching not implemented yet (T031)")
# Future implementation:
# import psutil
# import os
# from src.services.indexer import index_repository
#
# process = psutil.Process(os.getpid())
# initial_memory = process.memory_info().rss / 1024 / 1024 # MB
#
# # Index large repository
# await index_repository(
# path=large_test_repository,
# name="Large Test Repository",
# force_reindex=True,
# )
#
# final_memory = process.memory_info().rss / 1024 / 1024 # MB
# memory_increase = final_memory - initial_memory
#
# # Verify memory usage reasonable (< 1GB increase)
# assert memory_increase < 1024, f"Memory increased by {memory_increase:.0f}MB, too high"
@pytest.mark.integration
@pytest.mark.asyncio
async def test_database_query_performance(
large_test_repository: Path,
db_session: AsyncSession,
) -> None:
"""
Test database query performance with large dataset.
Expected behavior:
- Index 10,000 files (~50,000+ chunks)
- Run complex database queries
- Verify queries complete in reasonable time
- Verify indexes are working
This test MUST FAIL until T026 (migration with indexes) is implemented.
"""
pytest.skip("Database indexes not implemented yet (T026)")
# Future implementation:
# from src.services.indexer import index_repository
# from src.models.code_chunk import CodeChunk
# from sqlalchemy import select, func
#
# # Index repository
# await index_repository(
# path=large_test_repository,
# name="Large Test Repository",
# force_reindex=True,
# )
#
# # Test query: Count all chunks
# start_time = time.perf_counter()
# stmt = select(func.count()).select_from(CodeChunk)
# result = await db_session.scalar(stmt)
# duration_ms = (time.perf_counter() - start_time) * 1000
#
# assert result > 50_000 # Should have many chunks
# assert duration_ms < 100, f"Count query took {duration_ms:.0f}ms, too slow"
@pytest.mark.integration
@pytest.mark.asyncio
async def test_embedding_generation_throughput(
large_test_repository: Path,
db_session: AsyncSession,
) -> None:
"""
Test embedding generation throughput with batching.
Expected behavior:
- Generate embeddings for 50,000+ chunks
- Batching should improve throughput
- Verify reasonable generation rate
This test MUST FAIL until T030 (embedder with batching) is implemented.
"""
pytest.skip("Embedder batching not implemented yet (T030)")
# Future implementation:
# from src.services.indexer import index_repository
#
# start_time = time.perf_counter()
#
# result = await index_repository(
# path=large_test_repository,
# name="Large Test Repository",
# force_reindex=True,
# )
#
# duration = time.perf_counter() - start_time
#
# # Calculate throughput
# chunks_per_second = result.chunks_created / duration
#
# # Verify reasonable throughput (>800 chunks/sec with batching)
# assert chunks_per_second > 800, (
# f"Throughput {chunks_per_second:.0f} chunks/sec too low, "
# "batching may not be working"
# )
@pytest.mark.integration
@pytest.mark.asyncio
async def test_search_result_limit_performance(
large_test_repository: Path,
db_session: AsyncSession,
) -> None:
"""
Test that search limit parameter affects performance appropriately.
Expected behavior:
- Search with limit=10 should be fast
- Search with limit=100 should be slower but still reasonable
- Verify limit is applied at database level (not post-filtering)
This test MUST FAIL until T032 (searcher) is implemented.
"""
pytest.skip("Search service not implemented yet (T032)")
# Future implementation:
# from src.services.indexer import index_repository
# from src.services.searcher import search_code
#
# # Index repository
# await index_repository(
# path=large_test_repository,
# name="Large Test Repository",
# force_reindex=True,
# )
#
# # Search with small limit
# start_time = time.perf_counter()
# results_small = await search_code(query="function", limit=10)
# latency_small = (time.perf_counter() - start_time) * 1000
#
# # Search with larger limit
# start_time = time.perf_counter()
# results_large = await search_code(query="function", limit=100)
# latency_large = (time.perf_counter() - start_time) * 1000
#
# # Verify limits respected
# assert len(results_small) <= 10
# assert len(results_large) <= 100
#
# # Verify performance reasonable for both
# assert latency_small < 500, f"Small limit search took {latency_small:.0f}ms"
# assert latency_large < 1000, f"Large limit search took {latency_large:.0f}ms"
@pytest.mark.integration
@pytest.mark.asyncio
async def test_performance_regression_baseline(
large_test_repository: Path,
db_session: AsyncSession,
) -> None:
"""
Test performance baseline for regression detection.
This test establishes performance baselines:
- Indexing time for 10K files
- Search p95 latency
- Memory usage
- Database query performance
Expected behavior:
- All metrics within expected ranges
- Can be used to detect regressions in future
This test MUST FAIL until T031, T032 are implemented.
"""
pytest.skip("Indexer and searcher not implemented yet (T031, T032)")
# Future implementation:
# from src.services.indexer import index_repository
# from src.services.searcher import search_code
#
# # Baseline: Indexing
# start_time = time.perf_counter()
# result = await index_repository(
# path=large_test_repository,
# name="Large Test Repository",
# force_reindex=True,
# )
# indexing_time = time.perf_counter() - start_time
#
# # Baseline: Search
# queries = ["function", "class", "data", "process", "calculate"] * 20
# latencies = []
# for query in queries:
# start_time = time.perf_counter()
# await search_code(query=query, limit=10)
# latencies.append((time.perf_counter() - start_time) * 1000)
#
# latencies.sort()
# search_p95 = latencies[int(len(latencies) * 0.95)]
#
# # Print baselines for reference
# print(f"Performance Baseline:")
# print(f" Indexing (10K files): {indexing_time:.2f}s")
# print(f" Search P95: {search_p95:.0f}ms")
# print(f" Files indexed: {result.files_indexed}")
# print(f" Chunks created: {result.chunks_created}")
#
# # Verify baselines
# assert indexing_time < 60
# assert search_p95 < 500
