tests/failure_modes/test_database_connection_loss.py

"""
Database Connection Loss Tests

Test how the system handles database connection failures:
- Connection pool exhaustion
- Connection timeout
- Too many connections
- Connection closed mid-transaction (query, commit)
- Deadlock detection and retry
- Connection recovery after failure

Note: SQLite vs PostgreSQL differences documented where relevant.
"""

import pytest
import asyncio
from unittest.mock import patch, MagicMock, AsyncMock
from sqlalchemy.exc import OperationalError, DBAPIError, IntegrityError
from sqlalchemy.orm import Session
from sqlalchemy import text


class TestConnectionPoolExhaustion:
    """Test connection pool exhaustion handling."""

    def test_connection_pool_exhausted(self, mock_pool_exhausted):
        """
        FAILURE MODE: Connection pool exhausted (all connections in use).
        EXPECTED: Timeout error raised, pool recovers, no crash.
        """
        from core.database import SessionLocal

        # Mock connection pool to raise timeout
        # Note: SessionLocal() doesn't raise error until operation is executed
        with mock_pool_exhausted():
            with pytest.raises(OperationalError) as exc_info:
                db = SessionLocal()
                db.execute(text("SELECT 1"))

            # Should mention pool exhaustion or timeout
            error_msg = str(exc_info.value).lower()
            assert "pool" in error_msg or "exhausted" in error_msg or "connection" in error_msg

    def test_connection_timeout(self):
        """
        FAILURE MODE: Database connection times out.
        EXPECTED: OperationalError raised, timeout message present.
        """
        from core.database import SessionLocal

        # Mock connection to timeout
        # Note: Error occurs during query execution, not session creation
        with patch('core.database.engine.connect', side_effect=OperationalError("connection timeout", None, None)):
            with pytest.raises(OperationalError) as exc_info:
                db = SessionLocal()
                db.execute(text("SELECT 1"))

            # Should mention timeout
            assert "timeout" in str(exc_info.value).lower() or "connection" in str(exc_info.value).lower()

    def test_too_many_connections(self):
        """
        FAILURE MODE: Database returns "too many connections" error.
        EXPECTED: Error handled gracefully, system doesn't crash.
        """
        from core.database import SessionLocal

        # Mock too many connections error
        # Note: Error occurs during query execution
        with patch('core.database.engine.connect', side_effect=OperationalError("sorry, too many clients already", None, None)):
            with pytest.raises(OperationalError) as exc_info:
                db = SessionLocal()
                db.execute(text("SELECT 1"))

            # Should mention too many clients/connections
            error_msg = str(exc_info.value).lower()
            assert "too many" in error_msg or "client" in error_msg


class TestConnectionClosedMidTransaction:
    """Test connection closed during transaction handling."""

    def test_connection_closed_during_query(self):
        """
        FAILURE MODE: Connection closed during query execution.
        EXPECTED: DBAPIError raised, rollback called, resources released.
        """
        from core.database import get_db_session

        # Mock session.execute to raise connection closed error
        with patch('sqlalchemy.orm.Session.execute', side_effect=DBAPIError("connection closed", None, None)):
            with pytest.raises((DBAPIError, OperationalError)):
                with get_db_session() as db:
                    db.execute("SELECT * FROM agents")

            # Note: Verifying rollback called depends on implementation
            # The key is that the error is caught and handled

    def test_connection_closed_during_commit(self):
        """
        FAILURE MODE: Connection closed during commit.
        EXPECTED: DBAPIError raised, rollback executed, no partial commit.
        """
        from core.database import get_db_session

        # Mock session.commit to raise connection closed error
        with patch('sqlalchemy.orm.Session.commit', side_effect=DBAPIError("connection closed", None, None)):
            with pytest.raises((DBAPIError, OperationalError)) as exc_info:
                with get_db_session() as db:
                    # Simulate transaction
                    pass
                    db.commit()

            # Should mention connection
            assert "connection" in str(exc_info.value).lower()

            # Verify rollback was called (if implementation supports it)
            # This prevents partial commits

    def test_connection_leak(self):
        """
        FAILURE MODE: Many connections not closed properly (connection leak).
        EXPECTED: Pool recovers, new connections available after cleanup.
        """
        from core.database import SessionLocal

        # Simulate connection leak by creating many sessions
        sessions = []
        try:
            for i in range(10):
                db = SessionLocal()
                sessions.append(db)
        except OperationalError as e:
            # Pool exhausted is expected
            assert "pool" in str(e).lower() or "connection" in str(e).lower()
        finally:
            # Clean up sessions
            for db in sessions:
                try:
                    db.close()
                except:
                    pass

        # After cleanup, should be able to create new connection
        db = SessionLocal()
        assert db is not None
        db.close()


