File size: 8,203 Bytes
aef804e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 | """
Performance Regression Test Fixtures
Provides fixtures for performance regression detection:
- performance_baseline: Loads historical baseline measurements from JSON
- check_regression: Validates current performance against baseline with threshold
- update_baseline: Helper to update baseline when performance improves
Usage:
def test_api_latency(benchmark, check_regression):
response = benchmark(make_api_call)
# Check against baseline (20% threshold)
check_regression(benchmark.stats.stats.mean, "api_get_agents_latency")
Reference: Phase 243 Plan 02 - Performance Regression Detection
"""
import json
from pathlib import Path
from typing import Dict, Any
import pytest
@pytest.fixture(scope="session")
def performance_baseline() -> Dict[str, float]:
"""
Load performance baseline from JSON file.
Returns dictionary of baseline measurements keyed by metric name.
Returns empty dict if baseline file doesn't exist (first run).
Baseline format:
{
"generated_at": "2026-03-25T00:00:00Z",
"baselines": {
"api_get_agents_latency": 0.050,
"api_health_check_latency": 0.005,
"db_agent_query": 0.010,
"cache_hit_rate": 0.95,
"cache_get_latency": 0.001
}
}
Returns:
Dict[str, float]: Baseline measurements
"""
baseline_file = Path(__file__).parent.parent / "performance_baseline.json"
if not baseline_file.exists():
# No baseline yet - return empty dict
return {}
try:
with open(baseline_file, "r") as f:
data = json.load(f)
return data.get("baselines", {})
except (json.JSONDecodeError, IOError) as e:
# Log warning but don't fail - tests will run without baseline
print(f"Warning: Could not load baseline file: {e}")
return {}
@pytest.fixture(scope="function")
def check_regression(performance_baseline: Dict[str, float]) -> callable:
"""
Create regression checker function.
Validates current performance value against historical baseline.
Fails test if performance degrades beyond threshold (20% by default).
Args:
performance_baseline: Loaded baseline measurements
Returns:
callable: Function that checks regression
Example:
def test_api_latency(benchmark, check_regression):
response = benchmark(make_api_call)
check_regression(benchmark.stats.stats.mean, "api_get_agents_latency")
"""
def _check(
current_value: float,
metric_name: str,
threshold: float = 0.2
) -> None:
"""
Check if current value regresses against baseline.
Args:
current_value: Current performance measurement (seconds for latency,
count for hit rate, etc.)
metric_name: Name of metric to check against baseline
threshold: Regression threshold as decimal (0.2 = 20%)
Raises:
AssertionError: If performance regresses beyond threshold
Note:
- Skips check if no baseline exists (first run)
- Fails if current_value > baseline * (1 + threshold)
- For hit rates (higher is better), inverts the check
"""
if metric_name not in performance_baseline:
# No baseline - skip check (first run or new metric)
return
baseline = performance_baseline[metric_name]
# Detect if this is a hit rate (higher is better)
# Hit rates are typically 0-1, latencies are >0
if baseline <= 1.0 and current_value <= 1.0:
# Hit rate - check for degradation (lower is worse)
regression_threshold = baseline * (1 - threshold)
if current_value < regression_threshold:
percent_change = ((current_value - baseline) / baseline) * 100
raise AssertionError(
f"Performance regression: {metric_name} = {current_value:.4f} "
f"(baseline: {baseline:.4f}, threshold: {regression_threshold:.4f}, "
f"change: {percent_change:.1f}%)"
)
else:
# Latency - check for increase (higher is worse)
regression_threshold = baseline * (1 + threshold)
if current_value > regression_threshold:
percent_change = ((current_value - baseline) / baseline) * 100
raise AssertionError(
f"Performance regression: {metric_name} = {current_value:.4f}s "
f"(baseline: {baseline:.4f}s, threshold: {regression_threshold:.4f}s, "
f"change: +{percent_change:.1f}%)"
)
return _check
@pytest.fixture(scope="session")
def baseline_file_path() -> Path:
"""
Get path to performance baseline file.
Returns:
Path: Path to performance_baseline.json
"""
return Path(__file__).parent.parent / "performance_baseline.json"
def update_baseline_on_improvement(
metric_name: str,
current_value: float,
baseline_value: float,
improvement_threshold: float = 0.1
) -> bool:
"""
Check if performance improved enough to update baseline.
Args:
metric_name: Name of metric
current_value: Current performance value
baseline_value: Existing baseline value
improvement_threshold: Improvement threshold (0.1 = 10%)
Returns:
bool: True if baseline should be updated
Note:
For hit rates (higher is better), checks if current > baseline * (1 + threshold)
For latencies (lower is better), checks if current < baseline * (1 - threshold)
"""
# Detect hit rate vs latency
if baseline_value <= 1.0 and current_value <= 1.0:
# Hit rate - improvement means higher value
improvement_threshold_value = baseline_value * (1 + improvement_threshold)
return current_value > improvement_threshold_value
else:
# Latency - improvement means lower value
improvement_threshold_value = baseline_value * (1 - improvement_threshold)
return current_value < improvement_threshold_value
class BaselineUpdater:
"""
Helper class to update baseline file with new measurements.
Used by CI/CD scripts to auto-update baselines when performance improves.
"""
def __init__(self, baseline_path: Path = None):
"""
Initialize baseline updater.
Args:
baseline_path: Path to baseline file (defaults to performance_baseline.json)
"""
if baseline_path is None:
baseline_path = Path(__file__).parent.parent / "performance_baseline.json"
self.baseline_path = baseline_path
def load_baselines(self) -> Dict[str, Any]:
"""Load existing baselines from file."""
if not self.baseline_path.exists():
return {
"generated_at": None,
"baselines": {}
}
with open(self.baseline_path, "r") as f:
return json.load(f)
def save_baselines(self, data: Dict[str, Any]) -> None:
"""
Save baselines to file.
Args:
data: Baseline data with 'generated_at' and 'baselines' keys
"""
from datetime import datetime
data["generated_at"] = datetime.utcnow().isoformat() + "Z"
with open(self.baseline_path, "w") as f:
json.dump(data, f, indent=2)
def update_metric(self, metric_name: str, value: float) -> None:
"""
Update single metric in baseline.
Args:
metric_name: Name of metric to update
value: New baseline value
"""
data = self.load_baselines()
data["baselines"][metric_name] = value
self.save_baselines(data)
def update_metrics(self, metrics: Dict[str, float]) -> None:
"""
Update multiple metrics in baseline.
Args:
metrics: Dictionary of metric names to values
"""
data = self.load_baselines()
data["baselines"].update(metrics)
self.save_baselines(data)
|