Performance Tests¶
Performance tests measure execution time, memory usage, and computational efficiency to ensure hazelbean maintains acceptable performance characteristics.
Overview¶
The performance test suite includes:
- Benchmarking - Standardized performance measurements
- Function Performance - Testing individual function execution times
- Workflow Performance - End-to-end processing performance
- Baseline Management - Tracking performance changes over time
Performance Benchmarking¶
Comprehensive benchmarks that measure and track performance metrics across different operations.
Core Performance Tests¶
Consolidated Performance Benchmark Tests
This file consolidates tests from: - benchmarks/test_get_path_performance.py - benchmarks/test_integration_scenarios_benchmark.py - benchmarks/test_simple_benchmarks.py
Covers comprehensive performance benchmarking including: - Single and multiple call performance benchmarks - Integration scenario performance validation - Simple working performance benchmarks - Benchmark baseline establishment and validation - Performance regression testing - Benchmark artifact generation and storage
BasePerformanceTest (TestCase)
¶
Base class for performance benchmark tests with shared setup
Source code in hazelbean_tests/performance/test_benchmarks.py
class BasePerformanceTest(unittest.TestCase):
"""Base class for performance benchmark tests with shared setup"""
def setUp(self):
"""Set up test fixtures and data paths"""
self.test_dir = tempfile.mkdtemp()
self.data_dir = os.path.join(os.path.dirname(__file__), "../../data")
self.test_data_dir = os.path.join(self.data_dir, "tests")
self.cartographic_data_dir = os.path.join(self.data_dir, "cartographic/ee")
self.pyramid_data_dir = os.path.join(self.data_dir, "pyramids")
self.crops_data_dir = os.path.join(self.data_dir, "crops/johnson")
# Test file paths for different formats
self.raster_test_file = "ee_r264_ids_900sec.tif"
self.vector_test_file = "ee_r264_simplified900sec.gpkg"
self.csv_test_file = "ee_r264_correspondence.csv"
self.pyramid_file = "ha_per_cell_900sec.tif"
self.crops_file = "crop_calories/maize_calories_per_ha_masked.tif"
# Create ProjectFlow instance
self.p = hb.ProjectFlow(self.test_dir)
# Create test directory structure
os.makedirs(os.path.join(self.test_dir, "intermediate"), exist_ok=True)
os.makedirs(os.path.join(self.test_dir, "input"), exist_ok=True)
# Create test files in project directories
self.create_test_files()
def tearDown(self):
"""Clean up test directories"""
shutil.rmtree(self.test_dir, ignore_errors=True)
def create_test_files(self):
"""Create test files in project directories for testing"""
# Create some test files in intermediate and input directories
with open(os.path.join(self.test_dir, "intermediate", "test_intermediate.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "input", "test_input.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "test_cur_dir.txt"), 'w') as f:
f.write("test content")
create_test_files(self)
¶
Create test files in project directories for testing
Source code in hazelbean_tests/performance/test_benchmarks.py
def create_test_files(self):
"""Create test files in project directories for testing"""
# Create some test files in intermediate and input directories
with open(os.path.join(self.test_dir, "intermediate", "test_intermediate.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "input", "test_input.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "test_cur_dir.txt"), 'w') as f:
f.write("test content")
TestGetPathPerformance (BasePerformanceTest)
¶
Test ProjectFlow.get_path() performance benchmarks (from test_get_path_performance.py)
Source code in hazelbean_tests/performance/test_benchmarks.py
class TestGetPathPerformance(BasePerformanceTest):
"""Test ProjectFlow.get_path() performance benchmarks (from test_get_path_performance.py)"""
@pytest.mark.benchmark
def test_single_call_performance_local_file(self):
"""Benchmark single get_path call for local file - Target: <0.1 seconds"""
# Test file in current directory
test_file = "test_cur_dir.txt"
# Benchmark single call
start_time = time.time()
resolved_path = self.p.get_path(test_file)
end_time = time.time()
call_duration = end_time - start_time
# Performance assertion
assert call_duration < 0.1, f"Single call took {call_duration:.4f}s, should be <0.1s"
# Verify functionality
assert test_file in resolved_path
assert os.path.exists(resolved_path)
@pytest.mark.benchmark
def test_single_call_performance_nested_file(self):
"""Benchmark single get_path call for nested file - Target: <0.1 seconds"""
# Test file in nested directory
test_file = "intermediate/test_intermediate.txt"
# Benchmark single call
start_time = time.time()
resolved_path = self.p.get_path(test_file)
end_time = time.time()
call_duration = end_time - start_time
# Performance assertion
assert call_duration < 0.1, f"Nested file call took {call_duration:.4f}s, should be <0.1s"
# Verify functionality
assert "test_intermediate.txt" in resolved_path
assert os.path.exists(resolved_path)
@pytest.mark.benchmark
def test_multiple_calls_performance(self):
"""Benchmark multiple sequential get_path calls - Target: <1.0 seconds for 100 calls"""
test_files = [
"test_cur_dir.txt",
"intermediate/test_intermediate.txt",
"input/test_input.txt",
"nonexistent_file.txt" # Include missing file
]
call_count = 100
# Benchmark multiple calls
start_time = time.time()
for i in range(call_count):
for test_file in test_files:
resolved_path = self.p.get_path(test_file)
end_time = time.time()
total_duration = end_time - start_time
avg_duration = total_duration / (call_count * len(test_files))
# Performance assertions
assert total_duration < 10.0, f"100x4 calls took {total_duration:.4f}s, should be <10s"
assert avg_duration < 0.025, f"Average call took {avg_duration:.4f}s, should be <0.025s"
@pytest.mark.benchmark
def test_missing_file_resolution_performance(self):
"""Benchmark get_path performance for missing files - Target: <0.2 seconds"""
missing_file = "definitely_does_not_exist.txt"
# Benchmark missing file resolution
start_time = time.time()
resolved_path = self.p.get_path(missing_file)
end_time = time.time()
call_duration = end_time - start_time
# Performance assertion - missing files may take longer due to search
assert call_duration < 0.2, f"Missing file call took {call_duration:.4f}s, should be <0.2s"
# Verify functionality - should still return a path
assert missing_file in resolved_path
test_single_call_performance_local_file(self)
¶
Benchmark single get_path call for local file - Target: <0.1 seconds
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
def test_single_call_performance_local_file(self):
"""Benchmark single get_path call for local file - Target: <0.1 seconds"""
# Test file in current directory
test_file = "test_cur_dir.txt"
# Benchmark single call
start_time = time.time()
resolved_path = self.p.get_path(test_file)
end_time = time.time()
call_duration = end_time - start_time
# Performance assertion
assert call_duration < 0.1, f"Single call took {call_duration:.4f}s, should be <0.1s"
# Verify functionality
assert test_file in resolved_path
assert os.path.exists(resolved_path)
test_single_call_performance_nested_file(self)
¶
Benchmark single get_path call for nested file - Target: <0.1 seconds
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
def test_single_call_performance_nested_file(self):
"""Benchmark single get_path call for nested file - Target: <0.1 seconds"""
# Test file in nested directory
test_file = "intermediate/test_intermediate.txt"
# Benchmark single call
start_time = time.time()
resolved_path = self.p.get_path(test_file)
end_time = time.time()
call_duration = end_time - start_time
# Performance assertion
assert call_duration < 0.1, f"Nested file call took {call_duration:.4f}s, should be <0.1s"
# Verify functionality
assert "test_intermediate.txt" in resolved_path
assert os.path.exists(resolved_path)
test_multiple_calls_performance(self)
¶
Benchmark multiple sequential get_path calls - Target: <1.0 seconds for 100 calls
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
def test_multiple_calls_performance(self):
"""Benchmark multiple sequential get_path calls - Target: <1.0 seconds for 100 calls"""
test_files = [
"test_cur_dir.txt",
"intermediate/test_intermediate.txt",
"input/test_input.txt",
"nonexistent_file.txt" # Include missing file
]
call_count = 100
# Benchmark multiple calls
start_time = time.time()
for i in range(call_count):
for test_file in test_files:
resolved_path = self.p.get_path(test_file)
end_time = time.time()
total_duration = end_time - start_time
avg_duration = total_duration / (call_count * len(test_files))
# Performance assertions
assert total_duration < 10.0, f"100x4 calls took {total_duration:.4f}s, should be <10s"
assert avg_duration < 0.025, f"Average call took {avg_duration:.4f}s, should be <0.025s"
test_missing_file_resolution_performance(self)
¶
Benchmark get_path performance for missing files - Target: <0.2 seconds
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
def test_missing_file_resolution_performance(self):
"""Benchmark get_path performance for missing files - Target: <0.2 seconds"""
missing_file = "definitely_does_not_exist.txt"
# Benchmark missing file resolution
start_time = time.time()
resolved_path = self.p.get_path(missing_file)
end_time = time.time()
call_duration = end_time - start_time
# Performance assertion - missing files may take longer due to search
assert call_duration < 0.2, f"Missing file call took {call_duration:.4f}s, should be <0.2s"
# Verify functionality - should still return a path
assert missing_file in resolved_path
TestSimpleBenchmarks (BasePerformanceTest)
¶
Simple working performance benchmarks for testing the system (from test_simple_benchmarks.py)
Source code in hazelbean_tests/performance/test_benchmarks.py
class TestSimpleBenchmarks(BasePerformanceTest):
"""Simple working performance benchmarks for testing the system (from test_simple_benchmarks.py)"""
@pytest.fixture
def test_setup(self):
"""Set up test fixtures"""
test_dir = tempfile.mkdtemp()
p = hb.ProjectFlow(test_dir)
# Create a simple test file
test_file_path = os.path.join(test_dir, "test_file.txt")
with open(test_file_path, 'w') as f:
f.write("test content")
yield test_dir, p
# Cleanup
shutil.rmtree(test_dir, ignore_errors=True)
@pytest.mark.benchmark
def test_array_operations_benchmark(self):
"""Simple array operations benchmark"""
def array_operations():
# Create test arrays
arr1 = np.random.rand(1000, 1000)
arr2 = np.random.rand(1000, 1000)
# Perform operations
result = arr1 + arr2
result = result * 2
result = np.mean(result)
return result
# Benchmark the operations
start_time = time.time()
result = array_operations()
end_time = time.time()
duration = end_time - start_time
# Should complete in reasonable time
assert duration < 10.0, f"Array operations took {duration:.4f}s, should be <10s"
assert isinstance(result, (int, float, np.number))
@pytest.mark.benchmark
def test_file_io_benchmark(self):
"""Simple file I/O operations benchmark"""
temp_dir = tempfile.mkdtemp()
try:
test_file = os.path.join(temp_dir, "benchmark_test.txt")
test_data = "test data " * 1000 # Create some test data
# Benchmark write operation
start_time = time.time()
with open(test_file, 'w') as f:
for i in range(100):
f.write(f"{test_data} {i}\n")
write_time = time.time() - start_time
# Benchmark read operation
start_time = time.time()
with open(test_file, 'r') as f:
content = f.read()
read_time = time.time() - start_time
# Performance assertions
assert write_time < 5.0, f"Write operations took {write_time:.4f}s, should be <5s"
assert read_time < 1.0, f"Read operation took {read_time:.4f}s, should be <1s"
assert len(content) > 0
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
@pytest.mark.benchmark
def test_project_flow_creation_benchmark(self):
"""Benchmark ProjectFlow creation performance"""
temp_dir = tempfile.mkdtemp()
try:
# Benchmark ProjectFlow creation
start_time = time.time()
for i in range(10):
p = hb.ProjectFlow(temp_dir)
# Basic operation to ensure it's working
path = p.get_path("test.txt")
end_time = time.time()
duration = end_time - start_time
avg_duration = duration / 10
# Performance assertions
assert duration < 5.0, f"10 ProjectFlow creations took {duration:.4f}s, should be <5s"
assert avg_duration < 0.5, f"Average creation took {avg_duration:.4f}s, should be <0.5s"
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
@pytest.mark.benchmark
def test_hazelbean_temp_benchmark(self):
"""Benchmark hazelbean temp file operations"""
# Benchmark temp file creation
start_time = time.time()
temp_paths = []
for i in range(50):
temp_path = hb.temp('.txt', f'benchmark_{i}', True)
temp_paths.append(temp_path)
end_time = time.time()
duration = end_time - start_time
avg_duration = duration / 50
# Performance assertions
assert duration < 5.0, f"50 temp file creations took {duration:.4f}s, should be <5s"
assert avg_duration < 0.1, f"Average temp creation took {avg_duration:.4f}s, should be <0.1s"
# Verify files exist (they should be temporary)
assert len(temp_paths) == 50
@pytest.mark.benchmark
def test_numpy_save_load_benchmark(self):
"""Benchmark numpy array save/load operations with hazelbean"""
# Create test array
test_array = np.random.rand(500, 500)
temp_path = hb.temp('.npy', 'numpy_benchmark', True)
# Benchmark save operation
start_time = time.time()
hb.save_array_as_npy(test_array, temp_path)
save_time = time.time() - start_time
# Benchmark load operation
start_time = time.time()
loaded_array = np.load(temp_path)
load_time = time.time() - start_time
# Performance assertions
assert save_time < 2.0, f"Array save took {save_time:.4f}s, should be <2s"
assert load_time < 1.0, f"Array load took {load_time:.4f}s, should be <1s"
# Verify functionality
assert np.array_equal(test_array, loaded_array)
test_setup(self)
¶
Set up test fixtures
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.fixture
def test_setup(self):
"""Set up test fixtures"""
test_dir = tempfile.mkdtemp()
p = hb.ProjectFlow(test_dir)
# Create a simple test file
test_file_path = os.path.join(test_dir, "test_file.txt")
with open(test_file_path, 'w') as f:
f.write("test content")
yield test_dir, p
# Cleanup
shutil.rmtree(test_dir, ignore_errors=True)
test_array_operations_benchmark(self)
¶
Simple array operations benchmark
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
def test_array_operations_benchmark(self):
"""Simple array operations benchmark"""
def array_operations():
# Create test arrays
arr1 = np.random.rand(1000, 1000)
arr2 = np.random.rand(1000, 1000)
# Perform operations
result = arr1 + arr2
result = result * 2
result = np.mean(result)
return result
# Benchmark the operations
start_time = time.time()
result = array_operations()
end_time = time.time()
duration = end_time - start_time
# Should complete in reasonable time
assert duration < 10.0, f"Array operations took {duration:.4f}s, should be <10s"
assert isinstance(result, (int, float, np.number))
test_file_io_benchmark(self)
¶
Simple file I/O operations benchmark
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
def test_file_io_benchmark(self):
"""Simple file I/O operations benchmark"""
temp_dir = tempfile.mkdtemp()
try:
test_file = os.path.join(temp_dir, "benchmark_test.txt")
test_data = "test data " * 1000 # Create some test data
# Benchmark write operation
start_time = time.time()
with open(test_file, 'w') as f:
for i in range(100):
f.write(f"{test_data} {i}\n")
write_time = time.time() - start_time
# Benchmark read operation
start_time = time.time()
with open(test_file, 'r') as f:
content = f.read()
read_time = time.time() - start_time
# Performance assertions
assert write_time < 5.0, f"Write operations took {write_time:.4f}s, should be <5s"
assert read_time < 1.0, f"Read operation took {read_time:.4f}s, should be <1s"
assert len(content) > 0
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
test_project_flow_creation_benchmark(self)
¶
Benchmark ProjectFlow creation performance
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
def test_project_flow_creation_benchmark(self):
"""Benchmark ProjectFlow creation performance"""
temp_dir = tempfile.mkdtemp()
try:
# Benchmark ProjectFlow creation
start_time = time.time()
for i in range(10):
p = hb.ProjectFlow(temp_dir)
# Basic operation to ensure it's working
path = p.get_path("test.txt")
end_time = time.time()
duration = end_time - start_time
avg_duration = duration / 10
# Performance assertions
assert duration < 5.0, f"10 ProjectFlow creations took {duration:.4f}s, should be <5s"
assert avg_duration < 0.5, f"Average creation took {avg_duration:.4f}s, should be <0.5s"
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
test_hazelbean_temp_benchmark(self)
¶
Benchmark hazelbean temp file operations
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
def test_hazelbean_temp_benchmark(self):
"""Benchmark hazelbean temp file operations"""
# Benchmark temp file creation
start_time = time.time()
temp_paths = []
for i in range(50):
temp_path = hb.temp('.txt', f'benchmark_{i}', True)
temp_paths.append(temp_path)
end_time = time.time()
duration = end_time - start_time
avg_duration = duration / 50
# Performance assertions
assert duration < 5.0, f"50 temp file creations took {duration:.4f}s, should be <5s"
assert avg_duration < 0.1, f"Average temp creation took {avg_duration:.4f}s, should be <0.1s"
# Verify files exist (they should be temporary)
assert len(temp_paths) == 50
test_numpy_save_load_benchmark(self)
¶
Benchmark numpy array save/load operations with hazelbean
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
def test_numpy_save_load_benchmark(self):
"""Benchmark numpy array save/load operations with hazelbean"""
# Create test array
test_array = np.random.rand(500, 500)
temp_path = hb.temp('.npy', 'numpy_benchmark', True)
# Benchmark save operation
start_time = time.time()
hb.save_array_as_npy(test_array, temp_path)
save_time = time.time() - start_time
# Benchmark load operation
start_time = time.time()
loaded_array = np.load(temp_path)
load_time = time.time() - start_time
# Performance assertions
assert save_time < 2.0, f"Array save took {save_time:.4f}s, should be <2s"
assert load_time < 1.0, f"Array load took {load_time:.4f}s, should be <1s"
# Verify functionality
assert np.array_equal(test_array, loaded_array)
TestIntegrationScenarioBenchmarks (BasePerformanceTest)
¶
Integration scenario performance benchmarks (from test_integration_scenarios_benchmark.py)
Source code in hazelbean_tests/performance/test_benchmarks.py
class TestIntegrationScenarioBenchmarks(BasePerformanceTest):
"""Integration scenario performance benchmarks (from test_integration_scenarios_benchmark.py)"""
@pytest.mark.benchmark
@pytest.mark.slow
def test_data_processing_workflow_benchmark(self):
"""Benchmark complete data processing workflow"""
# This would test a complete hazelbean workflow
# Including raster processing, array operations, etc.
start_time = time.time()
# Simulate data processing workflow
p = hb.ProjectFlow(self.test_dir)
# Create test array
test_array = np.random.rand(100, 100)
temp_path = hb.temp('.npy', 'workflow_test', True)
# Save and process array
hb.save_array_as_npy(test_array, temp_path)
result = hb.describe(temp_path, surpress_print=True, surpress_logger=True)
end_time = time.time()
duration = end_time - start_time
# Performance assertion
assert duration < 10.0, f"Data processing workflow took {duration:.4f}s, should be <10s"
@pytest.mark.benchmark
@pytest.mark.slow
def test_multi_file_processing_benchmark(self):
"""Benchmark processing multiple files"""
# Create multiple test files
test_files = []
for i in range(10):
test_array = np.random.rand(50, 50)
temp_path = hb.temp('.npy', f'multi_test_{i}', True)
hb.save_array_as_npy(test_array, temp_path)
test_files.append(temp_path)
# Benchmark processing all files
start_time = time.time()
results = []
for file_path in test_files:
result = hb.describe(file_path, surpress_print=True, surpress_logger=True)
results.append(result)
end_time = time.time()
duration = end_time - start_time
avg_duration = duration / len(test_files)
# Performance assertions
assert duration < 20.0, f"Multi-file processing took {duration:.4f}s, should be <20s"
assert avg_duration < 2.0, f"Average file processing took {avg_duration:.4f}s, should be <2s"
assert len(results) == len(test_files)
@pytest.mark.benchmark
def test_path_resolution_stress_test(self):
"""Stress test path resolution performance with many files"""
# Create many test files
file_count = 100
test_files = []
for i in range(file_count):
subdir = f"subdir_{i % 10}" # Create 10 subdirectories
os.makedirs(os.path.join(self.test_dir, subdir), exist_ok=True)
file_path = os.path.join(subdir, f"test_file_{i}.txt")
full_path = os.path.join(self.test_dir, file_path)
with open(full_path, 'w') as f:
f.write(f"test content {i}")
test_files.append(file_path)
# Benchmark path resolution for all files
start_time = time.time()
resolved_paths = []
for file_path in test_files:
resolved = self.p.get_path(file_path)
resolved_paths.append(resolved)
end_time = time.time()
duration = end_time - start_time
avg_duration = duration / file_count
# Performance assertions
assert duration < 30.0, f"Stress test took {duration:.4f}s, should be <30s"
assert avg_duration < 0.3, f"Average resolution took {avg_duration:.4f}s, should be <0.3s"
assert len(resolved_paths) == file_count
# Verify all paths were resolved
for resolved in resolved_paths:
assert os.path.exists(resolved)
test_data_processing_workflow_benchmark(self)
¶
Benchmark complete data processing workflow
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
@pytest.mark.slow
def test_data_processing_workflow_benchmark(self):
"""Benchmark complete data processing workflow"""
# This would test a complete hazelbean workflow
# Including raster processing, array operations, etc.
start_time = time.time()
# Simulate data processing workflow
p = hb.ProjectFlow(self.test_dir)
# Create test array
test_array = np.random.rand(100, 100)
temp_path = hb.temp('.npy', 'workflow_test', True)
# Save and process array
hb.save_array_as_npy(test_array, temp_path)
result = hb.describe(temp_path, surpress_print=True, surpress_logger=True)
end_time = time.time()
duration = end_time - start_time
# Performance assertion
assert duration < 10.0, f"Data processing workflow took {duration:.4f}s, should be <10s"
test_multi_file_processing_benchmark(self)
¶
Benchmark processing multiple files
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
@pytest.mark.slow
def test_multi_file_processing_benchmark(self):
"""Benchmark processing multiple files"""
# Create multiple test files
test_files = []
for i in range(10):
test_array = np.random.rand(50, 50)
temp_path = hb.temp('.npy', f'multi_test_{i}', True)
hb.save_array_as_npy(test_array, temp_path)
test_files.append(temp_path)
# Benchmark processing all files
start_time = time.time()
results = []
for file_path in test_files:
result = hb.describe(file_path, surpress_print=True, surpress_logger=True)
results.append(result)
end_time = time.time()
duration = end_time - start_time
avg_duration = duration / len(test_files)
# Performance assertions
assert duration < 20.0, f"Multi-file processing took {duration:.4f}s, should be <20s"
assert avg_duration < 2.0, f"Average file processing took {avg_duration:.4f}s, should be <2s"
assert len(results) == len(test_files)
test_path_resolution_stress_test(self)
¶
Stress test path resolution performance with many files
Source code in hazelbean_tests/performance/test_benchmarks.py
@pytest.mark.benchmark
def test_path_resolution_stress_test(self):
"""Stress test path resolution performance with many files"""
# Create many test files
file_count = 100
test_files = []
for i in range(file_count):
subdir = f"subdir_{i % 10}" # Create 10 subdirectories
os.makedirs(os.path.join(self.test_dir, subdir), exist_ok=True)
file_path = os.path.join(subdir, f"test_file_{i}.txt")
full_path = os.path.join(self.test_dir, file_path)
with open(full_path, 'w') as f:
f.write(f"test content {i}")
test_files.append(file_path)
# Benchmark path resolution for all files
start_time = time.time()
resolved_paths = []
for file_path in test_files:
resolved = self.p.get_path(file_path)
resolved_paths.append(resolved)
end_time = time.time()
duration = end_time - start_time
avg_duration = duration / file_count
# Performance assertions
assert duration < 30.0, f"Stress test took {duration:.4f}s, should be <30s"
assert avg_duration < 0.3, f"Average resolution took {avg_duration:.4f}s, should be <0.3s"
assert len(resolved_paths) == file_count
# Verify all paths were resolved
for resolved in resolved_paths:
assert os.path.exists(resolved)
Function Performance Testing¶
Tests focused on measuring the performance of individual functions and methods.
Individual Function Benchmarks¶
Consolidated Performance Function Tests
This file consolidates tests from: - functions/test_get_path_benchmarks.py - functions/test_path_resolution_benchmarks.py - functions/test_tiling_benchmarks.py
Covers function-level performance testing including: - Individual function performance benchmarks - Path resolution algorithm performance - Tiling operation performance benchmarks - Function-specific regression testing - Memory usage and efficiency testing
BaseFunctionPerformanceTest (TestCase)
¶
Base class for function-level performance tests with shared setup
Source code in hazelbean_tests/performance/test_functions.py
class BaseFunctionPerformanceTest(unittest.TestCase):
"""Base class for function-level performance tests with shared setup"""
def setUp(self):
"""Set up test fixtures and data paths"""
self.test_dir = tempfile.mkdtemp()
self.data_dir = os.path.join(os.path.dirname(__file__), "../../data")
# Create ProjectFlow instance
self.p = hb.ProjectFlow(self.test_dir)
# Create test directory structure
os.makedirs(os.path.join(self.test_dir, "intermediate"), exist_ok=True)
os.makedirs(os.path.join(self.test_dir, "input"), exist_ok=True)
# Create test files in project directories
self.create_test_files()
def tearDown(self):
"""Clean up test directories"""
shutil.rmtree(self.test_dir, ignore_errors=True)
def create_test_files(self):
"""Create test files in project directories for testing"""
# Create some test files in intermediate and input directories
with open(os.path.join(self.test_dir, "intermediate", "test_intermediate.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "input", "test_input.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "test_cur_dir.txt"), 'w') as f:
f.write("test content")
create_test_files(self)
¶
Create test files in project directories for testing
Source code in hazelbean_tests/performance/test_functions.py
def create_test_files(self):
"""Create test files in project directories for testing"""
# Create some test files in intermediate and input directories
with open(os.path.join(self.test_dir, "intermediate", "test_intermediate.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "input", "test_input.txt"), 'w') as f:
f.write("test content")
with open(os.path.join(self.test_dir, "test_cur_dir.txt"), 'w') as f:
f.write("test content")
TestGetPathFunctionBenchmarks (BaseFunctionPerformanceTest)
¶
Test get_path function-specific benchmarks (from test_get_path_benchmarks.py)
Source code in hazelbean_tests/performance/test_functions.py
class TestGetPathFunctionBenchmarks(BaseFunctionPerformanceTest):
"""Test get_path function-specific benchmarks (from test_get_path_benchmarks.py)"""
@pytest.mark.benchmark
def test_get_path_function_overhead(self):
"""Benchmark just the get_path function call overhead"""
test_file = "test_cur_dir.txt"
# Warm up
for _ in range(5):
self.p.get_path(test_file)
# Benchmark pure function call
iterations = 1000
start_time = time.time()
for _ in range(iterations):
result = self.p.get_path(test_file)
end_time = time.time()
total_duration = end_time - start_time
avg_duration = total_duration / iterations
# Performance assertions
assert total_duration < 5.0, f"1000 function calls took {total_duration:.4f}s, should be <5s"
assert avg_duration < 0.005, f"Average function call took {avg_duration:.6f}s, should be <0.005s"
@pytest.mark.benchmark
def test_get_path_cache_performance(self):
"""Benchmark get_path caching efficiency"""
test_file = "test_cur_dir.txt"
# First call (no cache)
start_time = time.time()
result1 = self.p.get_path(test_file)
first_call_time = time.time() - start_time
# Subsequent calls (should use cache if implemented)
cached_times = []
for _ in range(100):
start_time = time.time()
result2 = self.p.get_path(test_file)
cached_times.append(time.time() - start_time)
avg_cached_time = sum(cached_times) / len(cached_times)
# Verify results are consistent
assert result1 == result2, "Cached results should be identical"
# Performance assertion (cached calls should be faster, if caching is implemented)
# If no caching, this test documents current performance
assert avg_cached_time < 0.01, f"Average cached call took {avg_cached_time:.6f}s, should be <0.01s"
@pytest.mark.benchmark
def test_get_path_different_patterns(self):
"""Benchmark get_path with different file name patterns"""
patterns = [
"simple.txt", # Simple filename
"intermediate/nested.txt", # Nested path
"deep/nested/path/file.txt", # Deep nesting
"file_with_long_name_and_numbers_12345.extension", # Long filename
"file-with-dashes.txt", # Special characters
"file_with_spaces.txt", # Spaces (if supported)
"UPPERCASE.TXT", # Uppercase
"mixed_Case_File.TxT", # Mixed case
]
# Create test files for existing patterns
for pattern in patterns:
if "/" in pattern:
dir_path = os.path.dirname(os.path.join(self.test_dir, pattern))
os.makedirs(dir_path, exist_ok=True)
full_path = os.path.join(self.test_dir, pattern)
if not os.path.exists(full_path):
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, 'w') as f:
f.write("test content")
# Benchmark each pattern
pattern_times = {}
for pattern in patterns:
start_time = time.time()
for _ in range(100): # Multiple calls per pattern
result = self.p.get_path(pattern)
end_time = time.time()
avg_time = (end_time - start_time) / 100
pattern_times[pattern] = avg_time
# Individual performance assertion
assert avg_time < 0.05, f"Pattern '{pattern}' took {avg_time:.6f}s avg, should be <0.05s"
# Verify performance consistency across patterns
max_time = max(pattern_times.values())
min_time = min(pattern_times.values())
time_variance = max_time - min_time
# Performance shouldn't vary dramatically by pattern
assert time_variance < 0.1, f"Performance variance {time_variance:.6f}s too high across patterns"
test_get_path_function_overhead(self)
¶
Benchmark just the get_path function call overhead
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
def test_get_path_function_overhead(self):
"""Benchmark just the get_path function call overhead"""
test_file = "test_cur_dir.txt"
# Warm up
for _ in range(5):
self.p.get_path(test_file)
# Benchmark pure function call
iterations = 1000
start_time = time.time()
for _ in range(iterations):
result = self.p.get_path(test_file)
end_time = time.time()
total_duration = end_time - start_time
avg_duration = total_duration / iterations
# Performance assertions
assert total_duration < 5.0, f"1000 function calls took {total_duration:.4f}s, should be <5s"
assert avg_duration < 0.005, f"Average function call took {avg_duration:.6f}s, should be <0.005s"
test_get_path_cache_performance(self)
¶
Benchmark get_path caching efficiency
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
def test_get_path_cache_performance(self):
"""Benchmark get_path caching efficiency"""
test_file = "test_cur_dir.txt"
# First call (no cache)
start_time = time.time()
result1 = self.p.get_path(test_file)
first_call_time = time.time() - start_time
# Subsequent calls (should use cache if implemented)
cached_times = []
for _ in range(100):
start_time = time.time()
result2 = self.p.get_path(test_file)
cached_times.append(time.time() - start_time)
avg_cached_time = sum(cached_times) / len(cached_times)
# Verify results are consistent
assert result1 == result2, "Cached results should be identical"
# Performance assertion (cached calls should be faster, if caching is implemented)
# If no caching, this test documents current performance
assert avg_cached_time < 0.01, f"Average cached call took {avg_cached_time:.6f}s, should be <0.01s"
test_get_path_different_patterns(self)
¶
Benchmark get_path with different file name patterns
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
def test_get_path_different_patterns(self):
"""Benchmark get_path with different file name patterns"""
patterns = [
"simple.txt", # Simple filename
"intermediate/nested.txt", # Nested path
"deep/nested/path/file.txt", # Deep nesting
"file_with_long_name_and_numbers_12345.extension", # Long filename
"file-with-dashes.txt", # Special characters
"file_with_spaces.txt", # Spaces (if supported)
"UPPERCASE.TXT", # Uppercase
"mixed_Case_File.TxT", # Mixed case
]
# Create test files for existing patterns
for pattern in patterns:
if "/" in pattern:
dir_path = os.path.dirname(os.path.join(self.test_dir, pattern))
os.makedirs(dir_path, exist_ok=True)
full_path = os.path.join(self.test_dir, pattern)
if not os.path.exists(full_path):
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, 'w') as f:
f.write("test content")
# Benchmark each pattern
pattern_times = {}
for pattern in patterns:
start_time = time.time()
for _ in range(100): # Multiple calls per pattern
result = self.p.get_path(pattern)
end_time = time.time()
avg_time = (end_time - start_time) / 100
pattern_times[pattern] = avg_time
# Individual performance assertion
assert avg_time < 0.05, f"Pattern '{pattern}' took {avg_time:.6f}s avg, should be <0.05s"
# Verify performance consistency across patterns
max_time = max(pattern_times.values())
min_time = min(pattern_times.values())
time_variance = max_time - min_time
# Performance shouldn't vary dramatically by pattern
assert time_variance < 0.1, f"Performance variance {time_variance:.6f}s too high across patterns"
TestPathResolutionBenchmarks (BaseFunctionPerformanceTest)
¶
Test path resolution algorithm benchmarks (from test_path_resolution_benchmarks.py)
Source code in hazelbean_tests/performance/test_functions.py
class TestPathResolutionBenchmarks(BaseFunctionPerformanceTest):
"""Test path resolution algorithm benchmarks (from test_path_resolution_benchmarks.py)"""
@pytest.mark.benchmark
def test_absolute_path_resolution(self):
"""Benchmark absolute path resolution performance"""
# Create absolute path
abs_path = os.path.abspath(os.path.join(self.test_dir, "test_cur_dir.txt"))
# Benchmark absolute path resolution
start_time = time.time()
for _ in range(100):
result = self.p.get_path(abs_path)
end_time = time.time()
avg_time = (end_time - start_time) / 100
# Performance assertion
assert avg_time < 0.01, f"Absolute path resolution took {avg_time:.6f}s avg, should be <0.01s"
# Verify result
assert result == abs_path or abs_path in result
@pytest.mark.benchmark
def test_relative_path_resolution(self):
"""Benchmark relative path resolution performance"""
rel_paths = [
"test_cur_dir.txt",
"intermediate/test_intermediate.txt",
"../hazelbean_tests/performance/test_functions.py", # Up and back down
"./test_cur_dir.txt", # Explicit current dir
]
for rel_path in rel_paths:
start_time = time.time()
for _ in range(50):
try:
result = self.p.get_path(rel_path)
except:
# Some paths may not exist, that's OK for performance testing
pass
end_time = time.time()
avg_time = (end_time - start_time) / 50
# Performance assertion
assert avg_time < 0.02, f"Relative path '{rel_path}' took {avg_time:.6f}s avg, should be <0.02s"
@pytest.mark.benchmark
def test_nonexistent_path_resolution(self):
"""Benchmark performance when resolving non-existent paths"""
nonexistent_paths = [
"does_not_exist.txt",
"missing/directory/file.txt",
"very/deep/nested/missing/path/file.extension",
]
for path in nonexistent_paths:
start_time = time.time()
for _ in range(50):
result = self.p.get_path(path) # Should still return a path
end_time = time.time()
avg_time = (end_time - start_time) / 50
# Nonexistent paths may take longer, but should still be reasonable
assert avg_time < 0.1, f"Nonexistent path '{path}' took {avg_time:.6f}s avg, should be <0.1s"
@pytest.mark.benchmark
def test_path_normalization_performance(self):
"""Benchmark path normalization and cleanup performance"""
messy_paths = [
"test_cur_dir.txt",
"./test_cur_dir.txt",
"intermediate/../test_cur_dir.txt",
"intermediate/./test_intermediate.txt",
"intermediate//test_intermediate.txt", # Double slash
"intermediate/subdir/../test_intermediate.txt",
]
# Create the actual files where possible
os.makedirs(os.path.join(self.test_dir, "intermediate", "subdir"), exist_ok=True)
for messy_path in messy_paths:
start_time = time.time()
for _ in range(100):
result = self.p.get_path(messy_path)
end_time = time.time()
avg_time = (end_time - start_time) / 100
# Path normalization should be fast
assert avg_time < 0.01, f"Path normalization '{messy_path}' took {avg_time:.6f}s avg, should be <0.01s"
test_absolute_path_resolution(self)
¶
Benchmark absolute path resolution performance
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
def test_absolute_path_resolution(self):
"""Benchmark absolute path resolution performance"""
# Create absolute path
abs_path = os.path.abspath(os.path.join(self.test_dir, "test_cur_dir.txt"))
# Benchmark absolute path resolution
start_time = time.time()
for _ in range(100):
result = self.p.get_path(abs_path)
end_time = time.time()
avg_time = (end_time - start_time) / 100
# Performance assertion
assert avg_time < 0.01, f"Absolute path resolution took {avg_time:.6f}s avg, should be <0.01s"
# Verify result
assert result == abs_path or abs_path in result
test_relative_path_resolution(self)
¶
Benchmark relative path resolution performance
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
def test_relative_path_resolution(self):
"""Benchmark relative path resolution performance"""
rel_paths = [
"test_cur_dir.txt",
"intermediate/test_intermediate.txt",
"../hazelbean_tests/performance/test_functions.py", # Up and back down
"./test_cur_dir.txt", # Explicit current dir
]
for rel_path in rel_paths:
start_time = time.time()
for _ in range(50):
try:
result = self.p.get_path(rel_path)
except:
# Some paths may not exist, that's OK for performance testing
pass
end_time = time.time()
avg_time = (end_time - start_time) / 50
# Performance assertion
assert avg_time < 0.02, f"Relative path '{rel_path}' took {avg_time:.6f}s avg, should be <0.02s"
test_nonexistent_path_resolution(self)
¶
Benchmark performance when resolving non-existent paths
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
def test_nonexistent_path_resolution(self):
"""Benchmark performance when resolving non-existent paths"""
nonexistent_paths = [
"does_not_exist.txt",
"missing/directory/file.txt",
"very/deep/nested/missing/path/file.extension",
]
for path in nonexistent_paths:
start_time = time.time()
for _ in range(50):
result = self.p.get_path(path) # Should still return a path
end_time = time.time()
avg_time = (end_time - start_time) / 50
# Nonexistent paths may take longer, but should still be reasonable
assert avg_time < 0.1, f"Nonexistent path '{path}' took {avg_time:.6f}s avg, should be <0.1s"
test_path_normalization_performance(self)
¶
Benchmark path normalization and cleanup performance
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
def test_path_normalization_performance(self):
"""Benchmark path normalization and cleanup performance"""
messy_paths = [
"test_cur_dir.txt",
"./test_cur_dir.txt",
"intermediate/../test_cur_dir.txt",
"intermediate/./test_intermediate.txt",
"intermediate//test_intermediate.txt", # Double slash
"intermediate/subdir/../test_intermediate.txt",
]
# Create the actual files where possible
os.makedirs(os.path.join(self.test_dir, "intermediate", "subdir"), exist_ok=True)
for messy_path in messy_paths:
start_time = time.time()
for _ in range(100):
result = self.p.get_path(messy_path)
end_time = time.time()
avg_time = (end_time - start_time) / 100
# Path normalization should be fast
assert avg_time < 0.01, f"Path normalization '{messy_path}' took {avg_time:.6f}s avg, should be <0.01s"
TestTilingBenchmarks (BaseFunctionPerformanceTest)
¶
Test tiling operation benchmarks (from test_tiling_benchmarks.py)
Source code in hazelbean_tests/performance/test_functions.py
class TestTilingBenchmarks(BaseFunctionPerformanceTest):
"""Test tiling operation benchmarks (from test_tiling_benchmarks.py)"""
@pytest.mark.benchmark
@pytest.mark.slow
def test_array_tiling_performance(self):
"""Benchmark array tiling operations"""
# Create test array
test_array = np.random.rand(1000, 1000)
temp_path = hb.temp('.npy', 'tiling_test', True)
hb.save_array_as_npy(test_array, temp_path)
# Benchmark array tiling (if implemented)
start_time = time.time()
# Simulate tiling operation - breaking array into chunks
tile_size = 100
tiles = []
for i in range(0, test_array.shape[0], tile_size):
for j in range(0, test_array.shape[1], tile_size):
tile = test_array[i:i+tile_size, j:j+tile_size]
tiles.append(tile)
end_time = time.time()
duration = end_time - start_time
tiles_per_second = len(tiles) / duration
# Performance assertions
assert duration < 5.0, f"Array tiling took {duration:.4f}s, should be <5s"
assert tiles_per_second > 10, f"Tiling rate {tiles_per_second:.2f} tiles/s, should be >10/s"
assert len(tiles) == 100 # 10x10 grid of tiles
@pytest.mark.benchmark
def test_small_array_tiling_performance(self):
"""Benchmark tiling performance for small arrays"""
# Test with smaller arrays to measure overhead
small_arrays = [
(100, 100),
(50, 50),
(25, 25),
(10, 10)
]
for width, height in small_arrays:
test_array = np.random.rand(width, height)
start_time = time.time()
# Tile into 5x5 chunks
tile_size = max(5, min(width, height) // 2)
tiles = []
for i in range(0, width, tile_size):
for j in range(0, height, tile_size):
tile = test_array[i:i+tile_size, j:j+tile_size]
tiles.append(tile)
end_time = time.time()
duration = end_time - start_time
# Small arrays should tile very quickly
assert duration < 0.1, f"Small array ({width}x{height}) tiling took {duration:.6f}s, should be <0.1s"
assert len(tiles) > 0, "Should generate at least one tile"
@pytest.mark.benchmark
def test_tile_reassembly_performance(self):
"""Benchmark tile reassembly performance"""
# Create original array
original_array = np.random.rand(200, 200)
# Break into tiles
tile_size = 50
tiles = []
positions = []
for i in range(0, original_array.shape[0], tile_size):
for j in range(0, original_array.shape[1], tile_size):
tile = original_array[i:i+tile_size, j:j+tile_size]
tiles.append(tile)
positions.append((i, j))
# Benchmark reassembly
start_time = time.time()
# Reassemble tiles
reassembled = np.zeros_like(original_array)
for tile, (i, j) in zip(tiles, positions):
end_i = min(i + tile_size, original_array.shape[0])
end_j = min(j + tile_size, original_array.shape[1])
reassembled[i:end_i, j:end_j] = tile
end_time = time.time()
duration = end_time - start_time
# Performance assertion
assert duration < 1.0, f"Tile reassembly took {duration:.4f}s, should be <1s"
# Verify correctness
assert np.array_equal(original_array, reassembled), "Reassembled array should match original"
@pytest.mark.benchmark
def test_memory_efficient_tiling(self):
"""Benchmark memory-efficient tiling operations"""
# Test tiling without loading entire array into memory at once
# Create a larger "virtual" array through file operations
temp_dir = tempfile.mkdtemp()
try:
# Create multiple small array files to simulate large dataset
file_count = 20
array_files = []
for i in range(file_count):
small_array = np.random.rand(50, 50)
file_path = os.path.join(temp_dir, f"array_{i:02d}.npy")
np.save(file_path, small_array)
array_files.append(file_path)
# Benchmark processing files individually (memory efficient)
start_time = time.time()
processed_count = 0
for file_path in array_files:
# Load, process, and immediately release
array = np.load(file_path)
# Simulate processing (tiling)
tiles = [array[i:i+10, j:j+10] for i in range(0, 50, 10) for j in range(0, 50, 10)]
processed_count += len(tiles)
del array, tiles # Explicit cleanup
end_time = time.time()
duration = end_time - start_time
files_per_second = file_count / duration
# Performance assertions
assert duration < 10.0, f"Memory-efficient processing took {duration:.4f}s, should be <10s"
assert files_per_second > 1, f"Processing rate {files_per_second:.2f} files/s, should be >1/s"
assert processed_count > 0, "Should have processed some tiles"
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
test_array_tiling_performance(self)
¶
Benchmark array tiling operations
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
@pytest.mark.slow
def test_array_tiling_performance(self):
"""Benchmark array tiling operations"""
# Create test array
test_array = np.random.rand(1000, 1000)
temp_path = hb.temp('.npy', 'tiling_test', True)
hb.save_array_as_npy(test_array, temp_path)
# Benchmark array tiling (if implemented)
start_time = time.time()
# Simulate tiling operation - breaking array into chunks
tile_size = 100
tiles = []
for i in range(0, test_array.shape[0], tile_size):
for j in range(0, test_array.shape[1], tile_size):
tile = test_array[i:i+tile_size, j:j+tile_size]
tiles.append(tile)
end_time = time.time()
duration = end_time - start_time
tiles_per_second = len(tiles) / duration
# Performance assertions
assert duration < 5.0, f"Array tiling took {duration:.4f}s, should be <5s"
assert tiles_per_second > 10, f"Tiling rate {tiles_per_second:.2f} tiles/s, should be >10/s"
assert len(tiles) == 100 # 10x10 grid of tiles
test_small_array_tiling_performance(self)
¶
Benchmark tiling performance for small arrays
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
def test_small_array_tiling_performance(self):
"""Benchmark tiling performance for small arrays"""
# Test with smaller arrays to measure overhead
small_arrays = [
(100, 100),
(50, 50),
(25, 25),
(10, 10)
]
for width, height in small_arrays:
test_array = np.random.rand(width, height)
start_time = time.time()
# Tile into 5x5 chunks
tile_size = max(5, min(width, height) // 2)
tiles = []
for i in range(0, width, tile_size):
for j in range(0, height, tile_size):
tile = test_array[i:i+tile_size, j:j+tile_size]
tiles.append(tile)
end_time = time.time()
duration = end_time - start_time
# Small arrays should tile very quickly
assert duration < 0.1, f"Small array ({width}x{height}) tiling took {duration:.6f}s, should be <0.1s"
assert len(tiles) > 0, "Should generate at least one tile"
test_tile_reassembly_performance(self)
¶
Benchmark tile reassembly performance
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
def test_tile_reassembly_performance(self):
"""Benchmark tile reassembly performance"""
# Create original array
original_array = np.random.rand(200, 200)
# Break into tiles
tile_size = 50
tiles = []
positions = []
for i in range(0, original_array.shape[0], tile_size):
for j in range(0, original_array.shape[1], tile_size):
tile = original_array[i:i+tile_size, j:j+tile_size]
tiles.append(tile)
positions.append((i, j))
# Benchmark reassembly
start_time = time.time()
# Reassemble tiles
reassembled = np.zeros_like(original_array)
for tile, (i, j) in zip(tiles, positions):
end_i = min(i + tile_size, original_array.shape[0])
end_j = min(j + tile_size, original_array.shape[1])
reassembled[i:end_i, j:end_j] = tile
end_time = time.time()
duration = end_time - start_time
# Performance assertion
assert duration < 1.0, f"Tile reassembly took {duration:.4f}s, should be <1s"
# Verify correctness
assert np.array_equal(original_array, reassembled), "Reassembled array should match original"
test_memory_efficient_tiling(self)
¶
Benchmark memory-efficient tiling operations
Source code in hazelbean_tests/performance/test_functions.py
@pytest.mark.benchmark
def test_memory_efficient_tiling(self):
"""Benchmark memory-efficient tiling operations"""
# Test tiling without loading entire array into memory at once
# Create a larger "virtual" array through file operations
temp_dir = tempfile.mkdtemp()
try:
# Create multiple small array files to simulate large dataset
file_count = 20
array_files = []
for i in range(file_count):
small_array = np.random.rand(50, 50)
file_path = os.path.join(temp_dir, f"array_{i:02d}.npy")
np.save(file_path, small_array)
array_files.append(file_path)
# Benchmark processing files individually (memory efficient)
start_time = time.time()
processed_count = 0
for file_path in array_files:
# Load, process, and immediately release
array = np.load(file_path)
# Simulate processing (tiling)
tiles = [array[i:i+10, j:j+10] for i in range(0, 50, 10) for j in range(0, 50, 10)]
processed_count += len(tiles)
del array, tiles # Explicit cleanup
end_time = time.time()
duration = end_time - start_time
files_per_second = file_count / duration
# Performance assertions
assert duration < 10.0, f"Memory-efficient processing took {duration:.4f}s, should be <10s"
assert files_per_second > 1, f"Processing rate {files_per_second:.2f} files/s, should be >1/s"
assert processed_count > 0, "Should have processed some tiles"
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
Workflow Performance Testing¶
Performance tests for complete workflows and processing pipelines.
End-to-End Performance Tests¶
Consolidated Performance Workflow Tests
This file consolidates tests from: - workflows/test_performance_aggregation.py - workflows/test_performance_integration.py
Covers workflow-level performance testing including: - End-to-end workflow performance benchmarks - Performance aggregation and reporting - Integration with CI/CD pipeline performance validation - JSON artifact storage and version control integration - Performance baseline establishment and validation - Cross-system performance consistency testing
BaseWorkflowPerformanceTest (TestCase)
¶
Base class for workflow-level performance tests with shared setup
Source code in hazelbean_tests/performance/test_workflows.py
class BaseWorkflowPerformanceTest(unittest.TestCase):
"""Base class for workflow-level performance tests with shared setup"""
def setUp(self):
"""Set up test fixtures"""
self.test_dir = tempfile.mkdtemp()
self.metrics_dir = os.path.join(os.path.dirname(__file__), "../../metrics")
os.makedirs(self.metrics_dir, exist_ok=True)
# Create ProjectFlow instance
self.p = hb.ProjectFlow(self.test_dir)
def tearDown(self):
"""Clean up test directories"""
shutil.rmtree(self.test_dir, ignore_errors=True)
TestPerformanceIntegration (BaseWorkflowPerformanceTest)
¶
Test performance integration workflows (from test_performance_integration.py)
Source code in hazelbean_tests/performance/test_workflows.py
class TestPerformanceIntegration(BaseWorkflowPerformanceTest):
"""Test performance integration workflows (from test_performance_integration.py)"""
@pytest.mark.benchmark
def test_json_artifact_storage_performance(self):
"""Test JSON artifact storage and version control integration performance"""
# Create performance data
performance_data = {
"timestamp": datetime.now().isoformat(),
"test_suite": "workflow_performance",
"metrics": {
"execution_time": 2.34,
"memory_usage_mb": 150.5,
"cpu_usage_percent": 45.2,
"disk_io_mb": 25.8
},
"environment": {
"python_version": sys.version,
"hazelbean_version": "dev",
"platform": sys.platform
},
"test_details": {
"total_tests": 10,
"passed_tests": 10,
"failed_tests": 0,
"skipped_tests": 0
}
}
# Benchmark JSON artifact creation and storage
import time
start_time = time.time()
# Create artifact file
artifact_file = os.path.join(self.metrics_dir, f"performance_artifact_{int(time.time())}.json")
with open(artifact_file, 'w') as f:
json.dump(performance_data, f, indent=2)
# Verify file was created
assert os.path.exists(artifact_file)
# Read back and verify
with open(artifact_file, 'r') as f:
loaded_data = json.load(f)
end_time = time.time()
storage_duration = end_time - start_time
# Performance assertions
assert storage_duration < 1.0, f"JSON artifact storage took {storage_duration:.4f}s, should be <1s"
assert loaded_data == performance_data, "Loaded data should match original"
# Cleanup
os.remove(artifact_file)
@pytest.mark.benchmark
def test_performance_baseline_validation_workflow(self):
"""Test performance baseline establishment and validation workflow"""
# Create baseline performance metrics
baseline_metrics = {
"get_path_single_call": 0.001,
"get_path_100_calls": 0.1,
"array_operations": 1.5,
"file_io_operations": 0.5
}
# Simulate current performance measurements
import time
start_time = time.time()
# Measure actual performance
measured_metrics = {}
# get_path performance
path_start = time.time()
result = self.p.get_path("test_file.txt")
measured_metrics["get_path_single_call"] = time.time() - path_start
# Array operations performance
array_start = time.time()
test_array = np.random.rand(100, 100)
processed = test_array * 2 + 1
result_sum = np.sum(processed)
measured_metrics["array_operations"] = time.time() - array_start
# File I/O performance
io_start = time.time()
temp_file = os.path.join(self.test_dir, "perf_test.txt")
with open(temp_file, 'w') as f:
f.write("performance test data" * 100)
with open(temp_file, 'r') as f:
content = f.read()
measured_metrics["file_io_operations"] = time.time() - io_start
end_time = time.time()
total_measurement_time = end_time - start_time
# Performance validation against baseline
performance_regressions = []
for metric, baseline_value in baseline_metrics.items():
if metric in measured_metrics:
measured_value = measured_metrics[metric]
# Allow 50% variance from baseline
if measured_value > baseline_value * 1.5:
performance_regressions.append({
"metric": metric,
"baseline": baseline_value,
"measured": measured_value,
"ratio": measured_value / baseline_value
})
# Performance assertions
assert total_measurement_time < 10.0, f"Performance measurement took {total_measurement_time:.4f}s, should be <10s"
# Allow some performance regressions in tests, but document them
if performance_regressions:
print(f"Performance regressions detected: {performance_regressions}")
# In real CI/CD, this might trigger warnings but not fail the test
# Verify all metrics were measured
assert len(measured_metrics) >= 3, "Should have measured multiple performance metrics"
@pytest.mark.benchmark
@pytest.mark.slow
def test_ci_cd_performance_integration(self):
"""Test integration with CI/CD pipeline performance validation"""
# Simulate CI/CD environment performance testing
import time
start_time = time.time()
ci_performance_data = {
"build_id": "test_build_12345",
"commit_hash": "abcd1234",
"branch": "main",
"timestamp": datetime.now().isoformat(),
"performance_tests": {}
}
# Run multiple performance tests as would happen in CI/CD
test_cases = [
("basic_operations", self._benchmark_basic_operations),
("file_processing", self._benchmark_file_processing),
("memory_usage", self._benchmark_memory_usage)
]
for test_name, test_function in test_cases:
test_start = time.time()
try:
test_result = test_function()
test_duration = time.time() - test_start
ci_performance_data["performance_tests"][test_name] = {
"status": "passed",
"duration": test_duration,
"result": test_result
}
except Exception as e:
test_duration = time.time() - test_start
ci_performance_data["performance_tests"][test_name] = {
"status": "failed",
"duration": test_duration,
"error": str(e)
}
end_time = time.time()
total_ci_time = end_time - start_time
# Save CI performance data
ci_artifact_file = os.path.join(self.metrics_dir, f"ci_performance_{int(time.time())}.json")
with open(ci_artifact_file, 'w') as f:
json.dump(ci_performance_data, f, indent=2)
# Performance assertions for CI/CD workflow
assert total_ci_time < 30.0, f"CI/CD performance tests took {total_ci_time:.4f}s, should be <30s"
assert os.path.exists(ci_artifact_file), "CI performance artifact should be created"
assert len(ci_performance_data["performance_tests"]) == len(test_cases), "All test cases should be executed"
# Cleanup
os.remove(ci_artifact_file)
def _benchmark_basic_operations(self):
"""Helper method for benchmarking basic operations"""
import time
start_time = time.time()
# Basic operations
for i in range(100):
path = self.p.get_path(f"test_file_{i}.txt")
duration = time.time() - start_time
return {"avg_time_per_operation": duration / 100}
def _benchmark_file_processing(self):
"""Helper method for benchmarking file processing"""
import time
start_time = time.time()
# File processing operations
test_files = []
for i in range(10):
temp_array = np.random.rand(50, 50)
temp_file = hb.temp('.npy', f'benchmark_{i}', True)
hb.save_array_as_npy(temp_array, temp_file)
test_files.append(temp_file)
duration = time.time() - start_time
return {"files_processed": len(test_files), "total_time": duration}
def _benchmark_memory_usage(self):
"""Helper method for benchmarking memory usage"""
import time
start_time = time.time()
# Memory-intensive operations
large_arrays = []
for i in range(5):
array = np.random.rand(200, 200)
large_arrays.append(array)
# Process arrays
results = []
for array in large_arrays:
result = np.sum(array)
results.append(result)
# Cleanup
del large_arrays
duration = time.time() - start_time
return {"arrays_processed": len(results), "total_time": duration}
test_json_artifact_storage_performance(self)
¶
Test JSON artifact storage and version control integration performance
Source code in hazelbean_tests/performance/test_workflows.py
@pytest.mark.benchmark
def test_json_artifact_storage_performance(self):
"""Test JSON artifact storage and version control integration performance"""
# Create performance data
performance_data = {
"timestamp": datetime.now().isoformat(),
"test_suite": "workflow_performance",
"metrics": {
"execution_time": 2.34,
"memory_usage_mb": 150.5,
"cpu_usage_percent": 45.2,
"disk_io_mb": 25.8
},
"environment": {
"python_version": sys.version,
"hazelbean_version": "dev",
"platform": sys.platform
},
"test_details": {
"total_tests": 10,
"passed_tests": 10,
"failed_tests": 0,
"skipped_tests": 0
}
}
# Benchmark JSON artifact creation and storage
import time
start_time = time.time()
# Create artifact file
artifact_file = os.path.join(self.metrics_dir, f"performance_artifact_{int(time.time())}.json")
with open(artifact_file, 'w') as f:
json.dump(performance_data, f, indent=2)
# Verify file was created
assert os.path.exists(artifact_file)
# Read back and verify
with open(artifact_file, 'r') as f:
loaded_data = json.load(f)
end_time = time.time()
storage_duration = end_time - start_time
# Performance assertions
assert storage_duration < 1.0, f"JSON artifact storage took {storage_duration:.4f}s, should be <1s"
assert loaded_data == performance_data, "Loaded data should match original"
# Cleanup
os.remove(artifact_file)
test_performance_baseline_validation_workflow(self)
¶
Test performance baseline establishment and validation workflow
Source code in hazelbean_tests/performance/test_workflows.py
@pytest.mark.benchmark
def test_performance_baseline_validation_workflow(self):
"""Test performance baseline establishment and validation workflow"""
# Create baseline performance metrics
baseline_metrics = {
"get_path_single_call": 0.001,
"get_path_100_calls": 0.1,
"array_operations": 1.5,
"file_io_operations": 0.5
}
# Simulate current performance measurements
import time
start_time = time.time()
# Measure actual performance
measured_metrics = {}
# get_path performance
path_start = time.time()
result = self.p.get_path("test_file.txt")
measured_metrics["get_path_single_call"] = time.time() - path_start
# Array operations performance
array_start = time.time()
test_array = np.random.rand(100, 100)
processed = test_array * 2 + 1
result_sum = np.sum(processed)
measured_metrics["array_operations"] = time.time() - array_start
# File I/O performance
io_start = time.time()
temp_file = os.path.join(self.test_dir, "perf_test.txt")
with open(temp_file, 'w') as f:
f.write("performance test data" * 100)
with open(temp_file, 'r') as f:
content = f.read()
measured_metrics["file_io_operations"] = time.time() - io_start
end_time = time.time()
total_measurement_time = end_time - start_time
# Performance validation against baseline
performance_regressions = []
for metric, baseline_value in baseline_metrics.items():
if metric in measured_metrics:
measured_value = measured_metrics[metric]
# Allow 50% variance from baseline
if measured_value > baseline_value * 1.5:
performance_regressions.append({
"metric": metric,
"baseline": baseline_value,
"measured": measured_value,
"ratio": measured_value / baseline_value
})
# Performance assertions
assert total_measurement_time < 10.0, f"Performance measurement took {total_measurement_time:.4f}s, should be <10s"
# Allow some performance regressions in tests, but document them
if performance_regressions:
print(f"Performance regressions detected: {performance_regressions}")
# In real CI/CD, this might trigger warnings but not fail the test
# Verify all metrics were measured
assert len(measured_metrics) >= 3, "Should have measured multiple performance metrics"
test_ci_cd_performance_integration(self)
¶
Test integration with CI/CD pipeline performance validation
Source code in hazelbean_tests/performance/test_workflows.py
@pytest.mark.benchmark
@pytest.mark.slow
def test_ci_cd_performance_integration(self):
"""Test integration with CI/CD pipeline performance validation"""
# Simulate CI/CD environment performance testing
import time
start_time = time.time()
ci_performance_data = {
"build_id": "test_build_12345",
"commit_hash": "abcd1234",
"branch": "main",
"timestamp": datetime.now().isoformat(),
"performance_tests": {}
}
# Run multiple performance tests as would happen in CI/CD
test_cases = [
("basic_operations", self._benchmark_basic_operations),
("file_processing", self._benchmark_file_processing),
("memory_usage", self._benchmark_memory_usage)
]
for test_name, test_function in test_cases:
test_start = time.time()
try:
test_result = test_function()
test_duration = time.time() - test_start
ci_performance_data["performance_tests"][test_name] = {
"status": "passed",
"duration": test_duration,
"result": test_result
}
except Exception as e:
test_duration = time.time() - test_start
ci_performance_data["performance_tests"][test_name] = {
"status": "failed",
"duration": test_duration,
"error": str(e)
}
end_time = time.time()
total_ci_time = end_time - start_time
# Save CI performance data
ci_artifact_file = os.path.join(self.metrics_dir, f"ci_performance_{int(time.time())}.json")
with open(ci_artifact_file, 'w') as f:
json.dump(ci_performance_data, f, indent=2)
# Performance assertions for CI/CD workflow
assert total_ci_time < 30.0, f"CI/CD performance tests took {total_ci_time:.4f}s, should be <30s"
assert os.path.exists(ci_artifact_file), "CI performance artifact should be created"
assert len(ci_performance_data["performance_tests"]) == len(test_cases), "All test cases should be executed"
# Cleanup
os.remove(ci_artifact_file)
_benchmark_basic_operations(self)
private
¶
Helper method for benchmarking basic operations
Source code in hazelbean_tests/performance/test_workflows.py
def _benchmark_basic_operations(self):
"""Helper method for benchmarking basic operations"""
import time
start_time = time.time()
# Basic operations
for i in range(100):
path = self.p.get_path(f"test_file_{i}.txt")
duration = time.time() - start_time
return {"avg_time_per_operation": duration / 100}
_benchmark_file_processing(self)
private
¶
Helper method for benchmarking file processing
Source code in hazelbean_tests/performance/test_workflows.py
def _benchmark_file_processing(self):
"""Helper method for benchmarking file processing"""
import time
start_time = time.time()
# File processing operations
test_files = []
for i in range(10):
temp_array = np.random.rand(50, 50)
temp_file = hb.temp('.npy', f'benchmark_{i}', True)
hb.save_array_as_npy(temp_array, temp_file)
test_files.append(temp_file)
duration = time.time() - start_time
return {"files_processed": len(test_files), "total_time": duration}
_benchmark_memory_usage(self)
private
¶
Helper method for benchmarking memory usage
Source code in hazelbean_tests/performance/test_workflows.py
def _benchmark_memory_usage(self):
"""Helper method for benchmarking memory usage"""
import time
start_time = time.time()
# Memory-intensive operations
large_arrays = []
for i in range(5):
array = np.random.rand(200, 200)
large_arrays.append(array)
# Process arrays
results = []
for array in large_arrays:
result = np.sum(array)
results.append(result)
# Cleanup
del large_arrays
duration = time.time() - start_time
return {"arrays_processed": len(results), "total_time": duration}
TestPerformanceAggregation (BaseWorkflowPerformanceTest)
¶
Test performance aggregation workflows (from test_performance_aggregation.py)
Source code in hazelbean_tests/performance/test_workflows.py
class TestPerformanceAggregation(BaseWorkflowPerformanceTest):
"""Test performance aggregation workflows (from test_performance_aggregation.py)"""
@pytest.mark.benchmark
def test_performance_metrics_aggregation(self):
"""Test aggregation of performance metrics from multiple sources"""
# Create multiple performance metric sources
metric_sources = [
{
"source": "unit_tests",
"metrics": {
"avg_execution_time": 0.05,
"max_execution_time": 0.2,
"total_tests": 150,
"memory_peak_mb": 45.2
}
},
{
"source": "integration_tests",
"metrics": {
"avg_execution_time": 1.2,
"max_execution_time": 5.5,
"total_tests": 25,
"memory_peak_mb": 120.8
}
},
{
"source": "performance_tests",
"metrics": {
"avg_execution_time": 2.8,
"max_execution_time": 15.0,
"total_tests": 10,
"memory_peak_mb": 200.5
}
}
]
# Benchmark aggregation process
import time
start_time = time.time()
# Aggregate metrics
aggregated = {
"total_tests": 0,
"weighted_avg_execution_time": 0,
"overall_max_execution_time": 0,
"total_memory_peak_mb": 0,
"sources": len(metric_sources)
}
total_execution_time = 0
for source_data in metric_sources:
metrics = source_data["metrics"]
aggregated["total_tests"] += metrics["total_tests"]
total_execution_time += metrics["avg_execution_time"] * metrics["total_tests"]
aggregated["overall_max_execution_time"] = max(
aggregated["overall_max_execution_time"],
metrics["max_execution_time"]
)
aggregated["total_memory_peak_mb"] = max(
aggregated["total_memory_peak_mb"],
metrics["memory_peak_mb"]
)
# Calculate weighted average
if aggregated["total_tests"] > 0:
aggregated["weighted_avg_execution_time"] = total_execution_time / aggregated["total_tests"]
end_time = time.time()
aggregation_time = end_time - start_time
# Performance assertions
assert aggregation_time < 1.0, f"Metrics aggregation took {aggregation_time:.4f}s, should be <1s"
assert aggregated["total_tests"] == 185, "Should aggregate all tests"
assert aggregated["sources"] == 3, "Should process all sources"
assert aggregated["weighted_avg_execution_time"] > 0, "Should calculate weighted average"
assert aggregated["overall_max_execution_time"] == 15.0, "Should find maximum execution time"
@pytest.mark.benchmark
def test_performance_trend_analysis(self):
"""Test performance trend analysis workflow"""
# Create historical performance data
historical_data = []
base_time = datetime.now().timestamp()
for i in range(10): # 10 data points
data_point = {
"timestamp": base_time - (i * 86400), # Daily intervals
"metrics": {
"avg_response_time": 0.1 + (i * 0.01), # Gradually increasing
"throughput": 1000 - (i * 10), # Gradually decreasing
"error_rate": 0.01 + (i * 0.001), # Gradually increasing
"memory_usage": 100 + (i * 5) # Gradually increasing
}
}
historical_data.append(data_point)
# Benchmark trend analysis
import time
start_time = time.time()
# Analyze trends
trends = {}
metrics_to_analyze = ["avg_response_time", "throughput", "error_rate", "memory_usage"]
for metric in metrics_to_analyze:
values = [dp["metrics"][metric] for dp in historical_data]
# Simple trend analysis
if len(values) >= 2:
trend_direction = "increasing" if values[0] > values[-1] else "decreasing"
trend_magnitude = abs(values[0] - values[-1]) / values[-1]
trends[metric] = {
"direction": trend_direction,
"magnitude_percent": trend_magnitude * 100,
"latest_value": values[0],
"oldest_value": values[-1]
}
end_time = time.time()
analysis_time = end_time - start_time
# Performance assertions
assert analysis_time < 2.0, f"Trend analysis took {analysis_time:.4f}s, should be <2s"
assert len(trends) == len(metrics_to_analyze), "Should analyze all metrics"
# Verify trend detection
assert trends["avg_response_time"]["direction"] == "decreasing", "Should detect response time trend"
assert trends["throughput"]["direction"] == "increasing", "Should detect throughput trend"
@pytest.mark.benchmark
def test_performance_report_generation(self):
"""Test performance report generation workflow"""
# Create comprehensive performance data
performance_data = {
"report_metadata": {
"generated_at": datetime.now().isoformat(),
"report_type": "comprehensive_performance",
"period": "weekly",
"version": "1.0"
},
"summary": {
"total_tests_executed": 500,
"total_execution_time": 125.5,
"avg_test_time": 0.251,
"performance_score": 85.2
},
"detailed_metrics": {
"by_category": {
"unit_tests": {"count": 400, "avg_time": 0.05, "total_time": 20.0},
"integration_tests": {"count": 75, "avg_time": 1.2, "total_time": 90.0},
"performance_tests": {"count": 25, "avg_time": 0.62, "total_time": 15.5}
},
"by_component": {
"get_path": {"calls": 10000, "avg_time": 0.001, "total_time": 10.0},
"array_operations": {"calls": 500, "avg_time": 0.15, "total_time": 75.0},
"file_io": {"calls": 200, "avg_time": 0.2, "total_time": 40.0}
}
}
}
# Benchmark report generation
import time
start_time = time.time()
# Generate report file
report_file = os.path.join(self.metrics_dir, f"performance_report_{int(time.time())}.json")
with open(report_file, 'w') as f:
json.dump(performance_data, f, indent=2)
# Generate summary statistics
summary_stats = {
"total_categories": len(performance_data["detailed_metrics"]["by_category"]),
"total_components": len(performance_data["detailed_metrics"]["by_component"]),
"fastest_category": min(
performance_data["detailed_metrics"]["by_category"].items(),
key=lambda x: x[1]["avg_time"]
)[0],
"slowest_category": max(
performance_data["detailed_metrics"]["by_category"].items(),
key=lambda x: x[1]["avg_time"]
)[0]
}
end_time = time.time()
report_generation_time = end_time - start_time
# Performance assertions
assert report_generation_time < 5.0, f"Report generation took {report_generation_time:.4f}s, should be <5s"
assert os.path.exists(report_file), "Report file should be created"
# Verify report content
with open(report_file, 'r') as f:
generated_report = json.load(f)
assert generated_report == performance_data, "Generated report should match input data"
assert summary_stats["fastest_category"] == "unit_tests", "Should identify fastest category"
assert summary_stats["slowest_category"] == "integration_tests", "Should identify slowest category"
# Cleanup
os.remove(report_file)
@pytest.mark.benchmark
@pytest.mark.slow
def test_cross_platform_performance_consistency(self):
"""Test performance consistency across different environments"""
# Simulate cross-platform performance testing
platforms = ["linux", "windows", "macos"] # Simulated platform data
platform_results = {}
# Benchmark cross-platform consistency measurement
import time
start_time = time.time()
for platform in platforms:
# Simulate platform-specific performance measurements
platform_start = time.time()
# Run standard performance tests
measurements = {
"get_path_performance": self._measure_get_path_performance(),
"array_processing": self._measure_array_processing(),
"file_io_performance": self._measure_file_io_performance()
}
platform_duration = time.time() - platform_start
platform_results[platform] = {
"measurements": measurements,
"total_measurement_time": platform_duration
}
end_time = time.time()
total_cross_platform_time = end_time - start_time
# Analyze consistency
consistency_analysis = {}
for metric in ["get_path_performance", "array_processing", "file_io_performance"]:
values = [results["measurements"][metric] for results in platform_results.values()]
avg_value = sum(values) / len(values)
max_deviation = max(abs(v - avg_value) for v in values)
consistency_percentage = (1 - max_deviation / avg_value) * 100 if avg_value > 0 else 0
consistency_analysis[metric] = {
"average": avg_value,
"max_deviation": max_deviation,
"consistency_percentage": consistency_percentage,
"values": dict(zip(platforms, values))
}
# Performance assertions
assert total_cross_platform_time < 20.0, f"Cross-platform testing took {total_cross_platform_time:.4f}s, should be <20s"
assert len(platform_results) == len(platforms), "Should test all platforms"
# Consistency assertions (allow reasonable variance)
for metric, analysis in consistency_analysis.items():
assert analysis["consistency_percentage"] > 50, f"Metric '{metric}' consistency {analysis['consistency_percentage']:.1f}% too low"
def _measure_get_path_performance(self):
"""Helper method to measure get_path performance"""
import time
start_time = time.time()
for i in range(100):
self.p.get_path(f"test_file_{i}.txt")
return time.time() - start_time
def _measure_array_processing(self):
"""Helper method to measure array processing performance"""
import time
start_time = time.time()
for i in range(10):
array = np.random.rand(100, 100)
result = np.sum(array * 2)
return time.time() - start_time
def _measure_file_io_performance(self):
"""Helper method to measure file I/O performance"""
import time
start_time = time.time()
for i in range(10):
temp_file = os.path.join(self.test_dir, f"perf_test_{i}.txt")
with open(temp_file, 'w') as f:
f.write("test data" * 100)
with open(temp_file, 'r') as f:
content = f.read()
return time.time() - start_time
test_performance_metrics_aggregation(self)
¶
Test aggregation of performance metrics from multiple sources
Source code in hazelbean_tests/performance/test_workflows.py
@pytest.mark.benchmark
def test_performance_metrics_aggregation(self):
"""Test aggregation of performance metrics from multiple sources"""
# Create multiple performance metric sources
metric_sources = [
{
"source": "unit_tests",
"metrics": {
"avg_execution_time": 0.05,
"max_execution_time": 0.2,
"total_tests": 150,
"memory_peak_mb": 45.2
}
},
{
"source": "integration_tests",
"metrics": {
"avg_execution_time": 1.2,
"max_execution_time": 5.5,
"total_tests": 25,
"memory_peak_mb": 120.8
}
},
{
"source": "performance_tests",
"metrics": {
"avg_execution_time": 2.8,
"max_execution_time": 15.0,
"total_tests": 10,
"memory_peak_mb": 200.5
}
}
]
# Benchmark aggregation process
import time
start_time = time.time()
# Aggregate metrics
aggregated = {
"total_tests": 0,
"weighted_avg_execution_time": 0,
"overall_max_execution_time": 0,
"total_memory_peak_mb": 0,
"sources": len(metric_sources)
}
total_execution_time = 0
for source_data in metric_sources:
metrics = source_data["metrics"]
aggregated["total_tests"] += metrics["total_tests"]
total_execution_time += metrics["avg_execution_time"] * metrics["total_tests"]
aggregated["overall_max_execution_time"] = max(
aggregated["overall_max_execution_time"],
metrics["max_execution_time"]
)
aggregated["total_memory_peak_mb"] = max(
aggregated["total_memory_peak_mb"],
metrics["memory_peak_mb"]
)
# Calculate weighted average
if aggregated["total_tests"] > 0:
aggregated["weighted_avg_execution_time"] = total_execution_time / aggregated["total_tests"]
end_time = time.time()
aggregation_time = end_time - start_time
# Performance assertions
assert aggregation_time < 1.0, f"Metrics aggregation took {aggregation_time:.4f}s, should be <1s"
assert aggregated["total_tests"] == 185, "Should aggregate all tests"
assert aggregated["sources"] == 3, "Should process all sources"
assert aggregated["weighted_avg_execution_time"] > 0, "Should calculate weighted average"
assert aggregated["overall_max_execution_time"] == 15.0, "Should find maximum execution time"
test_performance_trend_analysis(self)
¶
Test performance trend analysis workflow
Source code in hazelbean_tests/performance/test_workflows.py
@pytest.mark.benchmark
def test_performance_trend_analysis(self):
"""Test performance trend analysis workflow"""
# Create historical performance data
historical_data = []
base_time = datetime.now().timestamp()
for i in range(10): # 10 data points
data_point = {
"timestamp": base_time - (i * 86400), # Daily intervals
"metrics": {
"avg_response_time": 0.1 + (i * 0.01), # Gradually increasing
"throughput": 1000 - (i * 10), # Gradually decreasing
"error_rate": 0.01 + (i * 0.001), # Gradually increasing
"memory_usage": 100 + (i * 5) # Gradually increasing
}
}
historical_data.append(data_point)
# Benchmark trend analysis
import time
start_time = time.time()
# Analyze trends
trends = {}
metrics_to_analyze = ["avg_response_time", "throughput", "error_rate", "memory_usage"]
for metric in metrics_to_analyze:
values = [dp["metrics"][metric] for dp in historical_data]
# Simple trend analysis
if len(values) >= 2:
trend_direction = "increasing" if values[0] > values[-1] else "decreasing"
trend_magnitude = abs(values[0] - values[-1]) / values[-1]
trends[metric] = {
"direction": trend_direction,
"magnitude_percent": trend_magnitude * 100,
"latest_value": values[0],
"oldest_value": values[-1]
}
end_time = time.time()
analysis_time = end_time - start_time
# Performance assertions
assert analysis_time < 2.0, f"Trend analysis took {analysis_time:.4f}s, should be <2s"
assert len(trends) == len(metrics_to_analyze), "Should analyze all metrics"
# Verify trend detection
assert trends["avg_response_time"]["direction"] == "decreasing", "Should detect response time trend"
assert trends["throughput"]["direction"] == "increasing", "Should detect throughput trend"
test_performance_report_generation(self)
¶
Test performance report generation workflow
Source code in hazelbean_tests/performance/test_workflows.py
@pytest.mark.benchmark
def test_performance_report_generation(self):
"""Test performance report generation workflow"""
# Create comprehensive performance data
performance_data = {
"report_metadata": {
"generated_at": datetime.now().isoformat(),
"report_type": "comprehensive_performance",
"period": "weekly",
"version": "1.0"
},
"summary": {
"total_tests_executed": 500,
"total_execution_time": 125.5,
"avg_test_time": 0.251,
"performance_score": 85.2
},
"detailed_metrics": {
"by_category": {
"unit_tests": {"count": 400, "avg_time": 0.05, "total_time": 20.0},
"integration_tests": {"count": 75, "avg_time": 1.2, "total_time": 90.0},
"performance_tests": {"count": 25, "avg_time": 0.62, "total_time": 15.5}
},
"by_component": {
"get_path": {"calls": 10000, "avg_time": 0.001, "total_time": 10.0},
"array_operations": {"calls": 500, "avg_time": 0.15, "total_time": 75.0},
"file_io": {"calls": 200, "avg_time": 0.2, "total_time": 40.0}
}
}
}
# Benchmark report generation
import time
start_time = time.time()
# Generate report file
report_file = os.path.join(self.metrics_dir, f"performance_report_{int(time.time())}.json")
with open(report_file, 'w') as f:
json.dump(performance_data, f, indent=2)
# Generate summary statistics
summary_stats = {
"total_categories": len(performance_data["detailed_metrics"]["by_category"]),
"total_components": len(performance_data["detailed_metrics"]["by_component"]),
"fastest_category": min(
performance_data["detailed_metrics"]["by_category"].items(),
key=lambda x: x[1]["avg_time"]
)[0],
"slowest_category": max(
performance_data["detailed_metrics"]["by_category"].items(),
key=lambda x: x[1]["avg_time"]
)[0]
}
end_time = time.time()
report_generation_time = end_time - start_time
# Performance assertions
assert report_generation_time < 5.0, f"Report generation took {report_generation_time:.4f}s, should be <5s"
assert os.path.exists(report_file), "Report file should be created"
# Verify report content
with open(report_file, 'r') as f:
generated_report = json.load(f)
assert generated_report == performance_data, "Generated report should match input data"
assert summary_stats["fastest_category"] == "unit_tests", "Should identify fastest category"
assert summary_stats["slowest_category"] == "integration_tests", "Should identify slowest category"
# Cleanup
os.remove(report_file)
test_cross_platform_performance_consistency(self)
¶
Test performance consistency across different environments
Source code in hazelbean_tests/performance/test_workflows.py
@pytest.mark.benchmark
@pytest.mark.slow
def test_cross_platform_performance_consistency(self):
"""Test performance consistency across different environments"""
# Simulate cross-platform performance testing
platforms = ["linux", "windows", "macos"] # Simulated platform data
platform_results = {}
# Benchmark cross-platform consistency measurement
import time
start_time = time.time()
for platform in platforms:
# Simulate platform-specific performance measurements
platform_start = time.time()
# Run standard performance tests
measurements = {
"get_path_performance": self._measure_get_path_performance(),
"array_processing": self._measure_array_processing(),
"file_io_performance": self._measure_file_io_performance()
}
platform_duration = time.time() - platform_start
platform_results[platform] = {
"measurements": measurements,
"total_measurement_time": platform_duration
}
end_time = time.time()
total_cross_platform_time = end_time - start_time
# Analyze consistency
consistency_analysis = {}
for metric in ["get_path_performance", "array_processing", "file_io_performance"]:
values = [results["measurements"][metric] for results in platform_results.values()]
avg_value = sum(values) / len(values)
max_deviation = max(abs(v - avg_value) for v in values)
consistency_percentage = (1 - max_deviation / avg_value) * 100 if avg_value > 0 else 0
consistency_analysis[metric] = {
"average": avg_value,
"max_deviation": max_deviation,
"consistency_percentage": consistency_percentage,
"values": dict(zip(platforms, values))
}
# Performance assertions
assert total_cross_platform_time < 20.0, f"Cross-platform testing took {total_cross_platform_time:.4f}s, should be <20s"
assert len(platform_results) == len(platforms), "Should test all platforms"
# Consistency assertions (allow reasonable variance)
for metric, analysis in consistency_analysis.items():
assert analysis["consistency_percentage"] > 50, f"Metric '{metric}' consistency {analysis['consistency_percentage']:.1f}% too low"
_measure_get_path_performance(self)
private
¶
Helper method to measure get_path performance
_measure_array_processing(self)
private
¶
Helper method to measure array processing performance
Source code in hazelbean_tests/performance/test_workflows.py
_measure_file_io_performance(self)
private
¶
Helper method to measure file I/O performance
Source code in hazelbean_tests/performance/test_workflows.py
def _measure_file_io_performance(self):
"""Helper method to measure file I/O performance"""
import time
start_time = time.time()
for i in range(10):
temp_file = os.path.join(self.test_dir, f"perf_test_{i}.txt")
with open(temp_file, 'w') as f:
f.write("test data" * 100)
with open(temp_file, 'r') as f:
content = f.read()
return time.time() - start_time
Baseline Management Testing¶
Tests for the baseline management system that tracks performance changes over time.
Baseline Management Tests¶
Comprehensive tests for Baseline Management System
This test suite validates:
- Standardized baseline JSON structure creation
- Baseline comparison logic for performance regression detection
- Trend analysis and historical tracking capabilities
- Version control integration for baseline artifacts
Story 6: Baseline Establishment - All Tasks (6.1-6.4) Test Quality Standards: Tests must not fail due to test setup issues (unacceptable) but may discover bugs in baseline logic (acceptable and valuable discovery).
BaselineManagerTest (TestCase)
¶
Base test class for baseline manager functionality
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
class BaselineManagerTest(unittest.TestCase):
"""Base test class for baseline manager functionality"""
def setUp(self):
"""Set up test fixtures and temporary directories"""
self.test_dir = tempfile.mkdtemp()
self.metrics_dir = os.path.join(self.test_dir, "metrics")
os.makedirs(self.metrics_dir, exist_ok=True)
# Create baseline manager instance
self.manager = BaselineManager(self.metrics_dir)
# Create sample benchmark data for testing
self.sample_benchmark_data = self.create_sample_benchmark_data()
def tearDown(self):
"""Clean up test directories"""
shutil.rmtree(self.test_dir, ignore_errors=True)
def create_sample_benchmark_data(self):
"""Create sample benchmark data for testing"""
return {
"machine_info": {
"node": "test-machine",
"processor": "arm",
"machine": "arm64",
"python_compiler": "Test Compiler",
"python_implementation": "CPython",
"python_version": "3.13.2",
"system": "Darwin",
"release": "24.5.0",
"cpu": {
"arch": "ARM_8",
"bits": 64,
"count": 8,
"brand_raw": "Test CPU"
}
},
"commit_info": {
"id": "test_commit_hash_12345",
"time": "2025-01-01T12:00:00-05:00",
"dirty": False,
"branch": "test_branch"
},
"benchmarks": [
{
"name": "test_get_path_benchmark",
"fullname": "hazelbean_tests/performance/test_get_path_benchmark",
"stats": {
"min": 0.01,
"max": 0.02,
"mean": 0.015,
"stddev": 0.002,
"rounds": 50,
"median": 0.015
}
},
{
"name": "test_tiling_benchmark",
"fullname": "hazelbean_tests/performance/test_tiling_benchmark",
"stats": {
"min": 0.05,
"max": 0.08,
"mean": 0.065,
"stddev": 0.005,
"rounds": 30,
"median": 0.064
}
},
{
"name": "test_array_operations_benchmark",
"fullname": "hazelbean_tests/performance/test_array_benchmark",
"stats": {
"min": 0.001,
"max": 0.003,
"mean": 0.002,
"stddev": 0.0003,
"rounds": 100,
"median": 0.002
}
}
]
}
create_sample_benchmark_data(self)
¶
Create sample benchmark data for testing
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
def create_sample_benchmark_data(self):
"""Create sample benchmark data for testing"""
return {
"machine_info": {
"node": "test-machine",
"processor": "arm",
"machine": "arm64",
"python_compiler": "Test Compiler",
"python_implementation": "CPython",
"python_version": "3.13.2",
"system": "Darwin",
"release": "24.5.0",
"cpu": {
"arch": "ARM_8",
"bits": 64,
"count": 8,
"brand_raw": "Test CPU"
}
},
"commit_info": {
"id": "test_commit_hash_12345",
"time": "2025-01-01T12:00:00-05:00",
"dirty": False,
"branch": "test_branch"
},
"benchmarks": [
{
"name": "test_get_path_benchmark",
"fullname": "hazelbean_tests/performance/test_get_path_benchmark",
"stats": {
"min": 0.01,
"max": 0.02,
"mean": 0.015,
"stddev": 0.002,
"rounds": 50,
"median": 0.015
}
},
{
"name": "test_tiling_benchmark",
"fullname": "hazelbean_tests/performance/test_tiling_benchmark",
"stats": {
"min": 0.05,
"max": 0.08,
"mean": 0.065,
"stddev": 0.005,
"rounds": 30,
"median": 0.064
}
},
{
"name": "test_array_operations_benchmark",
"fullname": "hazelbean_tests/performance/test_array_benchmark",
"stats": {
"min": 0.001,
"max": 0.003,
"mean": 0.002,
"stddev": 0.0003,
"rounds": 100,
"median": 0.002
}
}
]
}
TestBaselineStructureCreation (BaselineManagerTest)
¶
Test Task 6.1: Create baseline JSON structure for all benchmark metrics
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
class TestBaselineStructureCreation(BaselineManagerTest):
"""Test Task 6.1: Create baseline JSON structure for all benchmark metrics"""
@pytest.mark.benchmark
@pytest.mark.performance
def test_create_standardized_baseline_structure(self):
"""Test creation of standardized baseline JSON structure"""
# Arrange
sample_data = self.sample_benchmark_data
# Act
baseline_structure = self.manager.create_standardized_baseline_structure(sample_data)
# Assert - Verify required top-level structure
required_sections = [
"baseline_metadata",
"version_control_info",
"system_environment",
"baseline_statistics",
"benchmark_categories",
"quality_metrics",
"raw_benchmark_data",
"validation_info"
]
for section in required_sections:
self.assertIn(section, baseline_structure, f"Missing required section: {section}")
# Verify baseline metadata structure
metadata = baseline_structure["baseline_metadata"]
self.assertIn("version", metadata)
self.assertIn("created_at", metadata)
self.assertIn("schema_version", metadata)
self.assertEqual(metadata["schema_version"], "2.0.0")
self.assertEqual(metadata["regression_threshold_percent"], 10.0)
# Verify validation info
validation = baseline_structure["validation_info"]
self.assertEqual(validation["total_benchmarks"], 3)
self.assertEqual(validation["valid_benchmarks"], 3)
self.assertTrue(validation["statistical_confidence_met"])
@pytest.mark.benchmark
def test_baseline_statistics_calculation(self):
"""Test comprehensive baseline statistics calculation"""
# Arrange
sample_data = self.sample_benchmark_data
# Act
baseline_structure = self.manager.create_standardized_baseline_structure(sample_data)
# Assert
stats = baseline_structure["baseline_statistics"]["aggregate_statistics"]
# Verify basic statistics are present and reasonable
self.assertIn("mean_execution_time", stats)
self.assertIn("median_execution_time", stats)
self.assertIn("std_deviation", stats)
self.assertIn("min_time", stats)
self.assertIn("max_time", stats)
# Verify statistical values are reasonable
self.assertGreater(stats["mean_execution_time"], 0)
self.assertGreaterEqual(stats["std_deviation"], 0)
self.assertLessEqual(stats["min_time"], stats["max_time"])
self.assertEqual(stats["total_benchmarks"], 3)
# Verify confidence intervals
ci = baseline_structure["baseline_statistics"]["confidence_intervals"]
self.assertIn("lower", ci)
self.assertIn("upper", ci)
self.assertLess(ci["lower"], ci["upper"])
@pytest.mark.benchmark
def test_benchmark_categorization(self):
"""Test benchmark categorization by functionality"""
# Arrange
sample_data = self.sample_benchmark_data
# Act
baseline_structure = self.manager.create_standardized_baseline_structure(sample_data)
# Assert
categories = baseline_structure["benchmark_categories"]
# Verify categories are created
expected_categories = [
"path_resolution",
"tiling_operations",
"data_processing",
"io_operations",
"computational",
"integration",
"uncategorized"
]
for category in expected_categories:
self.assertIn(category, categories)
# Verify specific categorization
self.assertIn("test_get_path_benchmark", categories["path_resolution"])
self.assertIn("test_tiling_benchmark", categories["tiling_operations"])
self.assertIn("test_array_operations_benchmark", categories["data_processing"])
@pytest.mark.benchmark
def test_quality_metrics_calculation(self):
"""Test quality metrics calculation for baseline establishment"""
# Arrange
sample_data = self.sample_benchmark_data
# Act
baseline_structure = self.manager.create_standardized_baseline_structure(sample_data)
# Assert
quality = baseline_structure["quality_metrics"]
# Verify quality score calculation
self.assertIn("baseline_quality_score", quality)
self.assertIsInstance(quality["baseline_quality_score"], (int, float))
self.assertGreaterEqual(quality["baseline_quality_score"], 0)
self.assertLessEqual(quality["baseline_quality_score"], 100)
# Verify statistical reliability assessment
reliability = quality["statistical_reliability"]
self.assertIn("sufficient_sample_size", reliability)
self.assertIn("acceptable_variance", reliability)
self.assertIn("outlier_percentage", reliability)
test_create_standardized_baseline_structure(self)
¶
Test creation of standardized baseline JSON structure
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
@pytest.mark.performance
def test_create_standardized_baseline_structure(self):
"""Test creation of standardized baseline JSON structure"""
# Arrange
sample_data = self.sample_benchmark_data
# Act
baseline_structure = self.manager.create_standardized_baseline_structure(sample_data)
# Assert - Verify required top-level structure
required_sections = [
"baseline_metadata",
"version_control_info",
"system_environment",
"baseline_statistics",
"benchmark_categories",
"quality_metrics",
"raw_benchmark_data",
"validation_info"
]
for section in required_sections:
self.assertIn(section, baseline_structure, f"Missing required section: {section}")
# Verify baseline metadata structure
metadata = baseline_structure["baseline_metadata"]
self.assertIn("version", metadata)
self.assertIn("created_at", metadata)
self.assertIn("schema_version", metadata)
self.assertEqual(metadata["schema_version"], "2.0.0")
self.assertEqual(metadata["regression_threshold_percent"], 10.0)
# Verify validation info
validation = baseline_structure["validation_info"]
self.assertEqual(validation["total_benchmarks"], 3)
self.assertEqual(validation["valid_benchmarks"], 3)
self.assertTrue(validation["statistical_confidence_met"])
test_baseline_statistics_calculation(self)
¶
Test comprehensive baseline statistics calculation
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_baseline_statistics_calculation(self):
"""Test comprehensive baseline statistics calculation"""
# Arrange
sample_data = self.sample_benchmark_data
# Act
baseline_structure = self.manager.create_standardized_baseline_structure(sample_data)
# Assert
stats = baseline_structure["baseline_statistics"]["aggregate_statistics"]
# Verify basic statistics are present and reasonable
self.assertIn("mean_execution_time", stats)
self.assertIn("median_execution_time", stats)
self.assertIn("std_deviation", stats)
self.assertIn("min_time", stats)
self.assertIn("max_time", stats)
# Verify statistical values are reasonable
self.assertGreater(stats["mean_execution_time"], 0)
self.assertGreaterEqual(stats["std_deviation"], 0)
self.assertLessEqual(stats["min_time"], stats["max_time"])
self.assertEqual(stats["total_benchmarks"], 3)
# Verify confidence intervals
ci = baseline_structure["baseline_statistics"]["confidence_intervals"]
self.assertIn("lower", ci)
self.assertIn("upper", ci)
self.assertLess(ci["lower"], ci["upper"])
test_benchmark_categorization(self)
¶
Test benchmark categorization by functionality
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_benchmark_categorization(self):
"""Test benchmark categorization by functionality"""
# Arrange
sample_data = self.sample_benchmark_data
# Act
baseline_structure = self.manager.create_standardized_baseline_structure(sample_data)
# Assert
categories = baseline_structure["benchmark_categories"]
# Verify categories are created
expected_categories = [
"path_resolution",
"tiling_operations",
"data_processing",
"io_operations",
"computational",
"integration",
"uncategorized"
]
for category in expected_categories:
self.assertIn(category, categories)
# Verify specific categorization
self.assertIn("test_get_path_benchmark", categories["path_resolution"])
self.assertIn("test_tiling_benchmark", categories["tiling_operations"])
self.assertIn("test_array_operations_benchmark", categories["data_processing"])
test_quality_metrics_calculation(self)
¶
Test quality metrics calculation for baseline establishment
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_quality_metrics_calculation(self):
"""Test quality metrics calculation for baseline establishment"""
# Arrange
sample_data = self.sample_benchmark_data
# Act
baseline_structure = self.manager.create_standardized_baseline_structure(sample_data)
# Assert
quality = baseline_structure["quality_metrics"]
# Verify quality score calculation
self.assertIn("baseline_quality_score", quality)
self.assertIsInstance(quality["baseline_quality_score"], (int, float))
self.assertGreaterEqual(quality["baseline_quality_score"], 0)
self.assertLessEqual(quality["baseline_quality_score"], 100)
# Verify statistical reliability assessment
reliability = quality["statistical_reliability"]
self.assertIn("sufficient_sample_size", reliability)
self.assertIn("acceptable_variance", reliability)
self.assertIn("outlier_percentage", reliability)
TestBaselineComparison (BaselineManagerTest)
¶
Test Task 6.2: Implement baseline comparison logic for performance regression detection
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
class TestBaselineComparison(BaselineManagerTest):
"""Test Task 6.2: Implement baseline comparison logic for performance regression detection"""
def setUp(self):
"""Set up test fixtures including baseline data"""
super().setUp()
# Create and save a baseline for comparison tests
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
self.manager.save_baseline(baseline_structure)
@pytest.mark.benchmark
@pytest.mark.performance
def test_baseline_comparison_no_regression(self):
"""Test baseline comparison when no regression is detected"""
# Arrange - Create current data with similar performance
current_data = self.sample_benchmark_data.copy()
# Slightly better performance (5% improvement)
for benchmark in current_data["benchmarks"]:
benchmark["stats"]["mean"] *= 0.95
# Act
comparison_results = self.manager.compare_with_baseline(current_data)
# Assert
self.assertIn("comparison_metadata", comparison_results)
self.assertIn("regression_analysis", comparison_results)
self.assertIn("overall_status", comparison_results)
self.assertEqual(comparison_results["overall_status"], "passed")
# Verify no regressions detected
for analysis in comparison_results["regression_analysis"].values():
self.assertFalse(analysis["is_regression"])
self.assertLess(analysis["percent_change"], 10.0) # Below threshold
@pytest.mark.benchmark
def test_baseline_comparison_with_regression(self):
"""Test baseline comparison when regression is detected"""
# Arrange - Create current data with performance regression
current_data = self.sample_benchmark_data.copy()
# Significant performance degradation (20% slower)
for benchmark in current_data["benchmarks"]:
benchmark["stats"]["mean"] *= 1.20
# Act
comparison_results = self.manager.compare_with_baseline(current_data)
# Assert
self.assertEqual(comparison_results["overall_status"], "regression_detected")
# Verify regressions are properly detected
for analysis in comparison_results["regression_analysis"].values():
self.assertTrue(analysis["is_regression"])
self.assertGreater(analysis["percent_change"], 10.0) # Above threshold
self.assertIn("severity", analysis)
self.assertIn(analysis["severity"], ["minor_regression", "major_regression", "critical_regression"])
@pytest.mark.benchmark
def test_regression_severity_classification(self):
"""Test regression severity classification logic"""
# Arrange - Create data with different levels of regression
test_cases = [
(1.15, "minor_regression"), # 15% slower
(1.25, "major_regression"), # 25% slower
(1.60, "critical_regression") # 60% slower
]
for multiplier, expected_severity in test_cases:
with self.subTest(multiplier=multiplier):
# Arrange
current_data = self.sample_benchmark_data.copy()
for benchmark in current_data["benchmarks"]:
benchmark["stats"]["mean"] *= multiplier
# Act
comparison_results = self.manager.compare_with_baseline(current_data)
# Assert
for analysis in comparison_results["regression_analysis"].values():
self.assertEqual(analysis["severity"], expected_severity)
@pytest.mark.benchmark
def test_statistical_significance_checking(self):
"""Test statistical significance checking in regression analysis"""
# Arrange - Create data with high variance but similar means
current_data = self.sample_benchmark_data.copy()
for benchmark in current_data["benchmarks"]:
benchmark["stats"]["stddev"] *= 10 # High variance
# Act
comparison_results = self.manager.compare_with_baseline(current_data)
# Assert
for analysis in comparison_results["regression_analysis"].values():
self.assertIn("statistical_significance", analysis)
sig_analysis = analysis["statistical_significance"]
self.assertIn("significant", sig_analysis)
self.assertIn("method", sig_analysis)
self.assertIsInstance(sig_analysis["significant"], bool)
@pytest.mark.benchmark
def test_comparison_with_missing_baseline(self):
"""Test comparison behavior when no baseline exists"""
# Arrange - Create manager with empty metrics directory
empty_dir = tempfile.mkdtemp()
try:
empty_manager = BaselineManager(empty_dir)
# Act
comparison_results = empty_manager.compare_with_baseline(self.sample_benchmark_data)
# Assert
self.assertEqual(comparison_results["status"], "no_baseline")
self.assertEqual(comparison_results["action"], "create_baseline")
finally:
shutil.rmtree(empty_dir, ignore_errors=True)
test_baseline_comparison_no_regression(self)
¶
Test baseline comparison when no regression is detected
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
@pytest.mark.performance
def test_baseline_comparison_no_regression(self):
"""Test baseline comparison when no regression is detected"""
# Arrange - Create current data with similar performance
current_data = self.sample_benchmark_data.copy()
# Slightly better performance (5% improvement)
for benchmark in current_data["benchmarks"]:
benchmark["stats"]["mean"] *= 0.95
# Act
comparison_results = self.manager.compare_with_baseline(current_data)
# Assert
self.assertIn("comparison_metadata", comparison_results)
self.assertIn("regression_analysis", comparison_results)
self.assertIn("overall_status", comparison_results)
self.assertEqual(comparison_results["overall_status"], "passed")
# Verify no regressions detected
for analysis in comparison_results["regression_analysis"].values():
self.assertFalse(analysis["is_regression"])
self.assertLess(analysis["percent_change"], 10.0) # Below threshold
test_baseline_comparison_with_regression(self)
¶
Test baseline comparison when regression is detected
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_baseline_comparison_with_regression(self):
"""Test baseline comparison when regression is detected"""
# Arrange - Create current data with performance regression
current_data = self.sample_benchmark_data.copy()
# Significant performance degradation (20% slower)
for benchmark in current_data["benchmarks"]:
benchmark["stats"]["mean"] *= 1.20
# Act
comparison_results = self.manager.compare_with_baseline(current_data)
# Assert
self.assertEqual(comparison_results["overall_status"], "regression_detected")
# Verify regressions are properly detected
for analysis in comparison_results["regression_analysis"].values():
self.assertTrue(analysis["is_regression"])
self.assertGreater(analysis["percent_change"], 10.0) # Above threshold
self.assertIn("severity", analysis)
self.assertIn(analysis["severity"], ["minor_regression", "major_regression", "critical_regression"])
test_regression_severity_classification(self)
¶
Test regression severity classification logic
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_regression_severity_classification(self):
"""Test regression severity classification logic"""
# Arrange - Create data with different levels of regression
test_cases = [
(1.15, "minor_regression"), # 15% slower
(1.25, "major_regression"), # 25% slower
(1.60, "critical_regression") # 60% slower
]
for multiplier, expected_severity in test_cases:
with self.subTest(multiplier=multiplier):
# Arrange
current_data = self.sample_benchmark_data.copy()
for benchmark in current_data["benchmarks"]:
benchmark["stats"]["mean"] *= multiplier
# Act
comparison_results = self.manager.compare_with_baseline(current_data)
# Assert
for analysis in comparison_results["regression_analysis"].values():
self.assertEqual(analysis["severity"], expected_severity)
test_statistical_significance_checking(self)
¶
Test statistical significance checking in regression analysis
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_statistical_significance_checking(self):
"""Test statistical significance checking in regression analysis"""
# Arrange - Create data with high variance but similar means
current_data = self.sample_benchmark_data.copy()
for benchmark in current_data["benchmarks"]:
benchmark["stats"]["stddev"] *= 10 # High variance
# Act
comparison_results = self.manager.compare_with_baseline(current_data)
# Assert
for analysis in comparison_results["regression_analysis"].values():
self.assertIn("statistical_significance", analysis)
sig_analysis = analysis["statistical_significance"]
self.assertIn("significant", sig_analysis)
self.assertIn("method", sig_analysis)
self.assertIsInstance(sig_analysis["significant"], bool)
test_comparison_with_missing_baseline(self)
¶
Test comparison behavior when no baseline exists
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_comparison_with_missing_baseline(self):
"""Test comparison behavior when no baseline exists"""
# Arrange - Create manager with empty metrics directory
empty_dir = tempfile.mkdtemp()
try:
empty_manager = BaselineManager(empty_dir)
# Act
comparison_results = empty_manager.compare_with_baseline(self.sample_benchmark_data)
# Assert
self.assertEqual(comparison_results["status"], "no_baseline")
self.assertEqual(comparison_results["action"], "create_baseline")
finally:
shutil.rmtree(empty_dir, ignore_errors=True)
TestTrendAnalysis (BaselineManagerTest)
¶
Test Task 6.3: Add trend analysis and historical tracking capabilities
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
class TestTrendAnalysis(BaselineManagerTest):
"""Test Task 6.3: Add trend analysis and historical tracking capabilities"""
def setUp(self):
"""Set up test fixtures with historical data"""
super().setUp()
# Create multiple historical baseline files
self.create_historical_baseline_files()
def create_historical_baseline_files(self):
"""Create multiple historical baseline files for trend analysis"""
historical_dir = self.manager.historical_dir
# Create 5 historical baselines with gradual performance degradation
for i in range(5):
timestamp = datetime.now() - timedelta(days=(4-i))
# Create benchmark data with gradual performance degradation
historical_data = self.sample_benchmark_data.copy()
# Simulate performance degradation over time
degradation_factor = 1.0 + (i * 0.02) # 2% degradation per baseline
for benchmark in historical_data["benchmarks"]:
benchmark["stats"]["mean"] *= degradation_factor
# Create baseline structure
baseline_structure = self.manager.create_standardized_baseline_structure(historical_data)
baseline_structure["baseline_metadata"]["created_at"] = timestamp.isoformat()
# Save historical file
filename = f"baseline_{timestamp.strftime('%Y%m%d_%H%M%S')}_test{i:02d}.json"
filepath = historical_dir / filename
with open(filepath, 'w') as f:
json.dump(baseline_structure, f, indent=2)
@pytest.mark.benchmark
@pytest.mark.performance
def test_trend_analysis_with_sufficient_data(self):
"""Test trend analysis with sufficient historical data"""
# Act
trend_results = self.manager.analyze_trends()
# Assert
self.assertNotEqual(trend_results.get("status"), "insufficient_data")
self.assertIn("trend_metadata", trend_results)
self.assertIn("benchmark_trends", trend_results)
self.assertIn("performance_trajectory", trend_results)
self.assertIn("trend_summary", trend_results)
# Verify trend metadata
metadata = trend_results["trend_metadata"]
self.assertIn("analysis_timestamp", metadata)
self.assertIn("analyzed_files", metadata)
self.assertGreater(metadata["analyzed_files"], 1)
# Verify trend summary categories
summary = trend_results["trend_summary"]
required_categories = ["improving_benchmarks", "degrading_benchmarks", "stable_benchmarks", "anomalous_benchmarks"]
for category in required_categories:
self.assertIn(category, summary)
self.assertIsInstance(summary[category], list)
@pytest.mark.benchmark
def test_performance_trajectory_calculation(self):
"""Test overall performance trajectory calculation"""
# Act
trend_results = self.manager.analyze_trends()
# Assert
trajectory = trend_results.get("performance_trajectory", {})
# Verify trajectory contains expected fields
expected_fields = [
"overall_trend",
"overall_change_percent",
"benchmarks_analyzed",
"average_earliest_time",
"average_latest_time",
"performance_health"
]
for field in expected_fields:
self.assertIn(field, trajectory)
# Since we created degrading data, expect degrading trend
self.assertIn(trajectory["overall_trend"], ["degrading", "stable", "improving"])
self.assertGreater(trajectory["benchmarks_analyzed"], 0)
@pytest.mark.benchmark
def test_individual_benchmark_trend_analysis(self):
"""Test trend analysis for individual benchmarks"""
# Act
trend_results = self.manager.analyze_trends()
# Assert
benchmark_trends = trend_results.get("benchmark_trends", {})
# Verify trends are calculated for expected benchmarks
expected_benchmarks = ["test_get_path_benchmark", "test_tiling_benchmark", "test_array_operations_benchmark"]
for benchmark_name in expected_benchmarks:
if benchmark_name in benchmark_trends:
trend_analysis = benchmark_trends[benchmark_name]
# Verify trend analysis structure
required_fields = [
"trend",
"slope",
"data_points",
"latest_value",
"earliest_value",
"total_change_percent",
"volatility"
]
for field in required_fields:
self.assertIn(field, trend_analysis)
# Verify trend classification
self.assertIn(trend_analysis["trend"], ["improving", "degrading", "stable", "insufficient_data"])
self.assertGreater(trend_analysis["data_points"], 0)
@pytest.mark.benchmark
def test_trend_analysis_with_insufficient_data(self):
"""Test trend analysis behavior with insufficient historical data"""
# Arrange - Create manager with minimal historical data
minimal_dir = tempfile.mkdtemp()
try:
minimal_manager = BaselineManager(minimal_dir)
# Create only one historical file
minimal_manager.historical_dir.mkdir(exist_ok=True)
baseline_structure = minimal_manager.create_standardized_baseline_structure(self.sample_benchmark_data)
with open(minimal_manager.historical_dir / "baseline_single.json", 'w') as f:
json.dump(baseline_structure, f, indent=2)
# Act
trend_results = minimal_manager.analyze_trends()
# Assert
self.assertEqual(trend_results["status"], "insufficient_data")
self.assertIn("message", trend_results)
finally:
shutil.rmtree(minimal_dir, ignore_errors=True)
create_historical_baseline_files(self)
¶
Create multiple historical baseline files for trend analysis
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
def create_historical_baseline_files(self):
"""Create multiple historical baseline files for trend analysis"""
historical_dir = self.manager.historical_dir
# Create 5 historical baselines with gradual performance degradation
for i in range(5):
timestamp = datetime.now() - timedelta(days=(4-i))
# Create benchmark data with gradual performance degradation
historical_data = self.sample_benchmark_data.copy()
# Simulate performance degradation over time
degradation_factor = 1.0 + (i * 0.02) # 2% degradation per baseline
for benchmark in historical_data["benchmarks"]:
benchmark["stats"]["mean"] *= degradation_factor
# Create baseline structure
baseline_structure = self.manager.create_standardized_baseline_structure(historical_data)
baseline_structure["baseline_metadata"]["created_at"] = timestamp.isoformat()
# Save historical file
filename = f"baseline_{timestamp.strftime('%Y%m%d_%H%M%S')}_test{i:02d}.json"
filepath = historical_dir / filename
with open(filepath, 'w') as f:
json.dump(baseline_structure, f, indent=2)
test_trend_analysis_with_sufficient_data(self)
¶
Test trend analysis with sufficient historical data
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
@pytest.mark.performance
def test_trend_analysis_with_sufficient_data(self):
"""Test trend analysis with sufficient historical data"""
# Act
trend_results = self.manager.analyze_trends()
# Assert
self.assertNotEqual(trend_results.get("status"), "insufficient_data")
self.assertIn("trend_metadata", trend_results)
self.assertIn("benchmark_trends", trend_results)
self.assertIn("performance_trajectory", trend_results)
self.assertIn("trend_summary", trend_results)
# Verify trend metadata
metadata = trend_results["trend_metadata"]
self.assertIn("analysis_timestamp", metadata)
self.assertIn("analyzed_files", metadata)
self.assertGreater(metadata["analyzed_files"], 1)
# Verify trend summary categories
summary = trend_results["trend_summary"]
required_categories = ["improving_benchmarks", "degrading_benchmarks", "stable_benchmarks", "anomalous_benchmarks"]
for category in required_categories:
self.assertIn(category, summary)
self.assertIsInstance(summary[category], list)
test_performance_trajectory_calculation(self)
¶
Test overall performance trajectory calculation
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_performance_trajectory_calculation(self):
"""Test overall performance trajectory calculation"""
# Act
trend_results = self.manager.analyze_trends()
# Assert
trajectory = trend_results.get("performance_trajectory", {})
# Verify trajectory contains expected fields
expected_fields = [
"overall_trend",
"overall_change_percent",
"benchmarks_analyzed",
"average_earliest_time",
"average_latest_time",
"performance_health"
]
for field in expected_fields:
self.assertIn(field, trajectory)
# Since we created degrading data, expect degrading trend
self.assertIn(trajectory["overall_trend"], ["degrading", "stable", "improving"])
self.assertGreater(trajectory["benchmarks_analyzed"], 0)
test_individual_benchmark_trend_analysis(self)
¶
Test trend analysis for individual benchmarks
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_individual_benchmark_trend_analysis(self):
"""Test trend analysis for individual benchmarks"""
# Act
trend_results = self.manager.analyze_trends()
# Assert
benchmark_trends = trend_results.get("benchmark_trends", {})
# Verify trends are calculated for expected benchmarks
expected_benchmarks = ["test_get_path_benchmark", "test_tiling_benchmark", "test_array_operations_benchmark"]
for benchmark_name in expected_benchmarks:
if benchmark_name in benchmark_trends:
trend_analysis = benchmark_trends[benchmark_name]
# Verify trend analysis structure
required_fields = [
"trend",
"slope",
"data_points",
"latest_value",
"earliest_value",
"total_change_percent",
"volatility"
]
for field in required_fields:
self.assertIn(field, trend_analysis)
# Verify trend classification
self.assertIn(trend_analysis["trend"], ["improving", "degrading", "stable", "insufficient_data"])
self.assertGreater(trend_analysis["data_points"], 0)
test_trend_analysis_with_insufficient_data(self)
¶
Test trend analysis behavior with insufficient historical data
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_trend_analysis_with_insufficient_data(self):
"""Test trend analysis behavior with insufficient historical data"""
# Arrange - Create manager with minimal historical data
minimal_dir = tempfile.mkdtemp()
try:
minimal_manager = BaselineManager(minimal_dir)
# Create only one historical file
minimal_manager.historical_dir.mkdir(exist_ok=True)
baseline_structure = minimal_manager.create_standardized_baseline_structure(self.sample_benchmark_data)
with open(minimal_manager.historical_dir / "baseline_single.json", 'w') as f:
json.dump(baseline_structure, f, indent=2)
# Act
trend_results = minimal_manager.analyze_trends()
# Assert
self.assertEqual(trend_results["status"], "insufficient_data")
self.assertIn("message", trend_results)
finally:
shutil.rmtree(minimal_dir, ignore_errors=True)
TestVersionControlIntegration (BaselineManagerTest)
¶
Test Task 6.4: Set up version control integration for baseline artifacts
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
class TestVersionControlIntegration(BaselineManagerTest):
"""Test Task 6.4: Set up version control integration for baseline artifacts"""
@pytest.mark.benchmark
def test_git_information_extraction(self):
"""Test extraction of git information for version control integration"""
# Act
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
# Assert
git_info = baseline_structure["version_control_info"]
# Verify git information structure
expected_fields = [
"commit_id",
"commit_timestamp",
"branch",
"is_dirty",
"author",
"repository_url"
]
for field in expected_fields:
self.assertIn(field, git_info)
# Verify data types
self.assertIsInstance(git_info["is_dirty"], bool)
self.assertIsInstance(git_info["commit_id"], str)
self.assertIsInstance(git_info["branch"], str)
@pytest.mark.benchmark
def test_baseline_save_with_version_control(self):
"""Test saving baseline with version control integration"""
# Arrange
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
# Act
saved_path = self.manager.save_baseline(baseline_structure)
# Assert
# Verify main baseline file exists
self.assertTrue(os.path.exists(saved_path))
# Verify historical snapshot was created
historical_files = list(self.manager.historical_dir.glob("baseline_*.json"))
self.assertGreater(len(historical_files), 0)
# Verify historical file contains git information
with open(historical_files[0], 'r') as f:
historical_data = json.load(f)
self.assertIn("version_control_info", historical_data)
@pytest.mark.benchmark
def test_historical_file_naming_convention(self):
"""Test historical file naming includes version control information"""
# Arrange
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
# Act
self.manager.save_baseline(baseline_structure)
# Assert
historical_files = list(self.manager.historical_dir.glob("baseline_*.json"))
self.assertGreater(len(historical_files), 0)
# Verify filename format includes timestamp and git hash
filename = historical_files[0].name
self.assertTrue(filename.startswith("baseline_"))
self.assertTrue(filename.endswith(".json"))
# Should contain timestamp and git hash-like pattern
parts = filename[:-5].split("_") # Remove .json extension
self.assertGreaterEqual(len(parts), 3) # baseline, timestamp, git_hash
test_git_information_extraction(self)
¶
Test extraction of git information for version control integration
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_git_information_extraction(self):
"""Test extraction of git information for version control integration"""
# Act
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
# Assert
git_info = baseline_structure["version_control_info"]
# Verify git information structure
expected_fields = [
"commit_id",
"commit_timestamp",
"branch",
"is_dirty",
"author",
"repository_url"
]
for field in expected_fields:
self.assertIn(field, git_info)
# Verify data types
self.assertIsInstance(git_info["is_dirty"], bool)
self.assertIsInstance(git_info["commit_id"], str)
self.assertIsInstance(git_info["branch"], str)
test_baseline_save_with_version_control(self)
¶
Test saving baseline with version control integration
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_baseline_save_with_version_control(self):
"""Test saving baseline with version control integration"""
# Arrange
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
# Act
saved_path = self.manager.save_baseline(baseline_structure)
# Assert
# Verify main baseline file exists
self.assertTrue(os.path.exists(saved_path))
# Verify historical snapshot was created
historical_files = list(self.manager.historical_dir.glob("baseline_*.json"))
self.assertGreater(len(historical_files), 0)
# Verify historical file contains git information
with open(historical_files[0], 'r') as f:
historical_data = json.load(f)
self.assertIn("version_control_info", historical_data)
test_historical_file_naming_convention(self)
¶
Test historical file naming includes version control information
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_historical_file_naming_convention(self):
"""Test historical file naming includes version control information"""
# Arrange
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
# Act
self.manager.save_baseline(baseline_structure)
# Assert
historical_files = list(self.manager.historical_dir.glob("baseline_*.json"))
self.assertGreater(len(historical_files), 0)
# Verify filename format includes timestamp and git hash
filename = historical_files[0].name
self.assertTrue(filename.startswith("baseline_"))
self.assertTrue(filename.endswith(".json"))
# Should contain timestamp and git hash-like pattern
parts = filename[:-5].split("_") # Remove .json extension
self.assertGreaterEqual(len(parts), 3) # baseline, timestamp, git_hash
TestBaselineReporting (BaselineManagerTest)
¶
Test baseline reporting and documentation capabilities
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
class TestBaselineReporting(BaselineManagerTest):
"""Test baseline reporting and documentation capabilities"""
@pytest.mark.benchmark
def test_baseline_report_generation(self):
"""Test generation of human-readable baseline establishment report"""
# Arrange
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
# Act
report = self.manager.generate_baseline_report(baseline_structure)
# Assert
self.assertIsInstance(report, str)
self.assertGreater(len(report), 100) # Should be substantial report
# Verify key sections are present
expected_sections = [
"HAZELBEAN PERFORMANCE BASELINE ESTABLISHMENT REPORT",
"BASELINE STATISTICS",
"QUALITY METRICS",
"BENCHMARK CATEGORIES",
"RECOMMENDATIONS"
]
for section in expected_sections:
self.assertIn(section, report)
# Verify specific data is included
self.assertIn("Total Benchmarks: 3", report)
self.assertIn("Git Commit:", report)
self.assertIn("Baseline Quality Score:", report)
test_baseline_report_generation(self)
¶
Test generation of human-readable baseline establishment report
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_baseline_report_generation(self):
"""Test generation of human-readable baseline establishment report"""
# Arrange
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
# Act
report = self.manager.generate_baseline_report(baseline_structure)
# Assert
self.assertIsInstance(report, str)
self.assertGreater(len(report), 100) # Should be substantial report
# Verify key sections are present
expected_sections = [
"HAZELBEAN PERFORMANCE BASELINE ESTABLISHMENT REPORT",
"BASELINE STATISTICS",
"QUALITY METRICS",
"BENCHMARK CATEGORIES",
"RECOMMENDATIONS"
]
for section in expected_sections:
self.assertIn(section, report)
# Verify specific data is included
self.assertIn("Total Benchmarks: 3", report)
self.assertIn("Git Commit:", report)
self.assertIn("Baseline Quality Score:", report)
TestBaselineManagerIntegration (BaselineManagerTest)
¶
Integration tests for complete baseline management workflow
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
class TestBaselineManagerIntegration(BaselineManagerTest):
"""Integration tests for complete baseline management workflow"""
@pytest.mark.benchmark
@pytest.mark.integration
def test_complete_baseline_workflow(self):
"""Test complete baseline establishment and comparison workflow"""
# Step 1: Create initial baseline
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
saved_path = self.manager.save_baseline(baseline_structure)
# Verify baseline was created
self.assertTrue(os.path.exists(saved_path))
# Step 2: Create modified data for comparison
modified_data = self.sample_benchmark_data.copy()
# Minor degradation that should not trigger regression
for benchmark in modified_data["benchmarks"]:
benchmark["stats"]["mean"] *= 1.05 # 5% slower
# Step 3: Compare with baseline
comparison_results = self.manager.compare_with_baseline(modified_data)
# Should not detect regression (5% < 10% threshold)
self.assertEqual(comparison_results["overall_status"], "passed")
# Step 4: Create major degradation
degraded_data = self.sample_benchmark_data.copy()
for benchmark in degraded_data["benchmarks"]:
benchmark["stats"]["mean"] *= 1.15 # 15% slower
# Step 5: Compare degraded data
degraded_comparison = self.manager.compare_with_baseline(degraded_data)
# Should detect regression (15% > 10% threshold)
self.assertEqual(degraded_comparison["overall_status"], "regression_detected")
# Step 6: Generate reports
report = self.manager.generate_baseline_report(baseline_structure)
self.assertIsInstance(report, str)
self.assertGreater(len(report), 100)
@pytest.mark.benchmark
def test_error_handling_and_edge_cases(self):
"""Test error handling for various edge cases"""
# Test with empty benchmark data
empty_data = {"benchmarks": []}
baseline_structure = self.manager.create_standardized_baseline_structure(empty_data)
# Should handle gracefully
self.assertIn("baseline_statistics", baseline_structure)
# Should have error in baseline statistics
self.assertIn("error", baseline_structure["baseline_statistics"])
# Test with invalid benchmark data
invalid_data = {
"benchmarks": [
{"name": "invalid_benchmark", "stats": {"mean": "invalid"}}
]
}
# Should not crash and handle gracefully
baseline_structure = self.manager.create_standardized_baseline_structure(invalid_data)
self.assertIsInstance(baseline_structure, dict)
# Should have valid baseline structure but with error in statistics
self.assertIn("baseline_statistics", baseline_structure)
self.assertIn("error", baseline_structure["baseline_statistics"])
# Validation should show 0 valid benchmarks
self.assertEqual(baseline_structure["validation_info"]["valid_benchmarks"], 0)
# Test with mixed valid and invalid data
mixed_data = {
"benchmarks": [
{"name": "valid_benchmark", "stats": {"mean": 0.05, "rounds": 10}},
{"name": "invalid_benchmark", "stats": {"mean": "invalid"}},
{"name": "another_valid", "stats": {"mean": 0.03, "rounds": 5}}
]
}
baseline_structure = self.manager.create_standardized_baseline_structure(mixed_data)
# Should process valid benchmarks and skip invalid ones
self.assertEqual(baseline_structure["validation_info"]["total_benchmarks"], 3)
self.assertEqual(baseline_structure["validation_info"]["valid_benchmarks"], 2)
# Should have valid statistics from the valid benchmarks
stats = baseline_structure["baseline_statistics"]
self.assertNotIn("error", stats)
self.assertIn("aggregate_statistics", stats)
test_complete_baseline_workflow(self)
¶
Test complete baseline establishment and comparison workflow
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
@pytest.mark.integration
def test_complete_baseline_workflow(self):
"""Test complete baseline establishment and comparison workflow"""
# Step 1: Create initial baseline
baseline_structure = self.manager.create_standardized_baseline_structure(self.sample_benchmark_data)
saved_path = self.manager.save_baseline(baseline_structure)
# Verify baseline was created
self.assertTrue(os.path.exists(saved_path))
# Step 2: Create modified data for comparison
modified_data = self.sample_benchmark_data.copy()
# Minor degradation that should not trigger regression
for benchmark in modified_data["benchmarks"]:
benchmark["stats"]["mean"] *= 1.05 # 5% slower
# Step 3: Compare with baseline
comparison_results = self.manager.compare_with_baseline(modified_data)
# Should not detect regression (5% < 10% threshold)
self.assertEqual(comparison_results["overall_status"], "passed")
# Step 4: Create major degradation
degraded_data = self.sample_benchmark_data.copy()
for benchmark in degraded_data["benchmarks"]:
benchmark["stats"]["mean"] *= 1.15 # 15% slower
# Step 5: Compare degraded data
degraded_comparison = self.manager.compare_with_baseline(degraded_data)
# Should detect regression (15% > 10% threshold)
self.assertEqual(degraded_comparison["overall_status"], "regression_detected")
# Step 6: Generate reports
report = self.manager.generate_baseline_report(baseline_structure)
self.assertIsInstance(report, str)
self.assertGreater(len(report), 100)
test_error_handling_and_edge_cases(self)
¶
Test error handling for various edge cases
Source code in hazelbean_tests/performance/unit/test_baseline_manager.py
@pytest.mark.benchmark
def test_error_handling_and_edge_cases(self):
"""Test error handling for various edge cases"""
# Test with empty benchmark data
empty_data = {"benchmarks": []}
baseline_structure = self.manager.create_standardized_baseline_structure(empty_data)
# Should handle gracefully
self.assertIn("baseline_statistics", baseline_structure)
# Should have error in baseline statistics
self.assertIn("error", baseline_structure["baseline_statistics"])
# Test with invalid benchmark data
invalid_data = {
"benchmarks": [
{"name": "invalid_benchmark", "stats": {"mean": "invalid"}}
]
}
# Should not crash and handle gracefully
baseline_structure = self.manager.create_standardized_baseline_structure(invalid_data)
self.assertIsInstance(baseline_structure, dict)
# Should have valid baseline structure but with error in statistics
self.assertIn("baseline_statistics", baseline_structure)
self.assertIn("error", baseline_structure["baseline_statistics"])
# Validation should show 0 valid benchmarks
self.assertEqual(baseline_structure["validation_info"]["valid_benchmarks"], 0)
# Test with mixed valid and invalid data
mixed_data = {
"benchmarks": [
{"name": "valid_benchmark", "stats": {"mean": 0.05, "rounds": 10}},
{"name": "invalid_benchmark", "stats": {"mean": "invalid"}},
{"name": "another_valid", "stats": {"mean": 0.03, "rounds": 5}}
]
}
baseline_structure = self.manager.create_standardized_baseline_structure(mixed_data)
# Should process valid benchmarks and skip invalid ones
self.assertEqual(baseline_structure["validation_info"]["total_benchmarks"], 3)
self.assertEqual(baseline_structure["validation_info"]["valid_benchmarks"], 2)
# Should have valid statistics from the valid benchmarks
stats = baseline_structure["baseline_statistics"]
self.assertNotIn("error", stats)
self.assertIn("aggregate_statistics", stats)
Performance Baseline Manager¶
The baseline management system helps track and validate performance changes.
Baseline Management System for Hazelbean Performance Benchmarks
This module provides a comprehensive baseline management system for storing, comparing, and tracking performance benchmarks over time.
Story 6: Baseline Establishment - All Tasks (6.1-6.4) Test Quality Standards: Baseline management must provide reliable regression detection
logger
¶
BaselineManager
¶
Manages performance baselines with comprehensive JSON structure, regression detection, trend analysis, and version control integration.
Implements Story 6 requirements: - 6.1: Standardized baseline JSON structure - 6.2: Baseline comparison logic for regression detection - 6.3: Trend analysis and historical tracking - 6.4: Version control integration
Source code in hazelbean_tests/performance/baseline_manager.py
class BaselineManager:
"""
Manages performance baselines with comprehensive JSON structure,
regression detection, trend analysis, and version control integration.
Implements Story 6 requirements:
- 6.1: Standardized baseline JSON structure
- 6.2: Baseline comparison logic for regression detection
- 6.3: Trend analysis and historical tracking
- 6.4: Version control integration
"""
def __init__(self, metrics_dir: str = None):
"""Initialize baseline manager with metrics directory"""
if metrics_dir is None:
# Default to project metrics directory
self.metrics_dir = Path(__file__).parent.parent.parent / "metrics"
else:
self.metrics_dir = Path(metrics_dir)
self.metrics_dir.mkdir(exist_ok=True)
# Create organized directory structure
self.baselines_dir = self.metrics_dir / "baselines"
self.snapshots_dir = self.baselines_dir / "snapshots"
self.benchmarks_dir = self.metrics_dir / "benchmarks"
# Create directories
self.baselines_dir.mkdir(exist_ok=True)
self.snapshots_dir.mkdir(exist_ok=True)
self.benchmarks_dir.mkdir(exist_ok=True)
# Main baseline file location
self.baseline_file = self.baselines_dir / "current_performance_baseline.json"
# Keep backward compatibility with historical_dir for existing code
self.historical_dir = self.snapshots_dir
logger.info(f"BaselineManager initialized with metrics_dir: {self.metrics_dir}")
def create_standardized_baseline_structure(self,
benchmark_data: Dict[str, Any],
baseline_version: str = "2.0") -> Dict[str, Any]:
"""
Create standardized baseline JSON structure for all benchmark metrics.
Task 6.1: Create baseline JSON structure for all benchmark metrics
Args:
benchmark_data: Raw benchmark data from pytest-benchmark
baseline_version: Version identifier for the baseline format
Returns:
Standardized baseline structure with comprehensive metadata
"""
# Extract git information for version control integration
git_info = self._get_git_information()
# Calculate comprehensive statistics
baseline_stats = self._calculate_baseline_statistics(benchmark_data)
# Create standardized structure
baseline_structure = {
"baseline_metadata": {
"version": baseline_version,
"created_at": datetime.now(timezone.utc).isoformat(),
"schema_version": "2.0.0",
"description": "Hazelbean performance baseline with comprehensive metrics",
"statistical_confidence": "95%",
"regression_threshold_percent": 10.0,
"trend_analysis_enabled": True
},
"version_control_info": git_info,
"system_environment": self._extract_system_info(benchmark_data),
"baseline_statistics": baseline_stats,
"benchmark_categories": self._categorize_benchmarks(benchmark_data),
"quality_metrics": self._calculate_quality_metrics(baseline_stats),
"raw_benchmark_data": benchmark_data.get("benchmarks", []),
"validation_info": {
"total_benchmarks": len(benchmark_data.get("benchmarks", [])),
"valid_benchmarks": len([b for b in benchmark_data.get("benchmarks", []) if self._is_valid_benchmark(b)]),
"statistical_confidence_met": True,
"baseline_establishment_criteria": {
"minimum_runs": 5,
"maximum_variance_threshold": 0.25,
"outlier_detection_enabled": True
}
}
}
logger.info(f"Created standardized baseline structure with {baseline_structure['validation_info']['total_benchmarks']} benchmarks")
return baseline_structure
def save_baseline(self, baseline_data: Dict[str, Any]) -> str:
"""
Save baseline data with version control integration.
Task 6.4: Set up version control integration for baseline artifacts
Args:
baseline_data: Standardized baseline data structure
Returns:
Path to saved baseline file
"""
# Save main baseline file
with open(self.baseline_file, 'w') as f:
json.dump(baseline_data, f, indent=2)
# Create baseline snapshot
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
git_hash = baseline_data.get("version_control_info", {}).get("commit_id", "unknown")[:8]
# Save snapshot with descriptive naming
snapshot_file = self.snapshots_dir / f"baseline_snapshot_{timestamp}_{git_hash}.json"
with open(snapshot_file, 'w') as f:
json.dump(baseline_data, f, indent=2)
logger.info(f"Saved current baseline to {self.baseline_file} and snapshot to {snapshot_file}")
return str(self.baseline_file)
def compare_with_baseline(self, current_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Implement baseline comparison logic for performance regression detection.
Task 6.2: Implement baseline comparison logic for performance regression detection
Args:
current_data: Current benchmark results to compare against baseline
Returns:
Comprehensive comparison results with regression detection
"""
if not self.baseline_file.exists():
logger.warning("No baseline file found. Creating initial baseline.")
return {"status": "no_baseline", "action": "create_baseline"}
# Load baseline data
with open(self.baseline_file, 'r') as f:
baseline_data = json.load(f)
comparison_results = {
"comparison_metadata": {
"comparison_timestamp": datetime.now(timezone.utc).isoformat(),
"baseline_version": baseline_data.get("baseline_metadata", {}).get("version", "unknown"),
"comparison_type": "performance_regression_analysis"
},
"regression_analysis": {},
"performance_changes": {},
"statistical_significance": {},
"recommendations": [],
"overall_status": "passed"
}
# Compare each benchmark
baseline_benchmarks = {b["name"]: b for b in baseline_data.get("raw_benchmark_data", [])}
current_benchmarks = {b["name"]: b for b in current_data.get("benchmarks", [])}
regression_threshold = baseline_data.get("baseline_metadata", {}).get("regression_threshold_percent", 10.0)
for benchmark_name, current_benchmark in current_benchmarks.items():
if benchmark_name in baseline_benchmarks:
baseline_benchmark = baseline_benchmarks[benchmark_name]
regression_analysis = self._analyze_regression(
baseline_benchmark, current_benchmark, regression_threshold
)
comparison_results["regression_analysis"][benchmark_name] = regression_analysis
if regression_analysis["is_regression"]:
comparison_results["overall_status"] = "regression_detected"
# Add recommendations
comparison_results["recommendations"] = self._generate_recommendations(comparison_results)
logger.info(f"Completed baseline comparison. Status: {comparison_results['overall_status']}")
return comparison_results
def analyze_trends(self, lookback_days: int = 30) -> Dict[str, Any]:
"""
Add trend analysis and historical tracking capabilities.
Task 6.3: Add trend analysis and historical tracking capabilities
Args:
lookback_days: Number of days to analyze for trends
Returns:
Comprehensive trend analysis with historical tracking
"""
# Collect baseline snapshot files
historical_files = list(self.snapshots_dir.glob("baseline_snapshot_*.json"))
historical_files.sort() # Sort by filename (which includes timestamp)
if len(historical_files) < 2:
return {
"status": "insufficient_data",
"message": f"Need at least 2 historical baselines for trend analysis. Found: {len(historical_files)}"
}
trend_data = {
"trend_metadata": {
"analysis_timestamp": datetime.now(timezone.utc).isoformat(),
"lookback_days": lookback_days,
"analyzed_files": len(historical_files),
"trend_detection_algorithm": "linear_regression_with_statistical_significance"
},
"benchmark_trends": {},
"performance_trajectory": {},
"anomaly_detection": {},
"trend_summary": {
"improving_benchmarks": [],
"degrading_benchmarks": [],
"stable_benchmarks": [],
"anomalous_benchmarks": []
}
}
# Analyze trends for each benchmark
benchmark_history = self._collect_benchmark_history(historical_files)
for benchmark_name, history in benchmark_history.items():
if len(history) >= 3: # Need minimum data points for trend analysis
trend_analysis = self._calculate_trend_metrics(benchmark_name, history)
trend_data["benchmark_trends"][benchmark_name] = trend_analysis
# Categorize benchmark trends
self._categorize_benchmark_trend(benchmark_name, trend_analysis, trend_data["trend_summary"])
# Generate performance trajectory summary
trend_data["performance_trajectory"] = self._generate_performance_trajectory(benchmark_history)
logger.info(f"Completed trend analysis for {len(benchmark_history)} benchmarks over {len(historical_files)} baseline files")
return trend_data
def _get_git_information(self) -> Dict[str, Any]:
"""Extract git information for version control integration"""
try:
git_info = {
"commit_id": subprocess.check_output(["git", "rev-parse", "HEAD"]).decode().strip(),
"commit_timestamp": subprocess.check_output(["git", "show", "-s", "--format=%ci", "HEAD"]).decode().strip(),
"branch": subprocess.check_output(["git", "rev-parse", "--abbrev-ref", "HEAD"]).decode().strip(),
"is_dirty": len(subprocess.check_output(["git", "status", "--porcelain"]).decode().strip()) > 0,
"author": subprocess.check_output(["git", "show", "-s", "--format=%an", "HEAD"]).decode().strip(),
"repository_url": self._get_repository_url()
}
except subprocess.CalledProcessError:
git_info = {
"commit_id": "unknown",
"commit_timestamp": "unknown",
"branch": "unknown",
"is_dirty": False,
"author": "unknown",
"repository_url": "unknown"
}
return git_info
def _get_repository_url(self) -> str:
"""Get repository URL for version control tracking"""
try:
origin_url = subprocess.check_output(["git", "config", "--get", "remote.origin.url"]).decode().strip()
return origin_url
except subprocess.CalledProcessError:
return "unknown"
def _extract_system_info(self, benchmark_data: Dict[str, Any]) -> Dict[str, Any]:
"""Extract and standardize system environment information"""
machine_info = benchmark_data.get("machine_info", {})
return {
"platform": {
"system": machine_info.get("system", "unknown"),
"node": machine_info.get("node", "unknown"),
"release": machine_info.get("release", "unknown"),
"machine": machine_info.get("machine", "unknown"),
"processor": machine_info.get("processor", "unknown")
},
"python_environment": {
"version": machine_info.get("python_version", "unknown"),
"implementation": machine_info.get("python_implementation", "unknown"),
"compiler": machine_info.get("python_compiler", "unknown")
},
"cpu_info": machine_info.get("cpu", {}),
"environment_hash": self._calculate_environment_hash(machine_info)
}
def _calculate_environment_hash(self, machine_info: Dict[str, Any]) -> str:
"""Calculate hash of environment for compatibility checking"""
import hashlib
env_string = f"{machine_info.get('system')}_{machine_info.get('machine')}_{machine_info.get('python_version')}"
return hashlib.sha256(env_string.encode()).hexdigest()[:16]
def _calculate_baseline_statistics(self, benchmark_data: Dict[str, Any]) -> Dict[str, Any]:
"""Calculate comprehensive baseline statistics"""
benchmarks = benchmark_data.get("benchmarks", [])
if not benchmarks:
return {"error": "no_benchmark_data"}
execution_times = []
for benchmark in benchmarks:
stats = benchmark.get("stats", {})
if "mean" in stats:
try:
mean_value = float(stats["mean"])
if mean_value > 0: # Only include positive values
execution_times.append(mean_value)
except (ValueError, TypeError):
# Skip invalid values
logger.warning(f"Invalid mean value in benchmark {benchmark.get('name', 'unknown')}: {stats['mean']}")
continue
if not execution_times:
return {"error": "no_valid_execution_times"}
return {
"aggregate_statistics": {
"mean_execution_time": statistics.mean(execution_times),
"median_execution_time": statistics.median(execution_times),
"std_deviation": statistics.stdev(execution_times) if len(execution_times) > 1 else 0,
"min_time": min(execution_times),
"max_time": max(execution_times),
"total_benchmarks": len(execution_times)
},
"confidence_intervals": self._calculate_confidence_intervals(execution_times),
"outlier_analysis": self._detect_outliers(execution_times),
"quality_indicators": {
"coefficient_of_variation": (statistics.stdev(execution_times) / statistics.mean(execution_times)) if len(execution_times) > 1 and statistics.mean(execution_times) > 0 else 0,
"acceptable_variance": (statistics.stdev(execution_times) / statistics.mean(execution_times)) < 0.25 if len(execution_times) > 1 and statistics.mean(execution_times) > 0 else False
}
}
def _calculate_confidence_intervals(self, data: List[float], confidence: float = 0.95) -> Dict[str, float]:
"""Calculate confidence intervals for baseline statistics"""
if len(data) < 2:
return {"lower": 0, "upper": 0, "confidence_level": confidence}
mean = statistics.mean(data)
std_dev = statistics.stdev(data) if len(data) > 1 else 0
n = len(data)
# Using t-distribution for small samples
import math
if n < 30:
# Simplified t-distribution approximation
t_value = 2.0 # Approximate t-value for 95% confidence
else:
t_value = 1.96 # z-value for 95% confidence
margin_error = t_value * (std_dev / math.sqrt(n))
return {
"lower": mean - margin_error,
"upper": mean + margin_error,
"confidence_level": confidence,
"margin_of_error": margin_error
}
def _detect_outliers(self, data: List[float]) -> Dict[str, Any]:
"""Detect outliers in benchmark data using IQR method"""
if len(data) < 4:
return {"outliers": [], "method": "insufficient_data"}
sorted_data = sorted(data)
n = len(sorted_data)
q1_index = n // 4
q3_index = 3 * n // 4
q1 = sorted_data[q1_index]
q3 = sorted_data[q3_index]
iqr = q3 - q1
lower_bound = q1 - 1.5 * iqr
upper_bound = q3 + 1.5 * iqr
outliers = [x for x in data if x < lower_bound or x > upper_bound]
return {
"outliers": outliers,
"outlier_count": len(outliers),
"outlier_percentage": (len(outliers) / len(data)) * 100,
"method": "IQR",
"bounds": {"lower": lower_bound, "upper": upper_bound},
"quartiles": {"q1": q1, "q3": q3, "iqr": iqr}
}
def _categorize_benchmarks(self, benchmark_data: Dict[str, Any]) -> Dict[str, List[str]]:
"""Categorize benchmarks by type and functionality"""
categories = {
"path_resolution": [],
"tiling_operations": [],
"data_processing": [],
"io_operations": [],
"computational": [],
"integration": [],
"uncategorized": []
}
for benchmark in benchmark_data.get("benchmarks", []):
name = benchmark.get("name", "").lower()
categorized = False
if any(keyword in name for keyword in ["path", "get_path", "resolution"]):
categories["path_resolution"].append(benchmark["name"])
categorized = True
elif any(keyword in name for keyword in ["tile", "tiling", "iterator"]):
categories["tiling_operations"].append(benchmark["name"])
categorized = True
elif any(keyword in name for keyword in ["array", "processing", "calculation"]):
categories["data_processing"].append(benchmark["name"])
categorized = True
elif any(keyword in name for keyword in ["io", "read", "write", "load", "save"]):
categories["io_operations"].append(benchmark["name"])
categorized = True
elif any(keyword in name for keyword in ["integration", "workflow", "end_to_end"]):
categories["integration"].append(benchmark["name"])
categorized = True
elif any(keyword in name for keyword in ["compute", "algorithm", "math"]):
categories["computational"].append(benchmark["name"])
categorized = True
if not categorized:
categories["uncategorized"].append(benchmark["name"])
return categories
def _calculate_quality_metrics(self, baseline_stats: Dict[str, Any]) -> Dict[str, Any]:
"""Calculate quality metrics for baseline establishment"""
aggregate_stats = baseline_stats.get("aggregate_statistics", {})
return {
"baseline_quality_score": self._calculate_quality_score(baseline_stats),
"statistical_reliability": {
"sufficient_sample_size": aggregate_stats.get("total_benchmarks", 0) >= 5,
"acceptable_variance": baseline_stats.get("quality_indicators", {}).get("acceptable_variance", False),
"outlier_percentage": baseline_stats.get("outlier_analysis", {}).get("outlier_percentage", 0),
"confidence_interval_width": self._calculate_ci_width(baseline_stats)
},
"recommendations": self._generate_quality_recommendations(baseline_stats)
}
def _calculate_quality_score(self, baseline_stats: Dict[str, Any]) -> float:
"""Calculate overall quality score for baseline (0-100)"""
score = 100.0
# Penalize high variance
cv = baseline_stats.get("quality_indicators", {}).get("coefficient_of_variation", 0)
if cv > 0.25:
score -= min(30, cv * 100)
# Penalize high outlier percentage
outlier_pct = baseline_stats.get("outlier_analysis", {}).get("outlier_percentage", 0)
if outlier_pct > 10:
score -= min(20, outlier_pct)
# Penalize small sample size
sample_size = baseline_stats.get("aggregate_statistics", {}).get("total_benchmarks", 0)
if sample_size < 5:
score -= 40
elif sample_size < 10:
score -= 20
return max(0, score)
def _calculate_ci_width(self, baseline_stats: Dict[str, Any]) -> float:
"""Calculate confidence interval width as percentage of mean"""
ci = baseline_stats.get("confidence_intervals", {})
mean = baseline_stats.get("aggregate_statistics", {}).get("mean_execution_time", 0)
if mean > 0 and "upper" in ci and "lower" in ci:
width = ci["upper"] - ci["lower"]
return (width / mean) * 100
return 0
def _generate_quality_recommendations(self, baseline_stats: Dict[str, Any]) -> List[str]:
"""Generate recommendations for improving baseline quality"""
recommendations = []
quality_indicators = baseline_stats.get("quality_indicators", {})
if not quality_indicators.get("acceptable_variance", True):
recommendations.append("High variance detected. Consider running more benchmark iterations or investigating environmental factors.")
outlier_pct = baseline_stats.get("outlier_analysis", {}).get("outlier_percentage", 0)
if outlier_pct > 15:
recommendations.append("High outlier percentage. Review benchmark setup and consider environment stabilization.")
sample_size = baseline_stats.get("aggregate_statistics", {}).get("total_benchmarks", 0)
if sample_size < 10:
recommendations.append("Small sample size. Consider adding more benchmark tests for better statistical reliability.")
return recommendations
def _is_valid_benchmark(self, benchmark: Dict[str, Any]) -> bool:
"""Check if a benchmark result is valid for baseline inclusion"""
stats = benchmark.get("stats", {})
# Check if mean exists and can be converted to float
if "mean" not in stats:
return False
try:
mean_value = float(stats["mean"])
if mean_value <= 0:
return False
except (ValueError, TypeError):
return False
# Check rounds if present
try:
rounds = stats.get("rounds", 1)
if isinstance(rounds, str):
rounds = float(rounds)
if rounds <= 0:
return False
except (ValueError, TypeError):
return False
return True
def _analyze_regression(self, baseline_benchmark: Dict[str, Any],
current_benchmark: Dict[str, Any],
threshold_percent: float) -> Dict[str, Any]:
"""Analyze potential regression between baseline and current benchmark"""
baseline_mean = baseline_benchmark.get("stats", {}).get("mean", 0)
current_mean = current_benchmark.get("stats", {}).get("mean", 0)
if baseline_mean == 0:
return {"is_regression": False, "reason": "invalid_baseline_data"}
percent_change = ((current_mean - baseline_mean) / baseline_mean) * 100
is_regression = percent_change > threshold_percent
return {
"is_regression": is_regression,
"percent_change": percent_change,
"baseline_mean": baseline_mean,
"current_mean": current_mean,
"absolute_difference": current_mean - baseline_mean,
"threshold_percent": threshold_percent,
"severity": self._classify_regression_severity(percent_change, threshold_percent),
"statistical_significance": self._check_statistical_significance(baseline_benchmark, current_benchmark)
}
def _classify_regression_severity(self, percent_change: float, threshold: float) -> str:
"""Classify regression severity based on performance change"""
if percent_change <= threshold:
return "no_regression"
elif percent_change <= threshold * 2:
return "minor_regression"
elif percent_change <= threshold * 5:
return "major_regression"
else:
return "critical_regression"
def _check_statistical_significance(self, baseline: Dict[str, Any], current: Dict[str, Any]) -> Dict[str, Any]:
"""Check statistical significance of performance difference"""
baseline_stats = baseline.get("stats", {})
current_stats = current.get("stats", {})
# Simplified significance check using standard deviation
baseline_std = baseline_stats.get("stddev", 0)
current_std = current_stats.get("stddev", 0)
baseline_mean = baseline_stats.get("mean", 0)
current_mean = current_stats.get("mean", 0)
if baseline_std == 0 or current_std == 0:
return {"significant": False, "method": "insufficient_variance_data"}
# Simple two-standard-deviation test
combined_std = (baseline_std + current_std) / 2
difference = abs(current_mean - baseline_mean)
is_significant = difference > (2 * combined_std)
return {
"significant": is_significant,
"method": "two_standard_deviation_test",
"difference": difference,
"threshold": 2 * combined_std,
"confidence_level": "approximately_95_percent"
}
def _generate_recommendations(self, comparison_results: Dict[str, Any]) -> List[str]:
"""Generate recommendations based on comparison results"""
recommendations = []
if comparison_results["overall_status"] == "regression_detected":
recommendations.append("Performance regression detected. Review recent changes and consider performance optimization.")
# Count regressions by severity
severe_regressions = sum(1 for analysis in comparison_results["regression_analysis"].values()
if analysis.get("severity") in ["major_regression", "critical_regression"])
if severe_regressions > 0:
recommendations.append(f"Critical performance regressions found in {severe_regressions} benchmark(s). Immediate attention required.")
return recommendations
def _collect_benchmark_history(self, historical_files: List[Path]) -> Dict[str, List[Dict[str, Any]]]:
"""Collect benchmark history from historical baseline files"""
benchmark_history = {}
for file_path in historical_files:
try:
with open(file_path, 'r') as f:
historical_data = json.load(f)
# Extract timestamp from metadata or filename
timestamp = historical_data.get("baseline_metadata", {}).get("created_at")
if not timestamp:
# Extract from filename if not in metadata
timestamp = file_path.stem.split("_")[1] if "_" in file_path.stem else "unknown"
for benchmark in historical_data.get("raw_benchmark_data", []):
name = benchmark.get("name")
if name:
if name not in benchmark_history:
benchmark_history[name] = []
benchmark_history[name].append({
"timestamp": timestamp,
"stats": benchmark.get("stats", {}),
"file_source": str(file_path)
})
except (json.JSONDecodeError, FileNotFoundError) as e:
logger.warning(f"Could not process historical file {file_path}: {e}")
# Sort each benchmark's history by timestamp
for name in benchmark_history:
benchmark_history[name].sort(key=lambda x: x["timestamp"])
return benchmark_history
def _calculate_trend_metrics(self, benchmark_name: str, history: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Calculate trend metrics for a specific benchmark"""
execution_times = [entry["stats"].get("mean", 0) for entry in history if entry["stats"].get("mean", 0) > 0]
if len(execution_times) < 3:
return {"trend": "insufficient_data", "data_points": len(execution_times)}
# Simple linear trend calculation
x_values = list(range(len(execution_times)))
trend_slope = self._calculate_linear_trend(x_values, execution_times)
return {
"trend": "improving" if trend_slope < -0.001 else "degrading" if trend_slope > 0.001 else "stable",
"slope": trend_slope,
"data_points": len(execution_times),
"latest_value": execution_times[-1],
"earliest_value": execution_times[0],
"total_change_percent": ((execution_times[-1] - execution_times[0]) / execution_times[0]) * 100 if execution_times[0] > 0 else 0,
"volatility": statistics.stdev(execution_times) if len(execution_times) > 1 else 0
}
def _calculate_linear_trend(self, x_values: List[int], y_values: List[float]) -> float:
"""Calculate linear trend slope using least squares"""
n = len(x_values)
if n < 2:
return 0
sum_x = sum(x_values)
sum_y = sum(y_values)
sum_xy = sum(x * y for x, y in zip(x_values, y_values))
sum_x2 = sum(x * x for x in x_values)
denominator = n * sum_x2 - sum_x * sum_x
if denominator == 0:
return 0
slope = (n * sum_xy - sum_x * sum_y) / denominator
return slope
def _categorize_benchmark_trend(self, benchmark_name: str, trend_analysis: Dict[str, Any], trend_summary: Dict[str, List]):
"""Categorize benchmark into trend summary categories"""
trend = trend_analysis.get("trend", "unknown")
if trend == "improving":
trend_summary["improving_benchmarks"].append(benchmark_name)
elif trend == "degrading":
trend_summary["degrading_benchmarks"].append(benchmark_name)
elif trend == "stable":
trend_summary["stable_benchmarks"].append(benchmark_name)
else:
trend_summary["anomalous_benchmarks"].append(benchmark_name)
def _generate_performance_trajectory(self, benchmark_history: Dict[str, List[Dict[str, Any]]]) -> Dict[str, Any]:
"""Generate overall performance trajectory summary"""
if not benchmark_history:
return {"status": "no_data"}
# Calculate overall performance trend
all_latest_times = []
all_earliest_times = []
for benchmark_data in benchmark_history.values():
if len(benchmark_data) >= 2:
earliest = benchmark_data[0]["stats"].get("mean", 0)
latest = benchmark_data[-1]["stats"].get("mean", 0)
if earliest > 0 and latest > 0:
all_earliest_times.append(earliest)
all_latest_times.append(latest)
if not all_latest_times or not all_earliest_times:
return {"status": "insufficient_data"}
avg_earliest = statistics.mean(all_earliest_times)
avg_latest = statistics.mean(all_latest_times)
overall_change = ((avg_latest - avg_earliest) / avg_earliest) * 100 if avg_earliest > 0 else 0
return {
"overall_trend": "improving" if overall_change < -1 else "degrading" if overall_change > 1 else "stable",
"overall_change_percent": overall_change,
"benchmarks_analyzed": len(all_latest_times),
"average_earliest_time": avg_earliest,
"average_latest_time": avg_latest,
"performance_health": "good" if overall_change < 5 else "concerning" if overall_change < 15 else "poor"
}
def generate_baseline_report(self, baseline_data: Dict[str, Any]) -> str:
"""Generate a human-readable baseline establishment report"""
report_lines = [
"HAZELBEAN PERFORMANCE BASELINE ESTABLISHMENT REPORT",
"=" * 55,
"",
f"Baseline Version: {baseline_data.get('baseline_metadata', {}).get('version', 'unknown')}",
f"Created: {baseline_data.get('baseline_metadata', {}).get('created_at', 'unknown')}",
f"Git Commit: {baseline_data.get('version_control_info', {}).get('commit_id', 'unknown')[:12]}",
f"Branch: {baseline_data.get('version_control_info', {}).get('branch', 'unknown')}",
"",
"BASELINE STATISTICS",
"-" * 20,
]
stats = baseline_data.get("baseline_statistics", {}).get("aggregate_statistics", {})
report_lines.extend([
f"Total Benchmarks: {stats.get('total_benchmarks', 0)}",
f"Mean Execution Time: {stats.get('mean_execution_time', 0):.6f}s",
f"Standard Deviation: {stats.get('std_deviation', 0):.6f}s",
f"Min Time: {stats.get('min_time', 0):.6f}s",
f"Max Time: {stats.get('max_time', 0):.6f}s",
"",
"QUALITY METRICS",
"-" * 15,
])
quality = baseline_data.get("quality_metrics", {})
report_lines.extend([
f"Baseline Quality Score: {quality.get('baseline_quality_score', 0):.1f}/100",
f"Statistical Reliability: {'PASS' if quality.get('statistical_reliability', {}).get('sufficient_sample_size', False) else 'FAIL'}",
f"Variance Acceptable: {'YES' if quality.get('statistical_reliability', {}).get('acceptable_variance', False) else 'NO'}",
"",
"BENCHMARK CATEGORIES",
"-" * 20,
])
categories = baseline_data.get("benchmark_categories", {})
for category, benchmarks in categories.items():
if benchmarks:
report_lines.append(f"{category.replace('_', ' ').title()}: {len(benchmarks)} benchmarks")
report_lines.extend(["", "RECOMMENDATIONS", "-" * 15])
recommendations = quality.get("recommendations", [])
if recommendations:
for i, rec in enumerate(recommendations, 1):
report_lines.append(f"{i}. {rec}")
else:
report_lines.append("No specific recommendations. Baseline quality is acceptable.")
return "\n".join(report_lines)
__init__(self, metrics_dir: str = None)
special
¶
Initialize baseline manager with metrics directory
Source code in hazelbean_tests/performance/baseline_manager.py
def __init__(self, metrics_dir: str = None):
"""Initialize baseline manager with metrics directory"""
if metrics_dir is None:
# Default to project metrics directory
self.metrics_dir = Path(__file__).parent.parent.parent / "metrics"
else:
self.metrics_dir = Path(metrics_dir)
self.metrics_dir.mkdir(exist_ok=True)
# Create organized directory structure
self.baselines_dir = self.metrics_dir / "baselines"
self.snapshots_dir = self.baselines_dir / "snapshots"
self.benchmarks_dir = self.metrics_dir / "benchmarks"
# Create directories
self.baselines_dir.mkdir(exist_ok=True)
self.snapshots_dir.mkdir(exist_ok=True)
self.benchmarks_dir.mkdir(exist_ok=True)
# Main baseline file location
self.baseline_file = self.baselines_dir / "current_performance_baseline.json"
# Keep backward compatibility with historical_dir for existing code
self.historical_dir = self.snapshots_dir
logger.info(f"BaselineManager initialized with metrics_dir: {self.metrics_dir}")
create_standardized_baseline_structure(self, benchmark_data: Dict[str, Any], baseline_version: str = '2.0') -> Dict[str, Any]
¶
Create standardized baseline JSON structure for all benchmark metrics.
Task 6.1: Create baseline JSON structure for all benchmark metrics
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
benchmark_data |
Dict[str, Any] |
Raw benchmark data from pytest-benchmark |
required |
baseline_version |
str |
Version identifier for the baseline format |
'2.0' |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] |
Standardized baseline structure with comprehensive metadata |
Source code in hazelbean_tests/performance/baseline_manager.py
def create_standardized_baseline_structure(self,
benchmark_data: Dict[str, Any],
baseline_version: str = "2.0") -> Dict[str, Any]:
"""
Create standardized baseline JSON structure for all benchmark metrics.
Task 6.1: Create baseline JSON structure for all benchmark metrics
Args:
benchmark_data: Raw benchmark data from pytest-benchmark
baseline_version: Version identifier for the baseline format
Returns:
Standardized baseline structure with comprehensive metadata
"""
# Extract git information for version control integration
git_info = self._get_git_information()
# Calculate comprehensive statistics
baseline_stats = self._calculate_baseline_statistics(benchmark_data)
# Create standardized structure
baseline_structure = {
"baseline_metadata": {
"version": baseline_version,
"created_at": datetime.now(timezone.utc).isoformat(),
"schema_version": "2.0.0",
"description": "Hazelbean performance baseline with comprehensive metrics",
"statistical_confidence": "95%",
"regression_threshold_percent": 10.0,
"trend_analysis_enabled": True
},
"version_control_info": git_info,
"system_environment": self._extract_system_info(benchmark_data),
"baseline_statistics": baseline_stats,
"benchmark_categories": self._categorize_benchmarks(benchmark_data),
"quality_metrics": self._calculate_quality_metrics(baseline_stats),
"raw_benchmark_data": benchmark_data.get("benchmarks", []),
"validation_info": {
"total_benchmarks": len(benchmark_data.get("benchmarks", [])),
"valid_benchmarks": len([b for b in benchmark_data.get("benchmarks", []) if self._is_valid_benchmark(b)]),
"statistical_confidence_met": True,
"baseline_establishment_criteria": {
"minimum_runs": 5,
"maximum_variance_threshold": 0.25,
"outlier_detection_enabled": True
}
}
}
logger.info(f"Created standardized baseline structure with {baseline_structure['validation_info']['total_benchmarks']} benchmarks")
return baseline_structure
save_baseline(self, baseline_data: Dict[str, Any]) -> str
¶
Save baseline data with version control integration.
Task 6.4: Set up version control integration for baseline artifacts
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
baseline_data |
Dict[str, Any] |
Standardized baseline data structure |
required |
Returns:
| Type | Description |
|---|---|
str |
Path to saved baseline file |
Source code in hazelbean_tests/performance/baseline_manager.py
def save_baseline(self, baseline_data: Dict[str, Any]) -> str:
"""
Save baseline data with version control integration.
Task 6.4: Set up version control integration for baseline artifacts
Args:
baseline_data: Standardized baseline data structure
Returns:
Path to saved baseline file
"""
# Save main baseline file
with open(self.baseline_file, 'w') as f:
json.dump(baseline_data, f, indent=2)
# Create baseline snapshot
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
git_hash = baseline_data.get("version_control_info", {}).get("commit_id", "unknown")[:8]
# Save snapshot with descriptive naming
snapshot_file = self.snapshots_dir / f"baseline_snapshot_{timestamp}_{git_hash}.json"
with open(snapshot_file, 'w') as f:
json.dump(baseline_data, f, indent=2)
logger.info(f"Saved current baseline to {self.baseline_file} and snapshot to {snapshot_file}")
return str(self.baseline_file)
compare_with_baseline(self, current_data: Dict[str, Any]) -> Dict[str, Any]
¶
Implement baseline comparison logic for performance regression detection.
Task 6.2: Implement baseline comparison logic for performance regression detection
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
current_data |
Dict[str, Any] |
Current benchmark results to compare against baseline |
required |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] |
Comprehensive comparison results with regression detection |
Source code in hazelbean_tests/performance/baseline_manager.py
def compare_with_baseline(self, current_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Implement baseline comparison logic for performance regression detection.
Task 6.2: Implement baseline comparison logic for performance regression detection
Args:
current_data: Current benchmark results to compare against baseline
Returns:
Comprehensive comparison results with regression detection
"""
if not self.baseline_file.exists():
logger.warning("No baseline file found. Creating initial baseline.")
return {"status": "no_baseline", "action": "create_baseline"}
# Load baseline data
with open(self.baseline_file, 'r') as f:
baseline_data = json.load(f)
comparison_results = {
"comparison_metadata": {
"comparison_timestamp": datetime.now(timezone.utc).isoformat(),
"baseline_version": baseline_data.get("baseline_metadata", {}).get("version", "unknown"),
"comparison_type": "performance_regression_analysis"
},
"regression_analysis": {},
"performance_changes": {},
"statistical_significance": {},
"recommendations": [],
"overall_status": "passed"
}
# Compare each benchmark
baseline_benchmarks = {b["name"]: b for b in baseline_data.get("raw_benchmark_data", [])}
current_benchmarks = {b["name"]: b for b in current_data.get("benchmarks", [])}
regression_threshold = baseline_data.get("baseline_metadata", {}).get("regression_threshold_percent", 10.0)
for benchmark_name, current_benchmark in current_benchmarks.items():
if benchmark_name in baseline_benchmarks:
baseline_benchmark = baseline_benchmarks[benchmark_name]
regression_analysis = self._analyze_regression(
baseline_benchmark, current_benchmark, regression_threshold
)
comparison_results["regression_analysis"][benchmark_name] = regression_analysis
if regression_analysis["is_regression"]:
comparison_results["overall_status"] = "regression_detected"
# Add recommendations
comparison_results["recommendations"] = self._generate_recommendations(comparison_results)
logger.info(f"Completed baseline comparison. Status: {comparison_results['overall_status']}")
return comparison_results
analyze_trends(self, lookback_days: int = 30) -> Dict[str, Any]
¶
Add trend analysis and historical tracking capabilities.
Task 6.3: Add trend analysis and historical tracking capabilities
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
lookback_days |
int |
Number of days to analyze for trends |
30 |
Returns:
| Type | Description |
|---|---|
Dict[str, Any] |
Comprehensive trend analysis with historical tracking |
Source code in hazelbean_tests/performance/baseline_manager.py
def analyze_trends(self, lookback_days: int = 30) -> Dict[str, Any]:
"""
Add trend analysis and historical tracking capabilities.
Task 6.3: Add trend analysis and historical tracking capabilities
Args:
lookback_days: Number of days to analyze for trends
Returns:
Comprehensive trend analysis with historical tracking
"""
# Collect baseline snapshot files
historical_files = list(self.snapshots_dir.glob("baseline_snapshot_*.json"))
historical_files.sort() # Sort by filename (which includes timestamp)
if len(historical_files) < 2:
return {
"status": "insufficient_data",
"message": f"Need at least 2 historical baselines for trend analysis. Found: {len(historical_files)}"
}
trend_data = {
"trend_metadata": {
"analysis_timestamp": datetime.now(timezone.utc).isoformat(),
"lookback_days": lookback_days,
"analyzed_files": len(historical_files),
"trend_detection_algorithm": "linear_regression_with_statistical_significance"
},
"benchmark_trends": {},
"performance_trajectory": {},
"anomaly_detection": {},
"trend_summary": {
"improving_benchmarks": [],
"degrading_benchmarks": [],
"stable_benchmarks": [],
"anomalous_benchmarks": []
}
}
# Analyze trends for each benchmark
benchmark_history = self._collect_benchmark_history(historical_files)
for benchmark_name, history in benchmark_history.items():
if len(history) >= 3: # Need minimum data points for trend analysis
trend_analysis = self._calculate_trend_metrics(benchmark_name, history)
trend_data["benchmark_trends"][benchmark_name] = trend_analysis
# Categorize benchmark trends
self._categorize_benchmark_trend(benchmark_name, trend_analysis, trend_data["trend_summary"])
# Generate performance trajectory summary
trend_data["performance_trajectory"] = self._generate_performance_trajectory(benchmark_history)
logger.info(f"Completed trend analysis for {len(benchmark_history)} benchmarks over {len(historical_files)} baseline files")
return trend_data
_get_git_information(self) -> Dict[str, Any]
private
¶
Extract git information for version control integration
Source code in hazelbean_tests/performance/baseline_manager.py
def _get_git_information(self) -> Dict[str, Any]:
"""Extract git information for version control integration"""
try:
git_info = {
"commit_id": subprocess.check_output(["git", "rev-parse", "HEAD"]).decode().strip(),
"commit_timestamp": subprocess.check_output(["git", "show", "-s", "--format=%ci", "HEAD"]).decode().strip(),
"branch": subprocess.check_output(["git", "rev-parse", "--abbrev-ref", "HEAD"]).decode().strip(),
"is_dirty": len(subprocess.check_output(["git", "status", "--porcelain"]).decode().strip()) > 0,
"author": subprocess.check_output(["git", "show", "-s", "--format=%an", "HEAD"]).decode().strip(),
"repository_url": self._get_repository_url()
}
except subprocess.CalledProcessError:
git_info = {
"commit_id": "unknown",
"commit_timestamp": "unknown",
"branch": "unknown",
"is_dirty": False,
"author": "unknown",
"repository_url": "unknown"
}
return git_info
_get_repository_url(self) -> str
private
¶
Get repository URL for version control tracking
Source code in hazelbean_tests/performance/baseline_manager.py
_extract_system_info(self, benchmark_data: Dict[str, Any]) -> Dict[str, Any]
private
¶
Extract and standardize system environment information
Source code in hazelbean_tests/performance/baseline_manager.py
def _extract_system_info(self, benchmark_data: Dict[str, Any]) -> Dict[str, Any]:
"""Extract and standardize system environment information"""
machine_info = benchmark_data.get("machine_info", {})
return {
"platform": {
"system": machine_info.get("system", "unknown"),
"node": machine_info.get("node", "unknown"),
"release": machine_info.get("release", "unknown"),
"machine": machine_info.get("machine", "unknown"),
"processor": machine_info.get("processor", "unknown")
},
"python_environment": {
"version": machine_info.get("python_version", "unknown"),
"implementation": machine_info.get("python_implementation", "unknown"),
"compiler": machine_info.get("python_compiler", "unknown")
},
"cpu_info": machine_info.get("cpu", {}),
"environment_hash": self._calculate_environment_hash(machine_info)
}
_calculate_environment_hash(self, machine_info: Dict[str, Any]) -> str
private
¶
Calculate hash of environment for compatibility checking
Source code in hazelbean_tests/performance/baseline_manager.py
def _calculate_environment_hash(self, machine_info: Dict[str, Any]) -> str:
"""Calculate hash of environment for compatibility checking"""
import hashlib
env_string = f"{machine_info.get('system')}_{machine_info.get('machine')}_{machine_info.get('python_version')}"
return hashlib.sha256(env_string.encode()).hexdigest()[:16]
_calculate_baseline_statistics(self, benchmark_data: Dict[str, Any]) -> Dict[str, Any]
private
¶
Calculate comprehensive baseline statistics
Source code in hazelbean_tests/performance/baseline_manager.py
def _calculate_baseline_statistics(self, benchmark_data: Dict[str, Any]) -> Dict[str, Any]:
"""Calculate comprehensive baseline statistics"""
benchmarks = benchmark_data.get("benchmarks", [])
if not benchmarks:
return {"error": "no_benchmark_data"}
execution_times = []
for benchmark in benchmarks:
stats = benchmark.get("stats", {})
if "mean" in stats:
try:
mean_value = float(stats["mean"])
if mean_value > 0: # Only include positive values
execution_times.append(mean_value)
except (ValueError, TypeError):
# Skip invalid values
logger.warning(f"Invalid mean value in benchmark {benchmark.get('name', 'unknown')}: {stats['mean']}")
continue
if not execution_times:
return {"error": "no_valid_execution_times"}
return {
"aggregate_statistics": {
"mean_execution_time": statistics.mean(execution_times),
"median_execution_time": statistics.median(execution_times),
"std_deviation": statistics.stdev(execution_times) if len(execution_times) > 1 else 0,
"min_time": min(execution_times),
"max_time": max(execution_times),
"total_benchmarks": len(execution_times)
},
"confidence_intervals": self._calculate_confidence_intervals(execution_times),
"outlier_analysis": self._detect_outliers(execution_times),
"quality_indicators": {
"coefficient_of_variation": (statistics.stdev(execution_times) / statistics.mean(execution_times)) if len(execution_times) > 1 and statistics.mean(execution_times) > 0 else 0,
"acceptable_variance": (statistics.stdev(execution_times) / statistics.mean(execution_times)) < 0.25 if len(execution_times) > 1 and statistics.mean(execution_times) > 0 else False
}
}
_calculate_confidence_intervals(self, data: List[float], confidence: float = 0.95) -> Dict[str, float]
private
¶
Calculate confidence intervals for baseline statistics
Source code in hazelbean_tests/performance/baseline_manager.py
def _calculate_confidence_intervals(self, data: List[float], confidence: float = 0.95) -> Dict[str, float]:
"""Calculate confidence intervals for baseline statistics"""
if len(data) < 2:
return {"lower": 0, "upper": 0, "confidence_level": confidence}
mean = statistics.mean(data)
std_dev = statistics.stdev(data) if len(data) > 1 else 0
n = len(data)
# Using t-distribution for small samples
import math
if n < 30:
# Simplified t-distribution approximation
t_value = 2.0 # Approximate t-value for 95% confidence
else:
t_value = 1.96 # z-value for 95% confidence
margin_error = t_value * (std_dev / math.sqrt(n))
return {
"lower": mean - margin_error,
"upper": mean + margin_error,
"confidence_level": confidence,
"margin_of_error": margin_error
}
_detect_outliers(self, data: List[float]) -> Dict[str, Any]
private
¶
Detect outliers in benchmark data using IQR method
Source code in hazelbean_tests/performance/baseline_manager.py
def _detect_outliers(self, data: List[float]) -> Dict[str, Any]:
"""Detect outliers in benchmark data using IQR method"""
if len(data) < 4:
return {"outliers": [], "method": "insufficient_data"}
sorted_data = sorted(data)
n = len(sorted_data)
q1_index = n // 4
q3_index = 3 * n // 4
q1 = sorted_data[q1_index]
q3 = sorted_data[q3_index]
iqr = q3 - q1
lower_bound = q1 - 1.5 * iqr
upper_bound = q3 + 1.5 * iqr
outliers = [x for x in data if x < lower_bound or x > upper_bound]
return {
"outliers": outliers,
"outlier_count": len(outliers),
"outlier_percentage": (len(outliers) / len(data)) * 100,
"method": "IQR",
"bounds": {"lower": lower_bound, "upper": upper_bound},
"quartiles": {"q1": q1, "q3": q3, "iqr": iqr}
}
_categorize_benchmarks(self, benchmark_data: Dict[str, Any]) -> Dict[str, List[str]]
private
¶
Categorize benchmarks by type and functionality
Source code in hazelbean_tests/performance/baseline_manager.py
def _categorize_benchmarks(self, benchmark_data: Dict[str, Any]) -> Dict[str, List[str]]:
"""Categorize benchmarks by type and functionality"""
categories = {
"path_resolution": [],
"tiling_operations": [],
"data_processing": [],
"io_operations": [],
"computational": [],
"integration": [],
"uncategorized": []
}
for benchmark in benchmark_data.get("benchmarks", []):
name = benchmark.get("name", "").lower()
categorized = False
if any(keyword in name for keyword in ["path", "get_path", "resolution"]):
categories["path_resolution"].append(benchmark["name"])
categorized = True
elif any(keyword in name for keyword in ["tile", "tiling", "iterator"]):
categories["tiling_operations"].append(benchmark["name"])
categorized = True
elif any(keyword in name for keyword in ["array", "processing", "calculation"]):
categories["data_processing"].append(benchmark["name"])
categorized = True
elif any(keyword in name for keyword in ["io", "read", "write", "load", "save"]):
categories["io_operations"].append(benchmark["name"])
categorized = True
elif any(keyword in name for keyword in ["integration", "workflow", "end_to_end"]):
categories["integration"].append(benchmark["name"])
categorized = True
elif any(keyword in name for keyword in ["compute", "algorithm", "math"]):
categories["computational"].append(benchmark["name"])
categorized = True
if not categorized:
categories["uncategorized"].append(benchmark["name"])
return categories
_calculate_quality_metrics(self, baseline_stats: Dict[str, Any]) -> Dict[str, Any]
private
¶
Calculate quality metrics for baseline establishment
Source code in hazelbean_tests/performance/baseline_manager.py
def _calculate_quality_metrics(self, baseline_stats: Dict[str, Any]) -> Dict[str, Any]:
"""Calculate quality metrics for baseline establishment"""
aggregate_stats = baseline_stats.get("aggregate_statistics", {})
return {
"baseline_quality_score": self._calculate_quality_score(baseline_stats),
"statistical_reliability": {
"sufficient_sample_size": aggregate_stats.get("total_benchmarks", 0) >= 5,
"acceptable_variance": baseline_stats.get("quality_indicators", {}).get("acceptable_variance", False),
"outlier_percentage": baseline_stats.get("outlier_analysis", {}).get("outlier_percentage", 0),
"confidence_interval_width": self._calculate_ci_width(baseline_stats)
},
"recommendations": self._generate_quality_recommendations(baseline_stats)
}
_calculate_quality_score(self, baseline_stats: Dict[str, Any]) -> float
private
¶
Calculate overall quality score for baseline (0-100)
Source code in hazelbean_tests/performance/baseline_manager.py
def _calculate_quality_score(self, baseline_stats: Dict[str, Any]) -> float:
"""Calculate overall quality score for baseline (0-100)"""
score = 100.0
# Penalize high variance
cv = baseline_stats.get("quality_indicators", {}).get("coefficient_of_variation", 0)
if cv > 0.25:
score -= min(30, cv * 100)
# Penalize high outlier percentage
outlier_pct = baseline_stats.get("outlier_analysis", {}).get("outlier_percentage", 0)
if outlier_pct > 10:
score -= min(20, outlier_pct)
# Penalize small sample size
sample_size = baseline_stats.get("aggregate_statistics", {}).get("total_benchmarks", 0)
if sample_size < 5:
score -= 40
elif sample_size < 10:
score -= 20
return max(0, score)
_calculate_ci_width(self, baseline_stats: Dict[str, Any]) -> float
private
¶
Calculate confidence interval width as percentage of mean
Source code in hazelbean_tests/performance/baseline_manager.py
def _calculate_ci_width(self, baseline_stats: Dict[str, Any]) -> float:
"""Calculate confidence interval width as percentage of mean"""
ci = baseline_stats.get("confidence_intervals", {})
mean = baseline_stats.get("aggregate_statistics", {}).get("mean_execution_time", 0)
if mean > 0 and "upper" in ci and "lower" in ci:
width = ci["upper"] - ci["lower"]
return (width / mean) * 100
return 0
_generate_quality_recommendations(self, baseline_stats: Dict[str, Any]) -> List[str]
private
¶
Generate recommendations for improving baseline quality
Source code in hazelbean_tests/performance/baseline_manager.py
def _generate_quality_recommendations(self, baseline_stats: Dict[str, Any]) -> List[str]:
"""Generate recommendations for improving baseline quality"""
recommendations = []
quality_indicators = baseline_stats.get("quality_indicators", {})
if not quality_indicators.get("acceptable_variance", True):
recommendations.append("High variance detected. Consider running more benchmark iterations or investigating environmental factors.")
outlier_pct = baseline_stats.get("outlier_analysis", {}).get("outlier_percentage", 0)
if outlier_pct > 15:
recommendations.append("High outlier percentage. Review benchmark setup and consider environment stabilization.")
sample_size = baseline_stats.get("aggregate_statistics", {}).get("total_benchmarks", 0)
if sample_size < 10:
recommendations.append("Small sample size. Consider adding more benchmark tests for better statistical reliability.")
return recommendations
_is_valid_benchmark(self, benchmark: Dict[str, Any]) -> bool
private
¶
Check if a benchmark result is valid for baseline inclusion
Source code in hazelbean_tests/performance/baseline_manager.py
def _is_valid_benchmark(self, benchmark: Dict[str, Any]) -> bool:
"""Check if a benchmark result is valid for baseline inclusion"""
stats = benchmark.get("stats", {})
# Check if mean exists and can be converted to float
if "mean" not in stats:
return False
try:
mean_value = float(stats["mean"])
if mean_value <= 0:
return False
except (ValueError, TypeError):
return False
# Check rounds if present
try:
rounds = stats.get("rounds", 1)
if isinstance(rounds, str):
rounds = float(rounds)
if rounds <= 0:
return False
except (ValueError, TypeError):
return False
return True
_analyze_regression(self, baseline_benchmark: Dict[str, Any], current_benchmark: Dict[str, Any], threshold_percent: float) -> Dict[str, Any]
private
¶
Analyze potential regression between baseline and current benchmark
Source code in hazelbean_tests/performance/baseline_manager.py
def _analyze_regression(self, baseline_benchmark: Dict[str, Any],
current_benchmark: Dict[str, Any],
threshold_percent: float) -> Dict[str, Any]:
"""Analyze potential regression between baseline and current benchmark"""
baseline_mean = baseline_benchmark.get("stats", {}).get("mean", 0)
current_mean = current_benchmark.get("stats", {}).get("mean", 0)
if baseline_mean == 0:
return {"is_regression": False, "reason": "invalid_baseline_data"}
percent_change = ((current_mean - baseline_mean) / baseline_mean) * 100
is_regression = percent_change > threshold_percent
return {
"is_regression": is_regression,
"percent_change": percent_change,
"baseline_mean": baseline_mean,
"current_mean": current_mean,
"absolute_difference": current_mean - baseline_mean,
"threshold_percent": threshold_percent,
"severity": self._classify_regression_severity(percent_change, threshold_percent),
"statistical_significance": self._check_statistical_significance(baseline_benchmark, current_benchmark)
}
_classify_regression_severity(self, percent_change: float, threshold: float) -> str
private
¶
Classify regression severity based on performance change
Source code in hazelbean_tests/performance/baseline_manager.py
def _classify_regression_severity(self, percent_change: float, threshold: float) -> str:
"""Classify regression severity based on performance change"""
if percent_change <= threshold:
return "no_regression"
elif percent_change <= threshold * 2:
return "minor_regression"
elif percent_change <= threshold * 5:
return "major_regression"
else:
return "critical_regression"
_check_statistical_significance(self, baseline: Dict[str, Any], current: Dict[str, Any]) -> Dict[str, Any]
private
¶
Check statistical significance of performance difference
Source code in hazelbean_tests/performance/baseline_manager.py
def _check_statistical_significance(self, baseline: Dict[str, Any], current: Dict[str, Any]) -> Dict[str, Any]:
"""Check statistical significance of performance difference"""
baseline_stats = baseline.get("stats", {})
current_stats = current.get("stats", {})
# Simplified significance check using standard deviation
baseline_std = baseline_stats.get("stddev", 0)
current_std = current_stats.get("stddev", 0)
baseline_mean = baseline_stats.get("mean", 0)
current_mean = current_stats.get("mean", 0)
if baseline_std == 0 or current_std == 0:
return {"significant": False, "method": "insufficient_variance_data"}
# Simple two-standard-deviation test
combined_std = (baseline_std + current_std) / 2
difference = abs(current_mean - baseline_mean)
is_significant = difference > (2 * combined_std)
return {
"significant": is_significant,
"method": "two_standard_deviation_test",
"difference": difference,
"threshold": 2 * combined_std,
"confidence_level": "approximately_95_percent"
}
_generate_recommendations(self, comparison_results: Dict[str, Any]) -> List[str]
private
¶
Generate recommendations based on comparison results
Source code in hazelbean_tests/performance/baseline_manager.py
def _generate_recommendations(self, comparison_results: Dict[str, Any]) -> List[str]:
"""Generate recommendations based on comparison results"""
recommendations = []
if comparison_results["overall_status"] == "regression_detected":
recommendations.append("Performance regression detected. Review recent changes and consider performance optimization.")
# Count regressions by severity
severe_regressions = sum(1 for analysis in comparison_results["regression_analysis"].values()
if analysis.get("severity") in ["major_regression", "critical_regression"])
if severe_regressions > 0:
recommendations.append(f"Critical performance regressions found in {severe_regressions} benchmark(s). Immediate attention required.")
return recommendations
_collect_benchmark_history(self, historical_files: List[pathlib._local.Path]) -> Dict[str, List[Dict[str, Any]]]
private
¶
Collect benchmark history from historical baseline files
Source code in hazelbean_tests/performance/baseline_manager.py
def _collect_benchmark_history(self, historical_files: List[Path]) -> Dict[str, List[Dict[str, Any]]]:
"""Collect benchmark history from historical baseline files"""
benchmark_history = {}
for file_path in historical_files:
try:
with open(file_path, 'r') as f:
historical_data = json.load(f)
# Extract timestamp from metadata or filename
timestamp = historical_data.get("baseline_metadata", {}).get("created_at")
if not timestamp:
# Extract from filename if not in metadata
timestamp = file_path.stem.split("_")[1] if "_" in file_path.stem else "unknown"
for benchmark in historical_data.get("raw_benchmark_data", []):
name = benchmark.get("name")
if name:
if name not in benchmark_history:
benchmark_history[name] = []
benchmark_history[name].append({
"timestamp": timestamp,
"stats": benchmark.get("stats", {}),
"file_source": str(file_path)
})
except (json.JSONDecodeError, FileNotFoundError) as e:
logger.warning(f"Could not process historical file {file_path}: {e}")
# Sort each benchmark's history by timestamp
for name in benchmark_history:
benchmark_history[name].sort(key=lambda x: x["timestamp"])
return benchmark_history
_calculate_trend_metrics(self, benchmark_name: str, history: List[Dict[str, Any]]) -> Dict[str, Any]
private
¶
Calculate trend metrics for a specific benchmark
Source code in hazelbean_tests/performance/baseline_manager.py
def _calculate_trend_metrics(self, benchmark_name: str, history: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Calculate trend metrics for a specific benchmark"""
execution_times = [entry["stats"].get("mean", 0) for entry in history if entry["stats"].get("mean", 0) > 0]
if len(execution_times) < 3:
return {"trend": "insufficient_data", "data_points": len(execution_times)}
# Simple linear trend calculation
x_values = list(range(len(execution_times)))
trend_slope = self._calculate_linear_trend(x_values, execution_times)
return {
"trend": "improving" if trend_slope < -0.001 else "degrading" if trend_slope > 0.001 else "stable",
"slope": trend_slope,
"data_points": len(execution_times),
"latest_value": execution_times[-1],
"earliest_value": execution_times[0],
"total_change_percent": ((execution_times[-1] - execution_times[0]) / execution_times[0]) * 100 if execution_times[0] > 0 else 0,
"volatility": statistics.stdev(execution_times) if len(execution_times) > 1 else 0
}
_calculate_linear_trend(self, x_values: List[int], y_values: List[float]) -> float
private
¶
Calculate linear trend slope using least squares
Source code in hazelbean_tests/performance/baseline_manager.py
def _calculate_linear_trend(self, x_values: List[int], y_values: List[float]) -> float:
"""Calculate linear trend slope using least squares"""
n = len(x_values)
if n < 2:
return 0
sum_x = sum(x_values)
sum_y = sum(y_values)
sum_xy = sum(x * y for x, y in zip(x_values, y_values))
sum_x2 = sum(x * x for x in x_values)
denominator = n * sum_x2 - sum_x * sum_x
if denominator == 0:
return 0
slope = (n * sum_xy - sum_x * sum_y) / denominator
return slope
_categorize_benchmark_trend(self, benchmark_name: str, trend_analysis: Dict[str, Any], trend_summary: Dict[str, List])
private
¶
Categorize benchmark into trend summary categories
Source code in hazelbean_tests/performance/baseline_manager.py
def _categorize_benchmark_trend(self, benchmark_name: str, trend_analysis: Dict[str, Any], trend_summary: Dict[str, List]):
"""Categorize benchmark into trend summary categories"""
trend = trend_analysis.get("trend", "unknown")
if trend == "improving":
trend_summary["improving_benchmarks"].append(benchmark_name)
elif trend == "degrading":
trend_summary["degrading_benchmarks"].append(benchmark_name)
elif trend == "stable":
trend_summary["stable_benchmarks"].append(benchmark_name)
else:
trend_summary["anomalous_benchmarks"].append(benchmark_name)
_generate_performance_trajectory(self, benchmark_history: Dict[str, List[Dict[str, Any]]]) -> Dict[str, Any]
private
¶
Generate overall performance trajectory summary
Source code in hazelbean_tests/performance/baseline_manager.py
def _generate_performance_trajectory(self, benchmark_history: Dict[str, List[Dict[str, Any]]]) -> Dict[str, Any]:
"""Generate overall performance trajectory summary"""
if not benchmark_history:
return {"status": "no_data"}
# Calculate overall performance trend
all_latest_times = []
all_earliest_times = []
for benchmark_data in benchmark_history.values():
if len(benchmark_data) >= 2:
earliest = benchmark_data[0]["stats"].get("mean", 0)
latest = benchmark_data[-1]["stats"].get("mean", 0)
if earliest > 0 and latest > 0:
all_earliest_times.append(earliest)
all_latest_times.append(latest)
if not all_latest_times or not all_earliest_times:
return {"status": "insufficient_data"}
avg_earliest = statistics.mean(all_earliest_times)
avg_latest = statistics.mean(all_latest_times)
overall_change = ((avg_latest - avg_earliest) / avg_earliest) * 100 if avg_earliest > 0 else 0
return {
"overall_trend": "improving" if overall_change < -1 else "degrading" if overall_change > 1 else "stable",
"overall_change_percent": overall_change,
"benchmarks_analyzed": len(all_latest_times),
"average_earliest_time": avg_earliest,
"average_latest_time": avg_latest,
"performance_health": "good" if overall_change < 5 else "concerning" if overall_change < 15 else "poor"
}
generate_baseline_report(self, baseline_data: Dict[str, Any]) -> str
¶
Generate a human-readable baseline establishment report
Source code in hazelbean_tests/performance/baseline_manager.py
def generate_baseline_report(self, baseline_data: Dict[str, Any]) -> str:
"""Generate a human-readable baseline establishment report"""
report_lines = [
"HAZELBEAN PERFORMANCE BASELINE ESTABLISHMENT REPORT",
"=" * 55,
"",
f"Baseline Version: {baseline_data.get('baseline_metadata', {}).get('version', 'unknown')}",
f"Created: {baseline_data.get('baseline_metadata', {}).get('created_at', 'unknown')}",
f"Git Commit: {baseline_data.get('version_control_info', {}).get('commit_id', 'unknown')[:12]}",
f"Branch: {baseline_data.get('version_control_info', {}).get('branch', 'unknown')}",
"",
"BASELINE STATISTICS",
"-" * 20,
]
stats = baseline_data.get("baseline_statistics", {}).get("aggregate_statistics", {})
report_lines.extend([
f"Total Benchmarks: {stats.get('total_benchmarks', 0)}",
f"Mean Execution Time: {stats.get('mean_execution_time', 0):.6f}s",
f"Standard Deviation: {stats.get('std_deviation', 0):.6f}s",
f"Min Time: {stats.get('min_time', 0):.6f}s",
f"Max Time: {stats.get('max_time', 0):.6f}s",
"",
"QUALITY METRICS",
"-" * 15,
])
quality = baseline_data.get("quality_metrics", {})
report_lines.extend([
f"Baseline Quality Score: {quality.get('baseline_quality_score', 0):.1f}/100",
f"Statistical Reliability: {'PASS' if quality.get('statistical_reliability', {}).get('sufficient_sample_size', False) else 'FAIL'}",
f"Variance Acceptable: {'YES' if quality.get('statistical_reliability', {}).get('acceptable_variance', False) else 'NO'}",
"",
"BENCHMARK CATEGORIES",
"-" * 20,
])
categories = baseline_data.get("benchmark_categories", {})
for category, benchmarks in categories.items():
if benchmarks:
report_lines.append(f"{category.replace('_', ' ').title()}: {len(benchmarks)} benchmarks")
report_lines.extend(["", "RECOMMENDATIONS", "-" * 15])
recommendations = quality.get("recommendations", [])
if recommendations:
for i, rec in enumerate(recommendations, 1):
report_lines.append(f"{i}. {rec}")
else:
report_lines.append("No specific recommendations. Baseline quality is acceptable.")
return "\n".join(report_lines)
Running Performance Tests¶
To run the complete performance test suite:
# Activate the hazelbean environment
conda activate hazelbean_env
# Run all performance tests
pytest hazelbean_tests/performance/ -v
# Run performance tests with benchmarking
pytest hazelbean_tests/performance/ -v --benchmark-only
# Run with performance profiling
pytest hazelbean_tests/performance/ --profile
# Generate performance report
python scripts/run_performance_benchmarks.py
Performance Metrics¶
Performance tests measure:
- Execution Time - How long operations take to complete
- Memory Usage - RAM consumption during processing
- CPU Utilization - Processor usage patterns
- I/O Performance - File read/write speeds
- Scalability - Performance with different data sizes
Baseline Management¶
The performance testing system includes baseline management to:
- Track Changes - Monitor performance trends over time
- Detect Regressions - Alert when performance degrades
- Validate Optimizations - Confirm performance improvements
- Generate Reports - Create performance analysis documents
Performance Artifacts¶
Performance tests generate artifacts in:
hazelbean_tests/performance/artifacts/- Benchmark results and reportsbaselines/- Performance baseline snapshotsmetrics/- Historical performance data
Interpreting Results¶
When analyzing performance test results:
- Compare to Baselines - Look for significant deviations
- Consider Data Size - Performance scales with input data
- Account for System Variation - Results may vary between runs
- Focus on Trends - Long-term patterns are more meaningful than individual measurements
Optimization Guidelines¶
Based on performance test results:
- Identify Bottlenecks - Find the slowest operations
- Optimize Critical Paths - Focus on frequently used functions
- Consider Memory vs Speed - Balance memory usage and execution time
- Test with Real Data - Use realistic datasets for accurate measurements
Related Test Categories¶
- Unit Tests → Understand component performance in Unit Tests
- Integration Tests → See workflow performance in Integration Tests
- System Tests → Validate system-wide performance in System Tests
Performance Test Matrix¶
| Test Focus | Test File | Metrics Tracked | Related Components |
|---|---|---|---|
| Function Benchmarks | test_functions.py | Execution time, memory usage | Unit-tested functions |
| Workflow Performance | test_workflows.py | End-to-end timing | Integration workflows |
| Baseline Tracking | test_benchmarks.py | Historical trends | All components |
| Baseline Management | test_baseline_manager.py | System reliability | Performance infrastructure |
Performance Baselines¶
Current performance targets:
- Small datasets (<100MB): <30 seconds processing time
- Medium datasets (100MB-1GB): <5 minutes processing time
- Large datasets (>1GB): Tracked for optimization opportunities
- Memory usage: <2GB RAM for typical workflows