class TestDeadlockHandling:
    """Test deadlock detection and handling."""

    def test_database_deadlock(self, mock_deadlock):
        """
        FAILURE MODE: Database deadlock detected during commit.
        EXPECTED: OperationalError raised, deadlock mentioned, retry attempted or error raised.
        """
        from core.database import get_db_session

        # Mock session.commit to raise deadlock error
        with mock_deadlock():
            with pytest.raises(OperationalError) as exc_info:
                with get_db_session() as db:
                    # Simulate transaction
                    pass
                    db.commit()

            # Should mention deadlock
            assert "deadlock" in str(exc_info.value).lower()

    def test_deadlock_retry_succeeds(self):
        """
        FAILURE MODE: Deadlock occurs, retry succeeds on second attempt.
        EXPECTED: Retry logic works, transaction completes after retry.
        """
        from core.database import get_db_session

        # Mock commit to fail once, then succeed
        call_count = [0]
        def mock_commit_deadlock_then_success(*args, **kwargs):
            call_count[0] += 1
            if call_count[0] == 1:
                raise OperationalError("deadlock detected", None, None)
            # Second attempt succeeds
            return None

        with patch('sqlalchemy.orm.Session.commit', side_effect=mock_commit_deadlock_then_success):
            with get_db_session() as db:
                # First commit fails with deadlock
                with pytest.raises(OperationalError):
                    db.commit()

                # Note: Actual retry logic depends on implementation
                # This test documents the expected behavior

    def test_concurrent_update_deadlock(self):
        """
        FAILURE MODE: Concurrent updates to same row cause deadlock.
        EXPECTED: Deadlock detected, one transaction succeeds, one fails.

        Note: SQLite serializes writes (one-at-a-time), so deadlocks are rare.
          PostgreSQL with MVCC and SERIALIZABLE isolation can have true deadlocks.
        """
        from core.database import get_db_session
        import threading
        import time

        # Simulate concurrent updates
        results = {"success": 0, "failure": 0}
        errors = []

        def update_thread():
            """Thread that attempts concurrent update."""
            try:
                with get_db_session() as db:
                    # Simulate update
                    time.sleep(0.01)
                    # In SQLite, this would serialize, not deadlock
                    # In PostgreSQL with proper isolation, this could deadlock
                    pass
                    db.commit()
                results["success"] += 1
            except OperationalError as e:
                if "deadlock" in str(e).lower():
                    results["failure"] += 1
                    errors.append(str(e))
                else:
                    # Other operational errors are acceptable
                    results["success"] += 1

        # Run concurrent updates
        threads = []
        for _ in range(2):
            t = threading.Thread(target=update_thread)
            threads.append(t)
            t.start()

        for t in threads:
            t.join(timeout=5.0)

        # At least one should succeed (SQLite serialization)
        # Or both fail with deadlock (PostgreSQL with concurrent writes)
        assert results["success"] + results["failure"] == 2


class TestConnectionRecovery:
    """Test system recovery after connection failures."""

    def test_connection_recovers_after_timeout(self):
        """
        FAILURE MODE: Connection times out, then new connection succeeds.
        EXPECTED: System recovers, new connections work after timeout.
        """
        from core.database import SessionLocal

        # Mock timeout on first attempt, success on second
        call_count = [0]
        def mock_connect_then_success(*args, **kwargs):
            call_count[0] += 1
            if call_count[0] == 1:
                raise OperationalError("connection timeout", None, None)
            # Second attempt succeeds
            mock_conn = MagicMock()
            return mock_conn

        with patch('core.database.engine.connect', side_effect=mock_connect_then_success):
            # First attempt times out
            with pytest.raises(OperationalError):
                db = SessionLocal()
                db.execute("SELECT 1")

            # Note: Recovery depends on implementation retry logic
            # This test documents expected behavior

    def test_pool_recovers_after_exhaustion(self):
        """
        FAILURE MODE: Pool exhausted, then connections released.
        EXPECTED: Pool returns to healthy state, new connections available.
        """
        from core.database import SessionLocal

        # Simulate pool exhaustion
        connections = []
        try:
            # Create connections until pool exhausted
            for i in range(100):  # Exceed typical pool size
                db = SessionLocal()
                connections.append(db)
        except OperationalError as e:
            # Pool exhausted
            assert "pool" in str(e).lower() or "exhausted" in str(e).lower()
        finally:
            # Release all connections
            for db in connections:
                try:
                    db.close()
                except:
                    pass

        # Pool should recover, new connection available
        db = SessionLocal()
        assert db is not None
        db.close()

    def test_deadlock_does_not_hang_system(self):
        """
        FAILURE MODE: Deadlock occurs, verify system doesn't hang indefinitely.
        EXPECTED: Deadlock detected and raised, not infinite hang.
        """
        from core.database import get_db_session
        import threading
        import time

        # Mock deadlock that doesn't resolve
        with patch('sqlalchemy.orm.Session.commit', side_effect=OperationalError("deadlock detected", None, None)):
            hang_detected = [False]
            result = [None]

            def transaction_with_timeout():
                def deadlock_transaction():
                    start = time.time()
                    try:
                        with get_db_session() as db:
                            db.commit()
                    except OperationalError:
                        pass  # Expected
                    elapsed = time.time() - start
                    if elapsed > 5.0:  # 5 second timeout
                        hang_detected[0] = True
                    result[0] = elapsed

                thread = threading.Thread(target=deadlock_transaction)
                thread.start()
                thread.join(timeout=5.0)

                if thread.is_alive():
                    hang_detected[0] = True

            transaction_with_timeout()

            # Should not hang (deadlock should be raised quickly)
            assert not hang_detected[0], "Deadlock caused system to hang"


