tests/soak/test_cache_stability.py

"""
Soak Tests for Cache Stability Under Concurrent Load

Extended duration tests (30 minutes) to validate cache behavior:
- Concurrent operations (10 workers, 10000 ops each)
- Cache consistency (no race conditions, no data loss)
- TTL expiration (entries expire correctly)
- Cache unbounded growth detection

Cache issues detected:
- Race conditions in concurrent get/set operations
- Cache corruption (data loss, incorrect values)
- TTL not working (cache grows unbounded)
- Deadlocks under concurrent load

Tests use ThreadPoolExecutor for realistic concurrent access patterns.
"""

import gc
import pytest
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

from core.governance_cache import GovernanceCache


@pytest.mark.soak
@pytest.mark.timeout(1800)  # 30 minutes
def test_cache_concurrent_operations_30min(
    memory_monitor,
    enable_gc_control,
    soak_test_config
):
    """
    Soak test for cache stability under concurrent operations (30 minutes).

    Validates:
    - Cache remains consistent under concurrent load (10 workers)
    - No race conditions or deadlocks
    - No cache corruption (data loss, incorrect values)
    - Zero cache errors (get returns None after set)

    Test pattern:
    - Run for 30 minutes (1800 seconds)
    - Use ThreadPoolExecutor with 10 workers
    - Each worker performs 10000 cache operations (set + get)
    - Validate cache consistency: if get returns None after set, fail
    - Log worker completions
    - Force GC every 10 iterations

    Duration: 30 minutes
    Workers: 10 concurrent threads
    Operations per worker: 10000 (set + get)
    Failure: Any cache inconsistency detected

    Race conditions detected:
    - Get returns None after set (data loss)
    - Incorrect value returned (corruption)
    - Deadlock (test hangs)
    """
    cache = GovernanceCache(max_size=10000, ttl_seconds=60)
    process = memory_monitor["process"]
    initial_memory = memory_monitor["initial_memory_mb"]

    errors = []
    iterations = 0

    def worker_operations(worker_id: int):
        """
        Worker function for concurrent cache operations.

        Each worker performs 10000 cache operations:
        - Set key with unique worker_id to avoid conflicts
        - Get key immediately after set
        - Validate value is not None (cache consistency check)

        Raises AssertionError if cache inconsistency detected.
        """
        for i in range(10000):
            # Concurrent write with unique key (worker_id avoids conflicts)
            cache.set(
                agent_id=f"agent_{worker_id}_{i}",
                action_type="test_action",
                data={"allowed": True, "maturity_level": "AUTONOMOUS"}
            )

            # Concurrent read
            result = cache.get(
                agent_id=f"agent_{worker_id}_{i}",
                action_type="test_action"
            )

            # Validate cache consistency (get should not return None after set)
            if result is None:
                raise AssertionError(
                    f"Cache inconsistency: key agent_{worker_id}_{i} not found after write"
                )

        return f"Worker {worker_id} completed"

    start_time = time.time()

    # Run concurrent workers for 30 minutes
    while time.time() - start_time < 1800:
        # Submit 10 concurrent workers
        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = [
                executor.submit(worker_operations, worker_id)
                for worker_id in range(10)
            ]

            # Wait for all workers to complete, collect errors
            for future in as_completed(futures):
                try:
                    result = future.result()
                    print(result)
                except AssertionError as e:
                    errors.append(str(e))
                except Exception as e:
                    errors.append(f"Unexpected error: {e}")

        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

            print(
                f"Iteration {iterations}: "
                f"Memory growth = {memory_growth:.2f}MB, "
                f"Errors = {len(errors)}"
            )

            # Fail fast if memory growth > 500MB
            if memory_growth > soak_test_config["fail_fast_threshold_mb"]:
                pytest.fail(
                    f"FAIL-FAST: Memory leak detected - {memory_growth:.2f}MB growth "
                    f"(threshold: {soak_test_config['fail_fast_threshold_mb']}MB)"
                )

    # Final validation: zero cache errors
    assert len(errors) == 0, (
        f"Cache inconsistencies detected: {len(errors)} errors\n"
        f"Errors: {errors[:5]}"  # Show first 5 errors
    )

    print(f"\n✅ Soak test complete: {iterations} iterations, zero cache errors")


