everything-is-suitable/everything-is-suitable-api/api-test-tool/tests/unit/test_test_orchestrator.py

"""测试编排器单元测试"""

import pytest
from pathlib import Path
from unittest.mock import Mock, MagicMock, patch
from apitest.orchestrator.test_orchestrator import TestOrchestrator
from apitest.models.test_models import TestCase, TestSuiteResult, HTTPMethod
from apitest.models.exceptions import TestRunException


@pytest.fixture
def mock_config_manager():
    """创建模拟配置管理器"""
    config_manager = Mock()
    config_manager.get_base_url.return_value = "http://localhost:8080"
    config_manager.get_timeout.return_value = 30
    config_manager.get_log_level.return_value = "INFO"
    config_manager.get_log_format.return_value = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    return config_manager


@pytest.fixture
def mock_logger_manager():
    """创建模拟日志管理器"""
    logger_manager = Mock()
    logger = Mock()
    logger_manager.get_logger.return_value = logger
    return logger_manager


@pytest.fixture
def test_orchestrator(mock_config_manager, mock_logger_manager):
    """创建测试编排器实例"""
    return TestOrchestrator(
        config_manager=mock_config_manager,
        logger_manager=mock_logger_manager
    )


@pytest.fixture
def sample_test_cases():
    """创建示例测试用例"""
    return [
        TestCase(
            id="TC001",
            name="测试用例1",
            description="测试用例1",
            module="test",
            endpoint="/api/test1",
            method=HTTPMethod.GET,
            headers={},
            enabled=True
        ),
        TestCase(
            id="TC002",
            name="测试用例2",
            description="测试用例2",
            module="test",
            endpoint="/api/test2",
            method=HTTPMethod.POST,
            headers={},
            enabled=True
        )
    ]


def test_init_test_orchestrator(mock_config_manager, mock_logger_manager):
    """测试初始化测试编排器"""
    orchestrator = TestOrchestrator(
        config_manager=mock_config_manager,
        logger_manager=mock_logger_manager
    )
    
    assert orchestrator.config_manager == mock_config_manager
    assert orchestrator.logger_manager == mock_logger_manager
    assert orchestrator.api_client is not None
    assert orchestrator.auth_manager is not None
    assert orchestrator.validation_engine is not None
    assert orchestrator.test_engine is not None
    assert orchestrator.report_manager is not None
    assert orchestrator.logger is not None


def test_init_test_orchestrator_without_managers():
    """测试初始化测试编排器（不提供管理器）"""
    mock_logger = Mock()
    orchestrator = TestOrchestrator(logger=mock_logger)
    
    assert orchestrator.config_manager is not None
    assert orchestrator.logger_manager is not None
    assert orchestrator.api_client is not None
    assert orchestrator.auth_manager is not None
    assert orchestrator.validation_engine is not None
    assert orchestrator.test_engine is not None
    assert orchestrator.report_manager is not None
    assert orchestrator.logger is not None


def test_load_test_cases_success(test_orchestrator, tmp_path):
    """测试加载测试用例（成功）"""
    import json
    
    test_data = [
        {
            "id": "TC001",
            "name": "测试用例1",
            "description": "测试用例1",
            "module": "test",
            "endpoint": "/api/test1",
            "method": "GET",
            "headers": {},
            "enabled": True
        },
        {
            "id": "TC002",
            "name": "测试用例2",
            "description": "测试用例2",
            "module": "test",
            "endpoint": "/api/test2",
            "method": "POST",
            "headers": {},
            "enabled": True
        }
    ]
    
    test_file = tmp_path / "test_cases.json"
    with open(test_file, "w", encoding="utf-8") as f:
        json.dump(test_data, f)
    
    test_cases = test_orchestrator.load_test_cases(test_file)
    
    assert len(test_cases) == 2
    assert test_cases[0].id == "TC001"
    assert test_cases[1].id == "TC002"


