| """ |
| 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 |
|
|
| |
| |
| with mock_pool_exhausted(): |
| with pytest.raises(OperationalError) as exc_info: |
| db = SessionLocal() |
| db.execute(text("SELECT 1")) |
|
|
| |
| 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 |
|
|
| |
| |
| 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")) |
|
|
| |
| 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 |
|
|
| |
| |
| 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")) |
|
|
| |
| 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 |
|
|
| |
| 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") |
|
|
| |
| |
|
|
| 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 |
|
|
| |
| 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: |
| |
| pass |
| db.commit() |
|
|
| |
| assert "connection" in str(exc_info.value).lower() |
|
|
| |
| |
|
|
| 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 |
|
|
| |
| sessions = [] |
| try: |
| for i in range(10): |
| db = SessionLocal() |
| sessions.append(db) |
| except OperationalError as e: |
| |
| assert "pool" in str(e).lower() or "connection" in str(e).lower() |
| finally: |
| |
| for db in sessions: |
| try: |
| db.close() |
| except: |
| pass |
|
|
| |
| 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 |
|
|
| |
| with mock_deadlock(): |
| with pytest.raises(OperationalError) as exc_info: |
| with get_db_session() as db: |
| |
| pass |
| db.commit() |
|
|
| |
| 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 |
|
|
| |
| 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) |
| |
| return None |
|
|
| with patch('sqlalchemy.orm.Session.commit', side_effect=mock_commit_deadlock_then_success): |
| with get_db_session() as db: |
| |
| with pytest.raises(OperationalError): |
| db.commit() |
|
|
| |
| |
|
|
| 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 |
|
|
| |
| results = {"success": 0, "failure": 0} |
| errors = [] |
|
|
| def update_thread(): |
| """Thread that attempts concurrent update.""" |
| try: |
| with get_db_session() as db: |
| |
| time.sleep(0.01) |
| |
| |
| pass |
| db.commit() |
| results["success"] += 1 |
| except OperationalError as e: |
| if "deadlock" in str(e).lower(): |
| results["failure"] += 1 |
| errors.append(str(e)) |
| else: |
| |
| results["success"] += 1 |
|
|
| |
| 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) |
|
|
| |
| |
| 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 |
|
|
| |
| 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) |
| |
| mock_conn = MagicMock() |
| return mock_conn |
|
|
| with patch('core.database.engine.connect', side_effect=mock_connect_then_success): |
| |
| with pytest.raises(OperationalError): |
| db = SessionLocal() |
| db.execute("SELECT 1") |
|
|
| |
| |
|
|
| 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 |
|
|
| |
| connections = [] |
| try: |
| |
| for i in range(100): |
| db = SessionLocal() |
| connections.append(db) |
| except OperationalError as e: |
| |
| assert "pool" in str(e).lower() or "exhausted" in str(e).lower() |
| finally: |
| |
| for db in connections: |
| try: |
| db.close() |
| except: |
| pass |
|
|
| |
| 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 |
|
|
| |
| 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 |
| elapsed = time.time() - start |
| if elapsed > 5.0: |
| 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() |
|
|
| |
| 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 |
|
|
| |
| with patch('sqlalchemy.orm.Session.rollback', side_effect=DBAPIError("connection closed", None, None)): |
| with pytest.raises((DBAPIError, OperationalError)): |
| with get_db_session() as db: |
| |
| 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 |
|
|
| |
| 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] |
| |
| return MagicMock() |
|
|
| with patch('core.database.engine.connect', side_effect=mock_sequence_of_failures): |
| |
| for i in range(3): |
| with pytest.raises(OperationalError): |
| db = SessionLocal() |
| db.execute("SELECT 1") |
|
|
| |
|
|
| def test_database_unreachable(self): |
| """ |
| FAILURE MODE: Database server completely unreachable. |
| EXPECTED: Connection error raised, graceful degradation. |
| """ |
| from core.database import SessionLocal |
|
|
| |
| 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") |
|
|
| |
| 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 |
|
|
| |
| 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") |
|
|
| |
| 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 |
|
|
| |
| results = [] |
|
|
| def write_thread(value): |
| try: |
| with get_db_session() as db: |
| |
| |
| 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) |
|
|
| |
| 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. |
| """ |
| |
| |
| pass |
|
|