| """ |
| 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 |
| } |
|
|
|
|
| @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 |
| """ |
| |
| 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 |
| } |
|
|
|
|
| |
| 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 |
| """ |
|
|