File size: 7,702 Bytes
81e3673 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 | """
Soak Tests for Memory Leak Detection
Extended duration tests (1-2 hours) to detect memory leaks in critical components:
- GovernanceCache: Cache operations under extended load
- EpisodeService: Episode creation and retrieval patterns
- WorkflowEngine: Workflow execution memory patterns
Memory leak detection strategy:
1. Measure initial memory usage
2. Run operations for extended period (1-2 hours)
3. Force garbage collection periodically
4. Log memory growth at regular intervals
5. Fail fast if memory growth exceeds fail-fast threshold
6. Final assertion: memory growth < threshold for duration
Tests use psutil for accurate memory measurement and gc.collect() to
distinguish leaks from cached data.
"""
import gc
import pytest
import time
import psutil
from typing import Dict, Any
from core.governance_cache import GovernanceCache
@pytest.mark.soak
@pytest.mark.timeout(3600) # 1 hour
def test_governance_cache_memory_stability_1hr(
memory_monitor: Dict[str, Any],
enable_gc_control: Dict[str, Any],
soak_test_config: Dict[str, int]
):
"""
Soak test for GovernanceCache memory leak detection (1 hour).
Validates:
- Cache operations don't leak memory under extended load
- Memory growth remains within 100MB threshold over 1 hour
- Cache LRU eviction works correctly (doesn't grow unbounded)
Test pattern:
- Run for 1 hour (3600 seconds)
- Perform 1000 cache operations per iteration (set + get)
- Log memory growth every 60 iterations (~1 minute)
- Force GC every 10 iterations
- Fail fast if memory growth > 500MB
Duration: 1 hour
Memory threshold: < 100MB growth
Failure: Memory growth > 100MB or > 500MB (fail-fast)
"""
process = memory_monitor["process"]
initial_memory = memory_monitor["initial_memory_mb"]
config = soak_test_config
# Create cache with realistic configuration
cache = GovernanceCache(max_size=1000, ttl_seconds=60)
start_time = time.time()
iterations = 0
# Run cache operations for 1 hour
while time.time() - start_time < 3600:
# Perform 1000 cache operations per iteration
for i in range(1000):
cache.set(
agent_id=f"agent_{iterations}_{i}",
action_type="test_action",
data={"allowed": True, "maturity_level": "AUTONOMOUS"}
)
cache.get(agent_id=f"agent_{iterations}_{i}", action_type="test_action")
iterations += 1
# Force garbage collection every 10 iterations
if iterations % 10 == 0:
enable_gc_control["collect"]()
# Log memory growth every 60 iterations (~1 minute)
if iterations % 60 == 0:
current_memory = process.memory_info().rss / 1024 / 1024
memory_growth = current_memory - initial_memory
_log_memory_growth(process, initial_memory, iterations)
# Fail fast if memory growth > 500MB
if memory_growth > config["fail_fast_threshold_mb"]:
pytest.fail(
f"FAIL-FAST: Memory leak detected - {memory_growth:.2f}MB growth "
f"(threshold: {config['fail_fast_threshold_mb']}MB)"
)
# Final memory check
final_memory = process.memory_info().rss / 1024 / 1024
memory_growth = final_memory - initial_memory
# Assert memory growth < 100MB over 1 hour
assert memory_growth < config["memory_threshold_1hr_mb"], (
f"Memory leak detected: {memory_growth:.2f}MB growth over 1 hour "
f"(threshold: {config['memory_threshold_1hr_mb']}MB)"
)
print(f"\n✅ Soak test complete: {iterations} iterations, {memory_growth:.2f}MB memory growth")
@pytest.mark.soak
@pytest.mark.timeout(1800) # 30 minutes
def test_episode_service_memory_stability_30min(
memory_monitor: Dict[str, Any],
enable_gc_control: Dict[str, Any],
soak_test_config: Dict[str, int]
):
"""
Soak test for episode service memory patterns (30 minutes).
Validates:
- Episode creation doesn't leak memory
- Episode metadata tracking remains stable
- Memory growth < 50MB over 30 minutes (shorter test, lower threshold)
Test pattern:
- Run for 30 minutes (1800 seconds)
- Create mock episodes (100 episodes per iteration)
- Track memory growth
- Force GC every 10 iterations
- Log memory every 60 iterations
Duration: 30 minutes
Memory threshold: < 50MB growth
Purpose: Detect memory leaks in episode lifecycle operations
Note: This test uses mock episode creation to avoid database dependency.
Real episode service memory patterns should be tested with integration tests.
"""
process = memory_monitor["process"]
initial_memory = memory_monitor["initial_memory_mb"]
config = soak_test_config
# Simulated episode storage (in-memory for this test)
episodes = []
start_time = time.time()
iterations = 0
# Run episode creation for 30 minutes
while time.time() - start_time < 1800:
# Create 100 mock episodes per iteration
for i in range(100):
episode = {
"id": f"episode_{iterations}_{i}",
"agent_id": f"agent_{iterations}",
"start_time": time.time(),
"segments": [
{"action": "test_action", "timestamp": time.time()}
],
"metadata": {"test": "data"}
}
episodes.append(episode)
# Simulate episode cleanup (LRU eviction)
if len(episodes) > 10000:
episodes = episodes[-5000:] # Keep last 5000
iterations += 1
# Force garbage collection every 10 iterations
if iterations % 10 == 0:
enable_gc_control["collect"]()
# Log memory growth every 60 iterations
if iterations % 60 == 0:
current_memory = process.memory_info().rss / 1024 / 1024
memory_growth = current_memory - initial_memory
_log_memory_growth(process, initial_memory, iterations)
# Fail fast if memory growth > 500MB
if memory_growth > config["fail_fast_threshold_mb"]:
pytest.fail(
f"FAIL-FAST: Memory leak detected - {memory_growth:.2f}MB growth "
f"(threshold: {config['fail_fast_threshold_mb']}MB)"
)
# Final memory check
final_memory = process.memory_info().rss / 1024 / 1024
memory_growth = final_memory - initial_memory
# Assert memory growth < 50MB over 30 minutes (lower threshold for shorter test)
threshold_mb = config["memory_threshold_1hr_mb"] // 2 # 50MB for 30min
assert memory_growth < threshold_mb, (
f"Memory leak detected: {memory_growth:.2f}MB growth over 30 minutes "
f"(threshold: {threshold_mb}MB)"
)
print(f"\n✅ Soak test complete: {iterations} iterations, {memory_growth:.2f}MB memory growth")
def _log_memory_growth(process: psutil.Process, initial_memory_mb: float, iteration: int):
"""
Helper function to log memory growth during soak tests.
Args:
process: psutil.Process instance
initial_memory_mb: Initial memory usage in MB
iteration: Current iteration number
Prints formatted memory information including:
- Current memory usage (MB)
- Memory growth (MB)
- Iteration number
"""
current_memory = process.memory_info().rss / 1024 / 1024
memory_growth = current_memory - initial_memory_mb
print(
f"Iteration {iteration}: "
f"Memory = {current_memory:.2f}MB, "
f"Growth = {memory_growth:+.2f}MB"
)
|