class TestDatabaseFailureEdgeCases:
    """Test edge cases in database failure handling."""

    def test_connection_closed_during_rollback(self):
        """
        FAILURE MODE: Connection closes during rollback.
        EXPECTED: Error handled gracefully, no crash.
        """
        from core.database import get_db_session

        # Mock rollback to raise connection error
        with patch('sqlalchemy.orm.Session.rollback', side_effect=DBAPIError("connection closed", None, None)):
            with pytest.raises((DBAPIError, OperationalError)):
                with get_db_session() as db:
                    # Trigger rollback
                    raise IntegrityError("constraint violation", None, None)

    def test_multiple_connection_failures_in_sequence(self):
        """
        FAILURE MODE: Multiple connection failures occur in sequence.
        EXPECTED: Each failure handled independently, no cascading crashes.
        """
        from core.database import SessionLocal

        # Mock sequence of connection failures
        call_count = [0]
        def mock_sequence_of_failures(*args, **kwargs):
            call_count[0] += 1
            errors = [
                OperationalError("connection timeout", None, None),
                OperationalError("pool exhausted", None, None),
                OperationalError("too many connections", None, None),
            ]
            if call_count[0] <= len(errors):
                raise errors[call_count[0] - 1]
            # Eventually succeed
            return MagicMock()

        with patch('core.database.engine.connect', side_effect=mock_sequence_of_failures):
            # First three attempts fail
            for i in range(3):
                with pytest.raises(OperationalError):
                    db = SessionLocal()
                    db.execute("SELECT 1")

            # Fourth attempt might succeed (depends on retry logic)

    def test_database_unreachable(self):
        """
        FAILURE MODE: Database server completely unreachable.
        EXPECTED: Connection error raised, graceful degradation.
        """
        from core.database import SessionLocal

        # Mock database unreachable
        with patch('core.database.engine.connect', side_effect=OperationalError("could not connect to server: Connection refused", None, None)):
            with pytest.raises(OperationalError) as exc_info:
                db = SessionLocal()
                db.execute("SELECT 1")

            # Should mention connection refused or unreachable
            error_msg = str(exc_info.value).lower()
            assert "connection" in error_msg or "refused" in error_msg or "unreachable" in error_msg

    def test_database_connection_during_migration(self):
        """
        FAILURE MODE: Connection attempted during schema migration.
        EXPECTED: Error handled, clear message about migration in progress.
        """
        from core.database import SessionLocal

        # Mock migration in progress error
        with patch('core.database.engine.connect', side_effect=OperationalError("database is locked (migration in progress)", None, None)):
            with pytest.raises(OperationalError) as exc_info:
                db = SessionLocal()
                db.execute("SELECT 1")

            # Should mention locked or migration
            error_msg = str(exc_info.value).lower()
            assert "locked" in error_msg or "migration" in error_msg


class TestSQLitevsPostgreSQL:
    """Test differences between SQLite and PostgreSQL behavior."""

    def test_sqlite_write_serialization(self):
        """
        SQLite serializes writes (one transaction at a time).
        PostgreSQL allows concurrent writes with MVCC.

        This test documents SQLite behavior.
        """
        from core.database import get_db_session
        import threading

        # SQLite will serialize these writes
        results = []

        def write_thread(value):
            try:
                with get_db_session() as db:
                    # SQLite: writes are serialized
                    # PostgreSQL: writes can be concurrent
                    pass
                    db.commit()
                results.append("success")
            except Exception as e:
                results.append(f"error: {e}")

        threads = []
        for i in range(3):
            t = threading.Thread(target=write_thread, args=(i,))
            threads.append(t)
            t.start()

        for t in threads:
            t.join(timeout=5.0)

        # All should complete (SQLite serializes)
        assert len(results) == 3

    def test_postgresql_mvcc_isolation(self):
        """
        Test PostgreSQL MVCC behavior (documented, not tested).

        PostgreSQL uses MVCC (Multi-Version Concurrency Control):
        - Readers don't block writers
        - Writers don't block readers
        - SERIALIZABLE isolation prevents phantom reads

        SQLite behavior:
        - Readers can block writers (locking)
        - Writes are serialized
        - No true MVCC

        This test documents the expected difference.
        """
        # Note: This is a documentation test
        # Actual PostgreSQL testing would require PostgreSQL connection
        pass