ss-tools/backend/tests/test_db_executor.py

# #region Test.Core.DbExecutor [C:3] [TYPE Module] [SEMANTICS test, database, executor, direct]
# @BRIEF Pytest tests for DbExecutor — direct database INSERT execution.
# @RELATION BINDS_TO -> [DbExecutor]
# @TEST_EDGE: missing_connection -> Returns error "not found"
# @TEST_EDGE: invalid_type -> Unsupported dialect returns error
# @TEST_EDGE: external_fail -> Execution error handled gracefully
# @TEST_CONTRACT DbExecutor.execute_sql routes to correct dialect driver and returns DbExecutionResult.
# @TEST_CONTRACT Connection pool is reused across calls (same connection_id → same pool).
# AUDIT_NOTE: patch.object targets _execute_pg/_execute_ch/_execute_mysql which are
# thin wrappers around [EXT:asyncpg]/[EXT:clickhouse-connect]/[EXT:pymysql] DB drivers.
# These are VALID [EXT:Database] boundary mocks per semantics-testing §3c.

import pytest
from unittest.mock import MagicMock, patch

from src.core.db_executor import DbExecutor, DbExecutionResult
from src.core.config_models import DatabaseConnection


@pytest.fixture
def mock_conn_service():
    cs = MagicMock()
    return cs


@pytest.fixture
def pg_connection():
    return DatabaseConnection(
        id="pg1", name="Test PG", host="localhost", port=5432,
        database="test_db", username="tester", password="secret",
        dialect="postgresql", pool_size=5,
    )


@pytest.fixture
def ch_connection():
    return DatabaseConnection(
        id="ch1", name="Test CH", host="localhost", port=9000,
        database="analytics", username="default", password="",
        dialect="clickhouse", pool_size=2,
    )


@pytest.fixture
def mysql_connection():
    return DatabaseConnection(
        id="my1", name="Test MySQL", host="localhost", port=3306,
        database="test", username="root", password="root",
        dialect="mysql", pool_size=3,
    )


@pytest.fixture
def executor(mock_conn_service):
    return DbExecutor(mock_conn_service)


class TestPostgresql:
    @pytest.mark.asyncio
    async def test_execute_success(self, executor, mock_conn_service, pg_connection):
        mock_conn_service.get_connection.return_value = pg_connection
        sample_sql = 'INSERT INTO "products" ("id", "name") VALUES (\'1\', \'test\')'
        with patch.object(executor, '_execute_pg', return_value=DbExecutionResult(success=True, rows_affected=5, execution_time_ms=120)):
            result = await executor.execute_sql("pg1", sample_sql)
            assert result.success is True
            assert result.rows_affected == 5
            assert result.execution_time_ms == 120

    @pytest.mark.asyncio
    async def test_connection_not_found(self, executor, mock_conn_service):
        mock_conn_service.get_connection.return_value = None
        result = await executor.execute_sql("nonexistent", "SELECT 1")
        assert result.success is False
        assert "not found" in (result.error or "")

    @pytest.mark.asyncio
    async def test_dialect_routing_ch(self, executor, mock_conn_service, ch_connection):
        mock_conn_service.get_connection.return_value = ch_connection
        with patch.object(executor, '_execute_ch', return_value=DbExecutionResult(success=True, rows_affected=3, execution_time_ms=50)):
            result = await executor.execute_sql("ch1", "INSERT INTO t VALUES (1)")
            assert result.success is True

    @pytest.mark.asyncio
    async def test_dialect_routing_mysql(self, executor, mock_conn_service, mysql_connection):
        mock_conn_service.get_connection.return_value = mysql_connection
        with patch.object(executor, '_execute_mysql', return_value=DbExecutionResult(success=True, rows_affected=2, execution_time_ms=30)):
            result = await executor.execute_sql("my1", "INSERT INTO t VALUES (1)")
            assert result.success is True

    @pytest.mark.asyncio
    async def test_unsupported_dialect(self, executor, mock_conn_service):
        bad_conn = DatabaseConnection.model_construct(
            id="bad", name="Bad", host="x", port=1,
            database="x", username="u", password="p", dialect="oracle",
        )
        mock_conn_service.get_connection.return_value = bad_conn
        result = await executor.execute_sql("bad", "SELECT 1")
        assert result.success is False
        assert "Unsupported" in (result.error or "")

    @pytest.mark.asyncio
    async def test_execution_error(self, executor, mock_conn_service, pg_connection):
        mock_conn_service.get_connection.return_value = pg_connection
        with patch.object(executor, '_execute_pg', side_effect=RuntimeError("Connection timeout")):
            result = await executor.execute_sql("pg1", "BAD SQL")
            assert result.success is False
            assert "Connection timeout" in (result.error or "")


class TestPoolCaching:
    def test_pool_cached_by_conn_id(self, executor, mock_conn_service, pg_connection):
        mock_conn_service.get_connection.return_value = pg_connection
        assert "pg1" not in executor._pools
        executor._set_pool("pg1", "pool_obj", "config_key")
        assert executor._get_or_create_pool("pg1", "config_key") == "pool_obj"
        assert executor._get_or_create_pool("pg1", "other_key") is None

    def test_pool_recreated_on_config_change(self, executor, mock_conn_service, pg_connection):
        mock_conn_service.get_connection.return_value = pg_connection
        executor._set_pool("pg1", "old_pool", "old_config")
        result = executor._get_or_create_pool("pg1", "new_config")
        assert result is None


class TestMissingDriver:
    @pytest.mark.asyncio
    async def test_asyncpg_not_installed(self, executor, mock_conn_service, pg_connection):
        mock_conn_service.get_connection.return_value = pg_connection
        with patch.dict('sys.modules', {'asyncpg': None}):
            result = await executor._execute_pg(pg_connection, "SELECT 1", 0)
            assert result.success is False
            assert "asyncpg is not installed" in (result.error or "")

    def test_clickhouse_not_installed(self, executor, mock_conn_service, ch_connection):
        mock_conn_service.get_connection.return_value = ch_connection
        with patch.dict('sys.modules', {'clickhouse_connect': None}):
            result = executor._execute_ch(ch_connection, "SELECT 1", 0)
            assert result.success is False
            assert "clickhouse-connect is not installed" in (result.error or "")
# #endregion Test.Core.DbExecutor