def test_load_test_cases_file_not_found(test_orchestrator, tmp_path):
    """测试加载测试用例（文件不存在）"""
    test_file = tmp_path / "nonexistent.json"
    
    with pytest.raises(TestRunException) as exc_info:
        test_orchestrator.load_test_cases(test_file)
    
    assert "测试用例文件不存在" in str(exc_info.value)


def test_load_test_cases_invalid_json(test_orchestrator, tmp_path):
    """测试加载测试用例（无效JSON）"""
    test_file = tmp_path / "invalid.json"
    with open(test_file, "w", encoding="utf-8") as f:
        f.write("invalid json")
    
    with pytest.raises(TestRunException) as exc_info:
        test_orchestrator.load_test_cases(test_file)
    
    assert "加载测试用例失败" in str(exc_info.value)


def test_load_test_cases_with_all_fields(test_orchestrator, tmp_path):
    """测试加载测试用例（包含所有字段）"""
    import json
    
    test_data = [
        {
            "id": "TC001",
            "name": "测试用例1",
            "description": "测试用例1",
            "module": "test",
            "endpoint": "/api/test1",
            "method": "GET",
            "headers": {"Content-Type": "application/json"},
            "params": {"param1": "value1"},
            "body": {"key": "value"},
            "dependencies": ["TC002"],
            "tags": ["tag1", "tag2"],
            "priority": 1,
            "enabled": True,
            "timeout": 10,
            "validations": [{"type": "status_code", "value": 200}],
            "extract_config": [{"type": "extract", "field": "id", "var_name": "user_id"}]
        }
    ]
    
    test_file = tmp_path / "test_cases.json"
    with open(test_file, "w", encoding="utf-8") as f:
        json.dump(test_data, f)
    
    test_cases = test_orchestrator.load_test_cases(test_file)
    
    assert len(test_cases) == 1
    assert test_cases[0].id == "TC001"
    assert test_cases[0].method == HTTPMethod.GET
    assert test_cases[0].headers == {"Content-Type": "application/json"}
    assert test_cases[0].params == {"param1": "value1"}
    assert test_cases[0].body == {"key": "value"}
    assert test_cases[0].dependencies == ["TC002"]
    assert test_cases[0].tags == ["tag1", "tag2"]
    assert test_cases[0].priority == 1
    assert test_cases[0].enabled == True
    assert test_cases[0].timeout == 10
    assert len(test_cases[0].validations) == 1


def test_run_test_suite_success(test_orchestrator, sample_test_cases, tmp_path):
    """测试运行测试套件（成功）"""
    report_path = tmp_path / "test_report.html"
    
    with patch.object(test_orchestrator.test_engine, 'execute_test_suite') as mock_execute:
        mock_result = Mock(spec=TestSuiteResult)
        mock_result.suite_name = "Test Suite"
        mock_result.total = 2
        mock_result.passed = 2
        mock_result.failed = 0
        mock_result.skipped = 0
        mock_result.pass_rate = 100.0
        mock_result.duration = 1.0
        mock_execute.return_value = mock_result
        
        result = test_orchestrator.run_test_suite(
            sample_test_cases,
            generate_report=False
        )
        
        assert result == mock_result
        mock_execute.assert_called_once_with(sample_test_cases, stop_on_failure=False)


def test_run_test_suite_with_report(test_orchestrator, sample_test_cases, tmp_path):
    """测试运行测试套件（生成报告）"""
    report_path = tmp_path / "test_report.html"
    
    with patch.object(test_orchestrator.test_engine, 'execute_test_suite') as mock_execute, \
         patch.object(test_orchestrator.report_manager, 'generate_html_report') as mock_report:
        mock_result = Mock(spec=TestSuiteResult)
        mock_result.suite_name = "Test Suite"
        mock_result.total = 2
        mock_result.passed = 2
        mock_result.failed = 0
        mock_result.skipped = 0
        mock_result.pass_rate = 100.0
        mock_result.duration = 1.0
        mock_execute.return_value = mock_result
        
        result = test_orchestrator.run_test_suite(
            sample_test_cases,
            generate_report=True,
            report_format="html",
            report_path=report_path
        )
        
        assert result == mock_result
        mock_report.assert_called_once()