@pytest.mark.soak
@pytest.mark.timeout(900)  # 15 minutes
def test_cache_ttl_expiration_15min(memory_monitor, enable_gc_control):
    """
    Soak test for cache TTL expiration (15 minutes).

    Validates:
    - Cache entries expire after TTL (60 seconds)
    - Cache doesn't grow unbounded (TTL prevents memory leaks)
    - Expired entries are not returned (get returns None)

    Test pattern:
    - Create cache with TTL=60 seconds
    - Add 1000 entries per iteration
    - Validate entries exist immediately after adding
    - Wait 70 seconds, validate entries are gone (expired)
    - Repeat for 15 minutes
    - Monitor memory usage

    Duration: 15 minutes
    TTL: 60 seconds
    Validation: Entries expire correctly, cache doesn't grow unbounded

    Purpose: Detect TTL bugs that cause unbounded cache growth.
    """
    cache = GovernanceCache(max_size=10000, ttl_seconds=60)
    process = memory_monitor["process"]
    initial_memory = memory_monitor["initial_memory_mb"]

    iterations = 0

    start_time = time.time()

    # Run TTL validation for 15 minutes
    while time.time() - start_time < 900:
        # Add 1000 entries
        for i in range(1000):
            cache.set(
                agent_id=f"agent_{iterations}_{i}",
                action_type="test_action",
                data={"allowed": True, "maturity_level": "AUTONOMOUS"}
            )

        # Validate entries exist immediately
        for i in range(100):  # Check 100 random entries
            result = cache.get(
                agent_id=f"agent_{iterations}_{i}",
                action_type="test_action"
            )
            assert result is not None, f"Entry not found immediately after set: agent_{iterations}_{i}"

        print(f"Iteration {iterations}: Added 1000 entries, validated existence")

        # Wait 70 seconds (TTL=60, +10 buffer)
        time.sleep(70)

        # Validate entries are expired (get returns None)
        expired_count = 0
        for i in range(100):  # Check 100 random entries
            result = cache.get(
                agent_id=f"agent_{iterations}_{i}",
                action_type="test_action"
            )
            if result is None:
                expired_count += 1

        # All entries should be expired
        assert expired_count == 100, (
            f"TTL expiration failed: {expired_count}/100 entries expired "
            f"(expected: 100)"
        )

        print(f"Iteration {iterations}: All entries expired correctly (100/100)")

        iterations += 1

        # Force garbage collection
        enable_gc_control["collect"]()

        # Log memory growth
        current_memory = process.memory_info().rss / 1024 / 1024
        memory_growth = current_memory - initial_memory

        print(f"Iteration {iterations}: Memory growth = {memory_growth:.2f}MB")

        # Fail fast if memory growth > 500MB (indicates TTL not working)
        if memory_growth > 500:
            pytest.fail(
                f"TTL expiration not working: cache grew {memory_growth:.2f}MB "
                f"(entries should be expiring)"
            )

    # Final validation: memory growth should be reasonable (< 200MB for 15min)
    final_memory = process.memory_info().rss / 1024 / 1024
    memory_growth = final_memory - initial_memory

    assert memory_growth < 200, (
        f"Cache grew unbounded: {memory_growth:.2f}MB growth over 15 minutes "
        f"(TTL expiration may not be working)"
    )

    print(f"\n✅ Soak test complete: {iterations} iterations, TTL expiration working correctly")


@pytest.mark.soak
@pytest.mark.timeout(1800)  # 30 minutes
def test_cache_lru_eviction_stability_30min(memory_monitor, enable_gc_control):
    """
    Soak test for cache LRU eviction stability (30 minutes).

    Validates:
    - LRU eviction works correctly (cache doesn't exceed max_size)
    - Evicted entries are not returned (get returns None)
    - Cache performance remains stable under load

    Test pattern:
    - Create cache with max_size=1000
    - Add 2000 entries (exceeds max_size, triggers LRU eviction)
    - Validate cache size <= max_size
    - Verify oldest entries are evicted (get returns None)
    - Repeat for 30 minutes

    Duration: 30 minutes
    Max size: 1000 entries
    Validation: LRU eviction works, cache size stable

    Purpose: Detect LRU eviction bugs that cause unbounded cache growth.
    """
    cache = GovernanceCache(max_size=1000, ttl_seconds=60)
    process = memory_monitor["process"]
    initial_memory = memory_monitor["initial_memory_mb"]

    iterations = 0

    start_time = time.time()

    # Run LRU eviction validation for 30 minutes
    while time.time() - start_time < 1800:
        # Add 2000 entries (exceeds max_size=1000)
        for i in range(2000):
            cache.set(
                agent_id=f"agent_{iterations}_{i}",
                action_type="test_action",
                data={"allowed": True, "maturity_level": "AUTONOMOUS"}
            )

        # Validate oldest entries are evicted (first 1000 should be gone)
        evicted_count = 0
        for i in range(1000):  # Check oldest 1000 entries
            result = cache.get(
                agent_id=f"agent_{iterations}_{i}",
                action_type="test_action"
            )
            if result is None:
                evicted_count += 1

        # At least some entries should be evicted (LRU working)
        # Not all 1000 will be evicted (TTL may not have expired yet)
        # But at least 500 should be evicted due to LRU
        assert evicted_count >= 500, (
            f"LRU eviction not working: only {evicted_count}/1000 entries evicted "
            f"(expected: >= 500)"
        )

        print(f"Iteration {iterations}: LRU eviction working ({evicted_count}/1000 evicted)")

        iterations += 1

        # Force garbage collection
        enable_gc_control["collect"]()

        # Log memory growth
        current_memory = process.memory_info().rss / 1024 / 1024
        memory_growth = current_memory - initial_memory

        print(f"Iteration {iterations}: Memory growth = {memory_growth:.2f}MB")

        # Fail fast if memory growth > 500MB (indicates LRU not working)
        if memory_growth > 500:
            pytest.fail(
                f"LRU eviction not working: cache grew {memory_growth:.2f}MB "
                f"(entries should be evicted)"
            )

    # Final validation: memory growth should be reasonable
    final_memory = process.memory_info().rss / 1024 / 1024
    memory_growth = final_memory - initial_memory

    assert memory_growth < 200, (
        f"Cache grew unbounded: {memory_growth:.2f}MB growth over 30 minutes "
        f"(LRU eviction may not be working)"
    )

    print(f"\n✅ Soak test complete: {iterations} iterations, LRU eviction working correctly")