""" 语义逻辑审查模块单元测试 测试 semantic_logic.py 中的 SemanticLogicReviewer 类 """ import pytest import asyncio import sys import os from unittest.mock import Mock, patch, AsyncMock, MagicMock from typing import Dict, Any # 添加项目根目录到路径 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from core.construction_review.component.reviewers.semantic_logic import ( SemanticLogicReviewer, semantic_logic_reviewer, SEMANTIC_LOGIC_MODEL_CONFIG ) from core.construction_review.component.reviewers.base_reviewer import ReviewResult class TestSemanticLogicReviewer: """语义逻辑审查器测试类""" @pytest.fixture def reviewer(self): """创建审查器实例""" return SemanticLogicReviewer() @pytest.fixture def mock_state(self): """创建模拟的状态字典""" mock_progress_manager = AsyncMock() mock_progress_manager.update_stage_progress = AsyncMock() return { "progress_manager": mock_progress_manager, "callback_task_id": "test_task_123" } @pytest.fixture def sample_review_content(self): """示例审查内容""" return """ 施工方案概述： 本工程为高速公路桥梁施工项目，主要包括桥墩基础施工、桥梁上部结构施工等内容。 施工工期为12个月，计划2024年3月开工，2025年3月竣工。 """ @pytest.fixture def sample_review_references(self): """示例审查参考""" return "参考标准：《公路桥涵施工技术规范》JTG/T 3650-2020" def test_reviewer_initialization(self, reviewer): """测试审查器初始化""" assert reviewer is not None assert reviewer.model == SEMANTIC_LOGIC_MODEL_CONFIG["model"] assert reviewer.temperature == SEMANTIC_LOGIC_MODEL_CONFIG["temperature"] assert reviewer.max_tokens == SEMANTIC_LOGIC_MODEL_CONFIG["max_tokens"] assert reviewer.client is not None def test_global_singleton_instance(self): """测试全局单例实例""" assert semantic_logic_reviewer is not None assert isinstance(semantic_logic_reviewer, SemanticLogicReviewer) def test_model_config(self): """测试模型配置""" assert SEMANTIC_LOGIC_MODEL_CONFIG["base_url"] == "http://192.168.91.253:8003/v1" assert SEMANTIC_LOGIC_MODEL_CONFIG["api_key"] == "sk-123456" assert SEMANTIC_LOGIC_MODEL_CONFIG["model"] == "qwen3-30b" assert SEMANTIC_LOGIC_MODEL_CONFIG["temperature"] == 0.7 assert SEMANTIC_LOGIC_MODEL_CONFIG["max_tokens"] == 2000 @pytest.mark.asyncio async def test_check_semantic_logic_success( self, reviewer, sample_review_content, sample_review_references, mock_state ): """测试语义逻辑检查成功场景""" # 模拟 OpenAI API 响应 mock_response = MagicMock() mock_response.choices = [MagicMock()] mock_response.choices[0].message.content = "审查结果：内容逻辑清晰，无明显问题。" # 模拟提示词模板 mock_prompt_template = MagicMock() mock_message = MagicMock() mock_message.type = "user" mock_message.content = "请审查以下内容" mock_prompt_template.format_messages.return_value = [mock_message] with patch.object(reviewer.client.chat.completions, 'create', new_callable=AsyncMock) as mock_create, \ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt: mock_create.return_value = mock_response mock_get_prompt.return_value = mock_prompt_template # 执行测试 result = await reviewer.check_semantic_logic( trace_id="test_trace_001", review_content=sample_review_content, review_references=sample_review_references, review_location_label="第一章", state=mock_state, stage_name="basic_check" ) # 验证结果 assert isinstance(result, ReviewResult) assert result.success is True assert result.details["name"] == "semantic_logic_check" assert "审查结果" in result.details["response"] assert result.error_message is None assert result.execution_time is not None assert result.execution_time > 0 # 验证 API 调用 mock_create.assert_called_once() call_kwargs = mock_create.call_args.kwargs assert call_kwargs["model"] == "qwen3-30b" assert call_kwargs["temperature"] == 0.7 assert call_kwargs["max_tokens"] == 2000 # 验证进度管理器被调用 mock_state["progress_manager"].update_stage_progress.assert_called() @pytest.mark.asyncio async def test_check_semantic_logic_without_state( self, reviewer, sample_review_content ): """测试没有状态字典的语义逻辑检查""" mock_response = MagicMock() mock_response.choices = [MagicMock()] mock_response.choices[0].message.content = "审查通过" mock_prompt_template = MagicMock() mock_message = MagicMock() mock_message.type = "system" mock_message.content = "系统提示" mock_prompt_template.format_messages.return_value = [mock_message] with patch.object(reviewer.client.chat.completions, 'create', new_callable=AsyncMock) as mock_create, \ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt: mock_create.return_value = mock_response mock_get_prompt.return_value = mock_prompt_template result = await reviewer.check_semantic_logic( trace_id="test_trace_002", review_content=sample_review_content, state=None, stage_name=None ) assert result.success is True assert result.details["name"] == "semantic_logic_check" @pytest.mark.asyncio async def test_check_semantic_logic_api_error( self, reviewer, sample_review_content, mock_state ): """测试 API 调用失败场景""" mock_prompt_template = MagicMock() mock_message = MagicMock() mock_message.type = "user" mock_message.content = "测试内容" mock_prompt_template.format_messages.return_value = [mock_message] with patch.object(reviewer.client.chat.completions, 'create', new_callable=AsyncMock) as mock_create, \ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt: mock_create.side_effect = Exception("API连接失败") mock_get_prompt.return_value = mock_prompt_template result = await reviewer.check_semantic_logic( trace_id="test_trace_003", review_content=sample_review_content, state=mock_state, stage_name="basic_check" ) # 验证错误处理 assert isinstance(result, ReviewResult) assert result.success is False assert result.error_message is not None assert "API连接失败" in result.error_message assert result.execution_time is not None # 验证错误通知被发送 mock_state["progress_manager"].update_stage_progress.assert_called() @pytest.mark.asyncio async def test_check_semantic_logic_empty_content( self, reviewer, mock_state ): """测试空内容场景""" mock_response = MagicMock() mock_response.choices = [MagicMock()] mock_response.choices[0].message.content = "内容为空，无法审查" mock_prompt_template = MagicMock() mock_message = MagicMock() mock_message.type = "user" mock_message.content = "" mock_prompt_template.format_messages.return_value = [mock_message] with patch.object(reviewer.client.chat.completions, 'create', new_callable=AsyncMock) as mock_create, \ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt: mock_create.return_value = mock_response mock_get_prompt.return_value = mock_prompt_template result = await reviewer.check_semantic_logic( trace_id="test_trace_004", review_content="", state=mock_state, stage_name="basic_check" ) assert result.success is True assert result.details["name"] == "semantic_logic_check" @pytest.mark.asyncio async def test_check_semantic_logic_with_references( self, reviewer, sample_review_content, sample_review_references ): """测试带参考信息的语义逻辑检查""" mock_response = MagicMock() mock_response.choices = [MagicMock()] mock_response.choices[0].message.content = "根据参考标准，内容符合要求" mock_prompt_template = MagicMock() mock_message = MagicMock() mock_message.type = "user" mock_message.content = "审查内容" mock_prompt_template.format_messages.return_value = [mock_message] with patch.object(reviewer.client.chat.completions, 'create', new_callable=AsyncMock) as mock_create, \ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt: mock_create.return_value = mock_response mock_get_prompt.return_value = mock_prompt_template result = await reviewer.check_semantic_logic( trace_id="test_trace_005", review_content=sample_review_content, review_references=sample_review_references ) assert result.success is True assert "参考标准" in result.details["response"] # 验证提示词模板被正确调用 mock_get_prompt.assert_called_once() call_kwargs = mock_get_prompt.call_args.kwargs assert call_kwargs["review_content"] == sample_review_content assert call_kwargs["review_references"] == sample_review_references @pytest.mark.asyncio async def test_message_format_conversion(self, reviewer, sample_review_content): """测试消息格式转换""" mock_response = MagicMock() mock_response.choices = [MagicMock()] mock_response.choices[0].message.content = "测试响应" # 模拟不同类型的消息 mock_prompt_template = MagicMock() mock_system_msg = MagicMock() mock_system_msg.type = "system" mock_system_msg.content = "系统提示" mock_user_msg = MagicMock() mock_user_msg.type = "user" mock_user_msg.content = "用户输入" mock_unknown_msg = MagicMock() mock_unknown_msg.type = "unknown" mock_unknown_msg.content = "未知类型" mock_prompt_template.format_messages.return_value = [ mock_system_msg, mock_user_msg, mock_unknown_msg ] with patch.object(reviewer.client.chat.completions, 'create', new_callable=AsyncMock) as mock_create, \ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt: mock_create.return_value = mock_response mock_get_prompt.return_value = mock_prompt_template result = await reviewer.check_semantic_logic( trace_id="test_trace_006", review_content=sample_review_content ) # 验证消息格式转换 call_kwargs = mock_create.call_args.kwargs messages = call_kwargs["messages"] assert len(messages) == 3 assert messages[0]["role"] == "system" assert messages[0]["content"] == "系统提示" assert messages[1]["role"] == "user" assert messages[1]["content"] == "用户输入" assert messages[2]["role"] == "user" # unknown类型应转为user assert messages[2]["content"] == "未知类型" @pytest.mark.asyncio async def test_execution_time_tracking(self, reviewer, sample_review_content): """测试执行时间跟踪""" mock_response = MagicMock() mock_response.choices = [MagicMock()] mock_response.choices[0].message.content = "测试" mock_prompt_template = MagicMock() mock_message = MagicMock() mock_message.type = "user" mock_message.content = "测试" mock_prompt_template.format_messages.return_value = [mock_message] async def slow_api_call(*args, **kwargs): """模拟慢速API调用""" await asyncio.sleep(0.1) # 模拟100ms延迟 return mock_response with patch.object(reviewer.client.chat.completions, 'create', new_callable=AsyncMock) as mock_create, \ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt: mock_create.side_effect = slow_api_call mock_get_prompt.return_value = mock_prompt_template result = await reviewer.check_semantic_logic( trace_id="test_trace_007", review_content=sample_review_content ) # 验证执行时间大于100ms assert result.execution_time >= 0.1 assert result.execution_time < 1.0 # 应该不会太长 class TestIntegration: """集成测试类""" @pytest.mark.asyncio @pytest.mark.integration async def test_full_workflow(self): """测试完整工作流程（需要实际API可用）""" # 注意：此测试需要实际的API服务可用 # 在CI/CD环境中可能需要跳过 pytest.skip("需要实际API服务，跳过集成测试") reviewer = SemanticLogicReviewer() result = await reviewer.check_semantic_logic( trace_id="integration_test_001", review_content="测试施工方案内容", review_references="测试参考标准" ) assert isinstance(result, ReviewResult) assert result.execution_time is not None class TestEdgeCases: """边界情况测试类""" @pytest.mark.asyncio async def test_very_long_content(self, reviewer): """测试超长内容""" long_content = "测试内容 " * 10000 # 非常长的内容 mock_response = MagicMock() mock_response.choices = [MagicMock()] mock_response.choices[0].message.content = "内容过长" mock_prompt_template = MagicMock() mock_message = MagicMock() mock_message.type = "user" mock_message.content = long_content mock_prompt_template.format_messages.return_value = [mock_message] with patch.object(reviewer.client.chat.completions, 'create', new_callable=AsyncMock) as mock_create, \ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt: mock_create.return_value = mock_response mock_get_prompt.return_value = mock_prompt_template result = await reviewer.check_semantic_logic( trace_id="test_edge_001", review_content=long_content ) assert result.success is True @pytest.mark.asyncio async def test_special_characters(self, reviewer): """测试特殊字符""" special_content = "测试内容包含特殊字符：@#$%^&*(){}[]|\\:;\"'<>,.?/~`" mock_response = MagicMock() mock_response.choices = [MagicMock()] mock_response.choices[0].message.content = "特殊字符处理正常" mock_prompt_template = MagicMock() mock_message = MagicMock() mock_message.type = "user" mock_message.content = special_content mock_prompt_template.format_messages.return_value = [mock_message] with patch.object(reviewer.client.chat.completions, 'create', new_callable=AsyncMock) as mock_create, \ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt: mock_create.return_value = mock_response mock_get_prompt.return_value = mock_prompt_template result = await reviewer.check_semantic_logic( trace_id="test_edge_002", review_content=special_content ) assert result.success is True @pytest.mark.asyncio async def test_unicode_content(self, reviewer): """测试Unicode字符""" unicode_content = "测试内容包含各种语言：English, 中文, 日本語, 한국어, Русский, العربية, 🚀🎉" mock_response = MagicMock() mock_response.choices = [MagicMock()] mock_response.choices[0].message.content = "Unicode处理正常" mock_prompt_template = MagicMock() mock_message = MagicMock() mock_message.type = "user" mock_message.content = unicode_content mock_prompt_template.format_messages.return_value = [mock_message] with patch.object(reviewer.client.chat.completions, 'create', new_callable=AsyncMock) as mock_create, \ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt: mock_create.return_value = mock_response mock_get_prompt.return_value = mock_prompt_template result = await reviewer.check_semantic_logic( trace_id="test_edge_003", review_content=unicode_content ) assert result.success is True if __name__ == "__main__": # 运行测试 pytest.main([__file__, "-v", "-s", "--tb=short"])