def test_run_test_suite_stop_on_failure(test_orchestrator, sample_test_cases):
    """测试运行测试套件（失败时停止）"""
    with patch.object(test_orchestrator.test_engine, 'execute_test_suite') as mock_execute:
        mock_result = Mock(spec=TestSuiteResult)
        mock_result.suite_name = "Test Suite"
        mock_result.total = 2
        mock_result.passed = 1
        mock_result.failed = 1
        mock_result.skipped = 0
        mock_result.pass_rate = 50.0
        mock_result.duration = 0.5
        mock_execute.return_value = mock_result
        
        result = test_orchestrator.run_test_suite(
            sample_test_cases,
            stop_on_failure=True,
            generate_report=False
        )
        
        assert result == mock_result
        mock_execute.assert_called_once_with(sample_test_cases, stop_on_failure=True)


def test_run_test_suite_by_filter_module(test_orchestrator, sample_test_cases):
    """测试按模块过滤运行测试套件"""
    with patch.object(test_orchestrator.test_engine, 'execute_test_cases_by_filter') as mock_execute:
        mock_result = Mock(spec=TestSuiteResult)
        mock_result.suite_name = "Test Suite"
        mock_result.total = 2
        mock_result.passed = 2
        mock_result.failed = 0
        mock_result.skipped = 0
        mock_result.pass_rate = 100.0
        mock_result.duration = 1.0
        mock_execute.return_value = mock_result
        
        result = test_orchestrator.run_test_suite_by_filter(
            sample_test_cases,
            module_filter="test",
            generate_report=False
        )
        
        assert result == mock_result
        mock_execute.assert_called_once_with(
            sample_test_cases,
            module_filter="test",
            tag_filter=None,
            priority_filter=None
        )


def test_run_test_suite_by_filter_tag(test_orchestrator, sample_test_cases):
    """测试按标签过滤运行测试套件"""
    with patch.object(test_orchestrator.test_engine, 'execute_test_cases_by_filter') as mock_execute:
        mock_result = Mock(spec=TestSuiteResult)
        mock_result.suite_name = "Test Suite"
        mock_result.total = 2
        mock_result.passed = 2
        mock_result.failed = 0
        mock_result.skipped = 0
        mock_result.pass_rate = 100.0
        mock_result.duration = 1.0
        mock_execute.return_value = mock_result
        
        result = test_orchestrator.run_test_suite_by_filter(
            sample_test_cases,
            tag_filter=["smoke"],
            generate_report=False
        )
        
        assert result == mock_result
        mock_execute.assert_called_once_with(
            sample_test_cases,
            module_filter=None,
            tag_filter=["smoke"],
            priority_filter=None
        )


def test_run_test_suite_by_filter_priority(test_orchestrator, sample_test_cases):
    """测试按优先级过滤运行测试套件"""
    with patch.object(test_orchestrator.test_engine, 'execute_test_cases_by_filter') as mock_execute:
        mock_result = Mock(spec=TestSuiteResult)
        mock_result.suite_name = "Test Suite"
        mock_result.total = 2
        mock_result.passed = 2
        mock_result.failed = 0
        mock_result.skipped = 0
        mock_result.pass_rate = 100.0
        mock_result.duration = 1.0
        mock_execute.return_value = mock_result
        
        result = test_orchestrator.run_test_suite_by_filter(
            sample_test_cases,
            priority_filter=1,
            generate_report=False
        )
        
        assert result == mock_result
        mock_execute.assert_called_once_with(
            sample_test_cases,
            module_filter=None,
            tag_filter=None,
            priority_filter=1
        )


def test_set_base_url(test_orchestrator):
    """测试设置基础URL"""
    test_orchestrator.set_base_url("http://new-api.example.com")
    
    assert test_orchestrator.api_client.base_url == "http://new-api.example.com"


def test_set_auth_token(test_orchestrator):
    """测试设置认证令牌"""
    test_orchestrator.set_auth_token("test-token-123")
    
    assert test_orchestrator.auth_manager.get_token() == "test-token-123"