"""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