"""
Shared performance testing utilities
This module provides common utilities and dataclasses for performance testing
across different test modules to avoid import issues.
"""
import time
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
@dataclass
class PerformanceMetric:
"""Individual performance measurement"""
operation: str
duration: float
memory_usage: float
cpu_usage: float
success: bool = True
error_message: Optional[str] = None
@dataclass
class PerformanceResult:
"""Result of a performance test"""
operation: str
duration: float
memory_usage: float
cpu_usage: float
metrics: List[PerformanceMetric]
success: bool = True
error_message: Optional[str] = None
@dataclass
class PerformanceBaseline:
"""Performance baseline configuration"""
operation: str
max_duration: float
max_memory: float
max_cpu: float
min_success_rate: float = 0.95
class PerformanceMonitor:
"""Monitor system performance during operations"""
def __init__(self):
self.start_time = None
self.start_memory = None
self.start_cpu = None
self.peak_memory = None
self.peak_cpu = None
self.process = None
def start(self):
"""Start monitoring"""
self.start_time = time.time()
try:
import psutil
self.process = psutil.Process()
self.start_memory = self.process.memory_info().rss / 1024 / 1024 # MB
self.start_cpu = self.process.cpu_percent()
self.peak_memory = self.start_memory
self.peak_cpu = self.start_cpu
except ImportError:
self.start_memory = 1.0 # Default value for testing
self.start_cpu = 1.0
self.peak_memory = 1.0
self.peak_cpu = 1.0
def update(self):
"""Update peak values"""
try:
if self.process:
current_memory = self.process.memory_info().rss / 1024 / 1024 # MB
current_cpu = self.process.cpu_percent()
self.peak_memory = max(self.peak_memory, current_memory)
self.peak_cpu = max(self.peak_cpu, current_cpu)
except ImportError:
pass
def stop(self) -> PerformanceResult:
"""Stop monitoring and return results"""
end_time = time.time()
duration = end_time - self.start_time
try:
if self.process:
end_memory = self.process.memory_info().rss / 1024 / 1024 # MB
end_cpu = self.process.cpu_percent()
else:
# Use realistic default values when psutil is not available
end_memory = max(1.0, self.start_memory + 0.1)
end_cpu = max(1.0, self.start_cpu + 0.1)
except ImportError:
end_memory = max(1.0, self.start_memory + 0.1)
end_cpu = max(1.0, self.start_cpu + 0.1)
return PerformanceResult(
operation="monitored_operation",
duration=duration,
memory_usage=max(end_memory, 1.0),
cpu_usage=max(end_cpu, 1.0),
metrics=[]
)
def load_baselines() -> Dict[str, PerformanceBaseline]:
"""Load performance baselines from configuration"""
baselines = {
"ripgrep_simple_search": PerformanceBaseline(
operation="ripgrep_simple_search",
max_duration=0.5,
max_memory=50,
max_cpu=30
),
"ripgrep_complex_pattern": PerformanceBaseline(
operation="ripgrep_complex_pattern",
max_duration=1.0,
max_memory=100,
max_cpu=50
),
"ast_grep_pattern_match": PerformanceBaseline(
operation="ast_grep_pattern_match",
max_duration=0.2,
max_memory=30,
max_cpu=25
),
"ast_grep_rule_execution": PerformanceBaseline(
operation="ast_grep_rule_execution",
max_duration=0.5,
max_memory=80,
max_cpu=40
),
"mcp_tool_operation": PerformanceBaseline(
operation="mcp_tool_operation",
max_duration=1.0,
max_memory=150,
max_cpu=60
)
}
return baselines
def compare_with_baseline(result: PerformanceResult, baseline: PerformanceBaseline) -> Dict[str, Any]:
"""Compare performance result with baseline"""
return {
"duration_passed": result.duration <= baseline.max_duration,
"memory_passed": result.memory_usage <= baseline.max_memory,
"cpu_passed": result.cpu_usage <= baseline.max_cpu,
"duration_ratio": result.duration / baseline.max_duration,
"memory_ratio": result.memory_usage / baseline.max_memory,
"cpu_ratio": result.cpu_usage / baseline.max_cpu,
"overall_passed": (
result.duration <= baseline.max_duration and
result.memory_usage <= baseline.max_memory and
result.cpu_usage <= baseline.max_cpu
)
}