tests/soak/test_connection_pool_stability.py

"""
Soak Tests for Database Connection Pool Stability

Extended duration tests (2 hours) to detect connection pool exhaustion and leaks:
- Connection pool size stability under load
- Connection return validation (all connections checked in)
- Connection leak detection (unclosed connections)
- Pool growth monitoring (shouldn't grow unbounded)

Connection pool issues detected:
- Connections not returned to pool (leaks)
- Pool exhaustion (all connections checked out)
- Pool growing unbounded (connections created but not reused)

Tests use SQLAlchemy engine.pool.status() to monitor pool health.
Only applicable to QueuePool (SQLite uses StaticPool, skip if not QueuePool).
"""

import gc
import pytest
import time
from sqlalchemy.pool import QueuePool

from core.models import AgentRegistry


@pytest.mark.soak
@pytest.mark.timeout(7200)  # 2 hours
def test_connection_pool_no_leaks_2hr(db_session, memory_monitor):
    """
    Soak test for database connection pool stability (2 hours).

    Validates:
    - All connections are returned to pool after operations
    - Pool size doesn't grow >10 from initial (no connection leaks)
    - Pool remains stable under extended load
    - No connection exhaustion errors

    Test pattern:
    - Run for 2 hours (7200 seconds)
    - Perform 100 database operations per iteration
    - Each iteration: create agent, query agent, delete agent
    - Log pool status every 300 iterations (~5 minutes)
    - Final validation: all connections checked in, pool size stable

    Duration: 2 hours
    Pool threshold: Size growth < 10 connections
    Failure: Pool grown > 10 or connections not checked in

    Note: Skips if not using QueuePool (SQLite uses StaticPool).
    """
    from core.database import engine

    # Check if using QueuePool (SQLite uses StaticPool, skip test)
    if not isinstance(engine.pool, QueuePool):
        pytest.skip("Connection pool test requires QueuePool (SQLite uses StaticPool)")

    initial_size = engine.pool.size()
    initial_checked_in = engine.pool.checkedin()

    print(f"\n🔍 Initial pool state: size={initial_size}, checked_in={initial_checked_in}")

    start_time = time.time()
    iterations = 0

    # Run database operations for 2 hours
    while time.time() - start_time < 7200:
        # Perform 100 database operations per iteration
        for i in range(100):
            try:
                # Create agent
                agent = AgentRegistry(
                    id=f"soak_test_agent_{iterations}_{i}",
                    name=f"Soak Test Agent {iterations}_{i}",
                    category="test",
                    module_path="test.module",
                    class_name="TestClass",
                    status="AUTONOMOUS",
                    confidence_score=0.9
                )
                db_session.add(agent)
                db_session.commit()

                # Query agent
                agent = db_session.query(AgentRegistry).filter(
                    AgentRegistry.id == f"soak_test_agent_{iterations}_{i}"
                ).first()

                # Delete agent (cleanup)
                db_session.query(AgentRegistry).filter(
                    AgentRegistry.id == f"soak_test_agent_{iterations}_{i}"
                ).delete(synchronize_session=False)
                db_session.commit()

            except Exception as e:
                db_session.rollback()
                pytest.fail(f"Database operation failed at iteration {iterations}: {e}")

        iterations += 1

        # Force garbage collection every 10 iterations
        if iterations % 10 == 0:
            gc.collect()

        # Log pool status every 300 iterations (~5 minutes)
        if iterations % 300 == 0:
            current_size = engine.pool.size()
            current_checked_in = engine.pool.checkedin()

            print(
                f"Iteration {iterations}: "
                f"Pool size={current_size}, "
                f"Checked in={current_checked_in}"
            )

            # Fail if pool grown > 10 (indicates connection leak)
            if current_size > initial_size + 10:
                pytest.fail(
                    f"Connection pool grown significantly: "
                    f"{initial_size} -> {current_size} "
                    f"(threshold: +10)"
                )

    # Final validation: all connections should be checked in
    final_size = engine.pool.size()
    final_checked_in = engine.pool.checkedin()

    print(f"\n🔍 Final pool state: size={final_size}, checked_in={final_checked_in}")

    # Validate pool size hasn't grown > 10
    assert final_size <= initial_size + 10, (
        f"Connection pool grown: {initial_size} -> {final_size} "
        f"(threshold: +10)"
    )

    # Validate all connections are checked in (±1 for timing)
    # Allow 1 connection variance (test cleanup may have 1 connection open)
    checked_in_tolerance = 1
    assert abs(final_checked_in - initial_checked_in) <= checked_in_tolerance, (
        f"Connection leak detected: checked_in changed from {initial_checked_in} "
        f"to {final_checked_in} (tolerance: ±{checked_in_tolerance})"
    )

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


@pytest.mark.soak
@pytest.mark.timeout(3600)  # 1 hour
def test_connection_pool_rapid_operations_1hr(db_session):
    """
    Soak test for rapid connection pool operations (1 hour).

    Validates:
    - Pool handles rapid connection open/close cycles
    - No connection exhaustion under high-frequency operations
    - Pool performance remains stable

    Test pattern:
    - Run for 1 hour (3600 seconds)
    - Perform 500 rapid operations per iteration
    - Each operation: single query
    - Log pool status every 60 iterations (~1 minute)
    - Final validation: pool size stable

    Duration: 1 hour
    Purpose: Detect pool performance degradation under rapid operations

    Note: Skips if not using QueuePool.
    """
    from core.database import engine

    # Check if using QueuePool
    if not isinstance(engine.pool, QueuePool):
        pytest.skip("Connection pool test requires QueuePool (SQLite uses StaticPool)")

    initial_size = engine.pool.size()

    start_time = time.time()
    iterations = 0

    # Run rapid operations for 1 hour
    while time.time() - start_time < 3600:
        # Perform 500 rapid database operations per iteration
        for i in range(500):
            try:
                # Simple query operation
                _ = db_session.query(AgentRegistry).limit(1).first()
            except Exception as e:
                pytest.fail(f"Database query failed at iteration {iterations}: {e}")

        iterations += 1

        # Log pool status every 60 iterations (~1 minute)
        if iterations % 60 == 0:
            current_size = engine.pool.size()
            checked_in = engine.pool.checkedin()

            print(
                f"Iteration {iterations}: "
                f"Pool size={current_size}, "
                f"Checked in={checked_in}"
            )

    # Final validation
    final_size = engine.pool.size()

    # Pool size should be stable
    assert final_size <= initial_size + 10, (
        f"Connection pool grown: {initial_size} -> {final_size} "
        f"(threshold: +10)"
    )

    print(f"\n✅ Soak test complete: {iterations} iterations, pool performance stable")