|
|
@@ -0,0 +1,474 @@
|
|
|
+"""
|
|
|
+语义逻辑审查模块单元测试
|
|
|
+测试 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_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_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_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"])
|
|
|
+
|
