""" 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" )