erp-ass/backend/tests/test_executor.py

"""Tests for Config Executor.

This module tests the ConfigExecutor class for SQL validation and execution.
"""

import pytest
from unittest.mock import MagicMock, patch
from app.core.executor import ConfigExecutor


def test_executor_init():
    """Test executor initialization."""
    executor = ConfigExecutor()
    assert executor.db_engine is not None


def test_validate_sql_safe():
    """Test validation of safe SQL statements."""
    executor = ConfigExecutor()

    # Test SELECT
    is_valid, msg = executor.validate_sql("SELECT * FROM SYS_FORM")
    assert is_valid is True
    assert "验证通过" in msg

    # Test INSERT
    is_valid, msg = executor.validate_sql(
        "INSERT INTO SYS_FORM (IKEY, FORM_NAME) VALUES (1, 'Test')"
    )
    assert is_valid is True

    # Test UPDATE with safe WHERE clause
    is_valid, msg = executor.validate_sql(
        "UPDATE SYS_FORM SET FORM_NAME = 'Test' WHERE IKEY = 1"
    )
    assert is_valid is True


def test_validate_sql_dangerous():
    """Test validation catches dangerous SQL statements."""
    executor = ConfigExecutor()

    # Test DROP DATABASE
    is_valid, msg = executor.validate_sql("DROP DATABASE test_db")
    assert is_valid is False
    assert "危险操作" in msg

    # Test DROP TABLE
    is_valid, msg = executor.validate_sql("DROP TABLE users")
    assert is_valid is False
    assert "危险操作" in msg

    # Test TRUNCATE
    is_valid, msg = executor.validate_sql("TRUNCATE TABLE important_data")
    assert is_valid is False
    assert "危险操作" in msg

    # Test DELETE without WHERE
    is_valid, msg = executor.validate_sql("DELETE FROM users")
    assert is_valid is False
    assert "危险操作" in msg


def test_execute_config_success():
    """Test successful execution of SQL list."""
    with patch('app.core.executor.DatabaseEngine') as MockDBEngine:
        mock_db_engine = MagicMock()
        mock_db_engine.execute_transaction = MagicMock(return_value=True)
        MockDBEngine.return_value = mock_db_engine

        executor = ConfigExecutor()

        sql_list = [
            "INSERT INTO SYS_FORM (IKEY, FORM_NAME) VALUES (1, 'Test1')",
            "INSERT INTO SYS_FORM (IKEY, FORM_NAME) VALUES (2, 'Test2')"
        ]

        result = executor.execute_config(sql_list, session_id="test-session")

        assert result["success"] is True
        assert len(result["executed"]) == 2
        assert result["failed"] is None
        assert "成功执行" in result["message"]


def test_execute_config_validation_failure():
    """Test execution fails when SQL validation fails."""
    executor = ConfigExecutor()

    sql_list = [
        "SELECT * FROM SYS_FORM",
        "DROP DATABASE test"  # Dangerous SQL
    ]

    result = executor.execute_config(sql_list, session_id="test-session")

    assert result["success"] is False
    assert result["failed"] is not None
    assert "验证失败" in result["message"]
    assert len(result["executed"]) == 0


def test_execute_config_execution_failure():
    """Test execution handles database errors."""
    with patch('app.core.executor.DatabaseEngine') as MockDBEngine:
        mock_db_engine = MagicMock()
        mock_db_engine.execute_transaction = MagicMock(
            side_effect=Exception("Database connection error")
        )
        MockDBEngine.return_value = mock_db_engine

        executor = ConfigExecutor()

        sql_list = ["SELECT * FROM SYS_FORM"]

        result = executor.execute_config(sql_list, session_id="test-session")

        assert result["success"] is False
        assert result["failed"] is not None
        assert "执行失败" in result["message"]


def test_rollback_placeholder():
    """Test rollback functionality placeholder."""
    executor = ConfigExecutor()

    result = executor.rollback(session_id="test-session")

    assert result["success"] is False
    assert "待实现" in result["message"]


def test_dangerous_keywords_exist():
    """Test that dangerous keywords list is properly defined."""
    executor = ConfigExecutor()

    assert hasattr(executor, 'DANGEROUS_KEYWORDS')
    assert len(executor.DANGEROUS_KEYWORDS) > 0
    assert "DROP DATABASE" in executor.DANGEROUS_KEYWORDS
    assert "DROP TABLE" in executor.DANGEROUS_KEYWORDS
    assert "TRUNCATE TABLE" in executor.DANGEROUS_KEYWORDS