# Tests for SQL executor import pytest from unittest.mock import Mock, patch, MagicMock, PropertyMock from datetime import datetime from sqlalchemy.exc import OperationalError, ProgrammingError from src.database.sql_executor import ( SQLExecutor, QueryResult, get_sql_executor, reset_sql_executor, ) class TestQueryResult: # Tests for QueryResult dataclass def test_successful_result(self): result = QueryResult( success=True, query="SELECT * FROM users", rows=[{"id": 1, "name": "Test"}], columns=["id", "name"], row_count=1, execution_time_ms=50.0, ) assert result.success is True assert result.row_count == 1 assert result.truncated is False assert result.error is None def test_truncated_result(self): result = QueryResult( success=True, query="SELECT * FROM large_table", rows=[{"id": i} for i in range(100)], columns=["id"], row_count=100, original_count=500, truncated=True, execution_time_ms=100.0, ) assert result.truncated is True assert result.row_count == 100 assert result.original_count == 500 def test_error_result(self): result = QueryResult( success=False, query="SELECT * FROM nonexistent", error="Table not found", ) assert result.success is False assert result.error == "Table not found" assert result.row_count == 0 def test_to_dict(self): result = QueryResult( success=True, query="SELECT * FROM users", columns=["id"], rows=[{"id": 1}], row_count=1, execution_time_ms=50.0, ) d = result.to_dict() assert isinstance(d, dict) assert d["success"] is True assert "timestamp" in d def test_get_statistics(self): result = QueryResult( success=True, query="SELECT * FROM products", columns=["id", "price"], rows=[ {"id": 1, "price": 100}, {"id": 2, "price": 200}, {"id": 3, "price": 300}, ], row_count=3, execution_time_ms=50.0, ) stats = result.get_statistics() assert "price" in stats assert stats["price"]["sum"] == 600 assert stats["price"]["avg"] == 200 class TestSQLExecutor: # Tests for SQLExecutor class @pytest.fixture def mock_adapter(self): adapter = Mock() adapter.is_connected = True adapter.execute_query.return_value = [{"id": 1, "name": "Test"}] return adapter @pytest.fixture def executor(self, mock_adapter): return SQLExecutor(adapter=mock_adapter, query_timeout=30, max_rows=1000, truncate_at=100) def test_initialization(self, executor): assert executor is not None assert executor._query_timeout == 30 assert executor._max_rows == 1000 assert executor._truncate_at == 100 def test_execute_simple_query(self, executor, mock_adapter): mock_adapter.execute_query.return_value = [ {"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}, ] result = executor.execute("SELECT id, name FROM users", validate=False) assert result.success is True assert result.row_count == 2 assert result.columns == ["id", "name"] def test_execute_empty_result(self, executor, mock_adapter): mock_adapter.execute_query.return_value = [] result = executor.execute("SELECT * FROM users WHERE id = 999", validate=False) assert result.success is True assert result.row_count == 0 assert result.rows == [] def test_execute_with_parameters(self, executor, mock_adapter): mock_adapter.execute_query.return_value = [{"id": 1, "name": "Alice"}] result = executor.execute( "SELECT * FROM users WHERE id = :user_id", validate=False, params={"user_id": 1} ) assert result.success is True mock_adapter.execute_query.assert_called_once() def test_execute_truncates_large_result(self, mock_adapter): large_result = [{"id": i} for i in range(200)] mock_adapter.execute_query.return_value = large_result executor = SQLExecutor(adapter=mock_adapter, truncate_at=100) result = executor.execute("SELECT * FROM large_table", validate=False) assert result.success is True assert result.row_count == 100 assert result.original_count == 200 assert result.truncated is True def test_execute_records_timing(self, executor, mock_adapter): mock_adapter.execute_query.return_value = [{"id": 1}] result = executor.execute("SELECT * FROM users", validate=False) assert result.execution_time_ms >= 0 assert result.timestamp is not None def test_execute_handles_operational_error(self, executor, mock_adapter): mock_adapter.execute_query.side_effect = OperationalError( "connection failed", None, None ) result = executor.execute("SELECT * FROM users", validate=False) assert result.success is False assert result.error is not None def test_execute_handles_programming_error(self, executor, mock_adapter): mock_adapter.execute_query.side_effect = ProgrammingError( "syntax error", None, None ) result = executor.execute("SELECT * FORM users", validate=False) assert result.success is False assert result.error is not None def test_execute_handles_generic_exception(self, executor, mock_adapter): mock_adapter.execute_query.side_effect = Exception("Unknown error") result = executor.execute("SELECT * FROM users", validate=False) assert result.success is False assert result.error is not None def test_execute_fails_when_not_connected(self, mock_adapter): mock_adapter.is_connected = False executor = SQLExecutor(adapter=mock_adapter) result = executor.execute("SELECT * FROM users", validate=False) assert result.success is False assert "not connected" in result.error.lower() class TestSQLExecutorValidation: # Tests for validation integration @pytest.fixture def mock_adapter(self): adapter = Mock() adapter.is_connected = True adapter.execute_query.return_value = [{"id": 1}] return adapter @pytest.fixture def mock_validator(self): validator = Mock() validation_result = Mock() validation_result.is_valid = True validation_result.errors = [] validation_result.normalized_sql = "SELECT * FROM users LIMIT 100" validator.validate.return_value = validation_result return validator def test_execute_without_validation(self, mock_adapter, mock_validator): executor = SQLExecutor(adapter=mock_adapter) result = executor.execute("SELECT * FROM users") assert result.success is True def test_execute_runs_query_directly(self, mock_adapter, mock_validator): executor = SQLExecutor(adapter=mock_adapter) result = executor.execute("SELECT * FROM users") assert result.success is True mock_adapter.execute_query.assert_called_once() class TestSQLExecutorRetry: # Tests for retry mechanism @pytest.fixture def mock_adapter(self): adapter = Mock() adapter.is_connected = True return adapter def test_execute_with_retry_succeeds_on_first_try(self, mock_adapter): mock_adapter.execute_query.return_value = [{"id": 1}] executor = SQLExecutor(adapter=mock_adapter) result = executor.execute_with_retry("SELECT * FROM users", max_retries=3) assert result.success is True assert mock_adapter.execute_query.call_count == 1 def test_execute_with_retry_succeeds_after_failures(self, mock_adapter): mock_adapter.execute_query.side_effect = [ OperationalError("connection failed", None, None), OperationalError("connection failed", None, None), [{"id": 1}], ] executor = SQLExecutor(adapter=mock_adapter) result = executor.execute_with_retry("SELECT * FROM users", max_retries=3) assert result.success is True assert mock_adapter.execute_query.call_count == 3 def test_execute_with_retry_fails_after_max_retries(self, mock_adapter): mock_adapter.execute_query.side_effect = OperationalError( "connection failed", None, None ) executor = SQLExecutor(adapter=mock_adapter) result = executor.execute_with_retry("SELECT * FROM users", max_retries=2) assert result.success is False class TestSQLExecutorRowCount: # Tests for row count functionality @pytest.fixture def mock_adapter(self): adapter = Mock() adapter.is_connected = True return adapter def test_get_row_count(self, mock_adapter): mock_adapter.execute_query.return_value = [{"count": 100}] executor = SQLExecutor(adapter=mock_adapter) count = executor.get_row_count("users") assert count == 100 def test_get_row_count_error_returns_none(self, mock_adapter): mock_adapter.execute_query.side_effect = Exception("Error") executor = SQLExecutor(adapter=mock_adapter) count = executor.get_row_count("users") assert count is None class TestSQLExecutorConnection: # Tests for connection testing @pytest.fixture def mock_adapter(self): adapter = Mock() adapter.is_connected = True return adapter def test_test_connection_success(self, mock_adapter): mock_adapter.execute_query.return_value = [{"test": 1}] executor = SQLExecutor(adapter=mock_adapter) result = executor.test_connection() assert result is True def test_test_connection_failure(self, mock_adapter): mock_adapter.execute_query.side_effect = Exception("Connection failed") executor = SQLExecutor(adapter=mock_adapter) result = executor.test_connection() assert result is False class TestSQLExecutorSingleton: # Tests for singleton pattern def setup_method(self): reset_sql_executor() def teardown_method(self): reset_sql_executor() def test_singleton_returns_same_instance(self): instance1 = get_sql_executor() instance2 = get_sql_executor() assert instance1 is instance2 def test_reset_clears_instance(self): instance1 = get_sql_executor() reset_sql_executor() instance2 = get_sql_executor() assert instance1 is not instance2 class TestQueryResultConversion: # Tests for result data conversion @pytest.fixture def mock_adapter(self): adapter = Mock() adapter.is_connected = True return adapter def test_converts_rows_to_dicts(self, mock_adapter): mock_adapter.execute_query.return_value = [ {"id": 1, "name": "Alice", "active": True}, {"id": 2, "name": "Bob", "active": False}, ] executor = SQLExecutor(adapter=mock_adapter) result = executor.execute("SELECT * FROM users", validate=False) assert all(isinstance(row, dict) for row in result.rows) assert result.rows[0]["name"] == "Alice" assert result.rows[1]["active"] is False def test_handles_null_values(self, mock_adapter): mock_adapter.execute_query.return_value = [ {"id": 1, "name": None, "email": None}, ] executor = SQLExecutor(adapter=mock_adapter) result = executor.execute("SELECT * FROM users", validate=False) assert result.rows[0]["name"] is None assert result.rows[0]["email"] is None def test_handles_decimal_values(self, mock_adapter): from decimal import Decimal mock_adapter.execute_query.return_value = [ {"id": 1, "price": Decimal("19.99")}, ] executor = SQLExecutor(adapter=mock_adapter) result = executor.execute("SELECT * FROM products", validate=False) assert result.rows[0]["price"] == Decimal("19.99") def test_handles_datetime_values(self, mock_adapter): now = datetime.now() mock_adapter.execute_query.return_value = [ {"id": 1, "created_at": now}, ] executor = SQLExecutor(adapter=mock_adapter) result = executor.execute("SELECT * FROM logs", validate=False) assert result.rows[0]["created_at"] == now class TestErrorSanitization: # Tests for error message sanitization @pytest.fixture def mock_adapter(self): adapter = Mock() adapter.is_connected = True return adapter def test_sanitizes_password_from_error(self, mock_adapter): mock_adapter.execute_query.side_effect = OperationalError( "Failed with password='secretpass'", None, None ) executor = SQLExecutor(adapter=mock_adapter) result = executor.execute("SELECT * FROM users", validate=False) assert result.success is False assert "secretpass" not in result.error def test_limits_error_message_length(self, mock_adapter): long_error = "Error: " + "x" * 500 mock_adapter.execute_query.side_effect = OperationalError(long_error, None, None) executor = SQLExecutor(adapter=mock_adapter) result = executor.execute("SELECT * FROM users", validate=False) assert result.success is False assert len(result.error) <= 210 class TestTimeoutHandling: # Tests for timeout error handling @pytest.fixture def mock_adapter(self): adapter = Mock() adapter.is_connected = True return adapter def test_timeout_error_message(self, mock_adapter): mock_adapter.execute_query.side_effect = OperationalError( "query timeout exceeded", None, None ) executor = SQLExecutor(adapter=mock_adapter) result = executor.execute("SELECT * FROM users", validate=False) assert result.success is False assert "timeout" in result.error.lower() or "narrow" in result.error.lower()