techprotrade's picture
Deploy ATOM FastAPI command center runtime (part 9)
81e3673 verified
Raw
History Blame Contribute Delete
3.75 kB
"""
Soak Test Configuration and Fixtures
This module provides shared fixtures for soak testing (extended duration tests for
memory leak detection and resource stability validation).
Soak tests run for 1-4 hours to detect:
- Memory leaks (slow growth over time)
- Connection pool exhaustion
- Cache unbounded growth
- Resource leaks
Fixtures:
- memory_monitor: Track memory usage using psutil
- enable_gc_control: Control garbage collection during tests
- soak_test_config: Configuration for memory thresholds and logging intervals
"""
import gc
import pytest
import psutil
from typing import Dict, Any
@pytest.fixture
def memory_monitor() -> Dict[str, Any]:
"""
Fixture providing memory monitoring capabilities.
Returns a dictionary with:
- process: psutil.Process instance for memory tracking
- initial_memory_mb: Initial memory usage in MB
- check_interval: Seconds between memory checks (default: 60)
Usage:
def test_memory_leak(memory_monitor):
process = memory_monitor["process"]
initial = memory_monitor["initial_memory_mb"]
current = process.memory_info().rss / 1024 / 1024
growth = current - initial
assert growth < 100, f"Memory grew {growth:.2f}MB"
"""
process = psutil.Process()
initial_memory_mb = process.memory_info().rss / 1024 / 1024
return {
"process": process,
"initial_memory_mb": initial_memory_mb,
"check_interval": 60 # Check memory every 60 seconds
}
@pytest.fixture
def enable_gc_control() -> Dict[str, Any]:
"""
Fixture enabling garbage collection control for leak detection.
Enables GC debug stats and provides a control object with:
- collect(): Method to force garbage collection
- stats_enabled: Boolean indicating if stats are enabled
Purpose: Force GC during tests to distinguish memory leaks from
cached data that hasn't been collected yet.
Usage:
def test_with_gc(enable_gc_control):
# Run operations that may leak memory
enable_gc_control["collect"]() # Force GC
# Check memory usage after GC
"""
# Enable GC debug stats to track collection cycles
gc.set_debug(gc.DEBUG_STATS)
return {
"collect": lambda: gc.collect(),
"stats_enabled": True
}
@pytest.fixture
def soak_test_config() -> Dict[str, int]:
"""
Fixture providing soak test configuration thresholds.
Returns configuration dict with:
- memory_threshold_1hr_mb: Max allowed memory growth over 1 hour (100MB)
- memory_threshold_2hr_mb: Max allowed memory growth over 2 hours (200MB)
- fail_fast_threshold_mb: Immediate failure if memory growth exceeds this (500MB)
- log_interval_seconds: How often to log memory usage (60 seconds)
Purpose: Centralize soak test thresholds for consistent memory leak
detection across all soak tests.
Usage:
def test_soak(soak_test_config, memory_monitor):
threshold = soak_test_config["memory_threshold_1hr_mb"]
# Run test, fail if memory_growth > threshold
"""
return {
"memory_threshold_1hr_mb": 100,
"memory_threshold_2hr_mb": 200,
"fail_fast_threshold_mb": 500,
"log_interval_seconds": 60
}
# Pytest mark for filtering soak tests
soak = pytest.mark.soak
"""
Marker for soak tests (extended duration tests).
Usage:
@pytest.mark.soak
@pytest.mark.timeout(3600) # 1 hour
def test_memory_leak_detection():
...
Running soak tests:
pytest -m soak -v # All soak tests
pytest -m soak --timeout=7200 # With timeout override
pytest tests/soak/ -v # All tests in soak directory
"""