|
|
@@ -1,474 +1,426 @@
|
|
|
"""
|
|
|
-语义逻辑审查模块单元测试
|
|
|
-测试 semantic_logic.py 中的 SemanticLogicReviewer 类
|
|
|
+语义逻辑审查模块 — 链路测试
|
|
|
+测试 semantic_logic.py 中的 SemanticLogicReviewer 完整调用链路
|
|
|
+
|
|
|
+链路: prompt_loader → ChatPromptTemplate.format_messages() → generate_model_client.get_model_generate_invoke() → ReviewResult
|
|
|
"""
|
|
|
|
|
|
import pytest
|
|
|
import asyncio
|
|
|
import sys
|
|
|
import os
|
|
|
-from unittest.mock import Mock, patch, AsyncMock, MagicMock
|
|
|
-from typing import Dict, Any
|
|
|
+from unittest.mock import Mock, patch, AsyncMock, MagicMock, PropertyMock
|
|
|
|
|
|
-# 添加项目根目录到路径
|
|
|
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
|
|
+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
|
|
|
+# ─── helpers ───────────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+def _make_mock_prompt_template(messages=None):
|
|
|
+ """构造一个模拟的 ChatPromptTemplate，format_messages() 返回指定消息列表"""
|
|
|
+ from langchain_core.messages import SystemMessage, HumanMessage
|
|
|
+
|
|
|
+ if messages is None:
|
|
|
+ messages = [
|
|
|
+ SystemMessage(content="你是语义逻辑审查专家"),
|
|
|
+ HumanMessage(content="请审查：测试施工方案内容"),
|
|
|
+ ]
|
|
|
+ tmpl = MagicMock()
|
|
|
+ tmpl.format_messages.return_value = messages
|
|
|
+ return tmpl
|
|
|
+
|
|
|
+
|
|
|
+def _make_mock_state():
|
|
|
+ """构造模拟的 state 字典（含 progress_manager）"""
|
|
|
+ pm = AsyncMock()
|
|
|
+ pm.update_stage_progress = AsyncMock()
|
|
|
+ return {
|
|
|
+ "progress_manager": pm,
|
|
|
+ "callback_task_id": "test_callback_001",
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+# ─── 单元测试：初始化 ──────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+class TestInit:
|
|
|
+ """审查器初始化测试"""
|
|
|
+
|
|
|
+ def test_init_model_client_set(self):
|
|
|
+ """初始化后 model_client 应为 GenerateModelClient 实例"""
|
|
|
+ r = SemanticLogicReviewer()
|
|
|
+ from foundation.ai.agent.generate.model_generate import GenerateModelClient
|
|
|
+ assert isinstance(r.model_client, GenerateModelClient)
|
|
|
+
|
|
|
+ def test_global_singleton(self):
|
|
|
+ """全局单例实例"""
|
|
|
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
|
|
|
-
|
|
|
+
|
|
|
+
|
|
|
+# ─── 单元测试：成功路径 ────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+class TestCheckSemanticLogicSuccess:
|
|
|
+ """检查语义逻辑 — 成功路径"""
|
|
|
+
|
|
|
@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, \
|
|
|
+ async def test_success_path(self):
|
|
|
+ """
|
|
|
+ 验证完整成功链路:
|
|
|
+ prompt_loader.get_prompt_template → format_messages → model_client.get_model_generate_invoke → ReviewResult
|
|
|
+ """
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+ mock_tmpl = _make_mock_prompt_template()
|
|
|
+ fake_response = '{"issue_point": "无", "location": "", "suggestion": "", "reason": "", "risk_level": ""}'
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke, \
|
|
|
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
|
|
|
-
|
|
|
- # 执行测试
|
|
|
+ mock_invoke.return_value = fake_response
|
|
|
+ mock_get_prompt.return_value = mock_tmpl
|
|
|
+
|
|
|
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"
|
|
|
+ trace_id="trace_001",
|
|
|
+ review_content="测试施工方案内容",
|
|
|
)
|
|
|
-
|
|
|
- # 验证结果
|
|
|
+
|
|
|
+ # 链路1: prompt_loader 被正确调用
|
|
|
+ mock_get_prompt.assert_called_once_with(
|
|
|
+ "basic", "semantic_logic_check",
|
|
|
+ review_content="测试施工方案内容",
|
|
|
+ review_references="",
|
|
|
+ )
|
|
|
+
|
|
|
+ # 链路2: format_messages 被调用
|
|
|
+ mock_tmpl.format_messages.assert_called_once()
|
|
|
+
|
|
|
+ # 链路3: model_client 使用 function_name="grammar_check"
|
|
|
+ mock_invoke.assert_awaited_once()
|
|
|
+ call_kwargs = mock_invoke.call_args.kwargs
|
|
|
+ assert call_kwargs["trace_id"] == "trace_001"
|
|
|
+ assert call_kwargs["function_name"] == "grammar_check"
|
|
|
+ assert call_kwargs["messages"] == mock_tmpl.format_messages.return_value
|
|
|
+
|
|
|
+ # 链路4: 返回正确的 ReviewResult
|
|
|
assert isinstance(result, ReviewResult)
|
|
|
assert result.success is True
|
|
|
assert result.details["name"] == "semantic_logic_check"
|
|
|
- assert "审查结果" in result.details["response"]
|
|
|
+ assert result.details["response"] == fake_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, \
|
|
|
+ async def test_with_state_triggers_progress(self):
|
|
|
+ """有 state 时触发进度推送"""
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+ mock_tmpl = _make_mock_prompt_template()
|
|
|
+ fake_response = "无明显问题"
|
|
|
+ state = _make_mock_state()
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke, \
|
|
|
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
|
|
|
-
|
|
|
+ mock_invoke.return_value = fake_response
|
|
|
+ mock_get_prompt.return_value = mock_tmpl
|
|
|
+
|
|
|
result = await reviewer.check_semantic_logic(
|
|
|
- trace_id="test_trace_002",
|
|
|
- review_content=sample_review_content,
|
|
|
+ trace_id="trace_002",
|
|
|
+ review_content="施工方案内容...",
|
|
|
+ state=state,
|
|
|
+ stage_name="basic_check",
|
|
|
+ )
|
|
|
+
|
|
|
+ assert result.success is True
|
|
|
+
|
|
|
+ # 进度推送是 fire-and-forget (asyncio.create_task)，等待一下
|
|
|
+ await asyncio.sleep(0.1)
|
|
|
+
|
|
|
+ pm = state["progress_manager"]
|
|
|
+ assert pm.update_stage_progress.called
|
|
|
+ call_kwargs = pm.update_stage_progress.call_args.kwargs
|
|
|
+ assert call_kwargs["callback_task_id"] == "test_callback_001"
|
|
|
+ assert call_kwargs["stage_name"] == "basic_check"
|
|
|
+ # issues 中包含审查结果数据
|
|
|
+ assert len(call_kwargs["issues"]) == 1
|
|
|
+ assert call_kwargs["issues"][0]["name"] == "semantic_logic_check"
|
|
|
+ assert call_kwargs["issues"][0]["success"] is True
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_without_state_no_progress_call(self):
|
|
|
+ """无 state 时不触发进度推送，不抛异常"""
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+ mock_tmpl = _make_mock_prompt_template()
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke, \
|
|
|
+ patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt:
|
|
|
+ mock_invoke.return_value = "无明显问题"
|
|
|
+ mock_get_prompt.return_value = mock_tmpl
|
|
|
+
|
|
|
+ result = await reviewer.check_semantic_logic(
|
|
|
+ trace_id="trace_003",
|
|
|
+ review_content="内容",
|
|
|
state=None,
|
|
|
- stage_name=None
|
|
|
+ stage_name=None,
|
|
|
)
|
|
|
-
|
|
|
+
|
|
|
assert result.success is True
|
|
|
- assert result.details["name"] == "semantic_logic_check"
|
|
|
-
|
|
|
+ # 不应抛异常
|
|
|
+
|
|
|
+
|
|
|
+# ─── 单元测试：错误路径 ────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+class TestCheckSemanticLogicError:
|
|
|
+ """检查语义逻辑 — 错误路径"""
|
|
|
+
|
|
|
@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, \
|
|
|
+ async def test_model_call_failure(self):
|
|
|
+ """模型调用抛异常 → 返回 success=False 的 ReviewResult"""
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+ mock_tmpl = _make_mock_prompt_template()
|
|
|
+ state = _make_mock_state()
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke, \
|
|
|
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
|
|
|
-
|
|
|
+ mock_invoke.side_effect = Exception("模型服务连接超时")
|
|
|
+ mock_get_prompt.return_value = mock_tmpl
|
|
|
+
|
|
|
result = await reviewer.check_semantic_logic(
|
|
|
- trace_id="test_trace_003",
|
|
|
- review_content=sample_review_content,
|
|
|
- state=mock_state,
|
|
|
- stage_name="basic_check"
|
|
|
+ trace_id="trace_err_001",
|
|
|
+ review_content="内容",
|
|
|
+ state=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()
|
|
|
-
|
|
|
+ assert result.details["name"] == "semantic_logic_check"
|
|
|
+ assert "模型服务连接超时" in result.error_message
|
|
|
+ assert result.execution_time > 0
|
|
|
+
|
|
|
+ # 失败也应推送进度
|
|
|
+ await asyncio.sleep(0.1)
|
|
|
+ assert state["progress_manager"].update_stage_progress.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, \
|
|
|
+ async def test_prompt_loader_failure(self):
|
|
|
+ """prompt_loader 抛异常 → 应被捕获"""
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+
|
|
|
+ with patch('core.construction_review.component.reviewers.semantic_logic.prompt_loader.get_prompt_template') as mock_get_prompt:
|
|
|
+ mock_get_prompt.side_effect = RuntimeError("YAML 解析失败")
|
|
|
+
|
|
|
+ result = await reviewer.check_semantic_logic(
|
|
|
+ trace_id="trace_err_002",
|
|
|
+ review_content="内容",
|
|
|
+ )
|
|
|
+
|
|
|
+ assert result.success is False
|
|
|
+ assert "YAML 解析失败" in result.error_message
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_error_without_state(self):
|
|
|
+ """错误时无 state 也不应抛异常"""
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke, \
|
|
|
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
|
|
|
-
|
|
|
+ mock_invoke.side_effect = Exception("boom")
|
|
|
+ mock_get_prompt.return_value = _make_mock_prompt_template()
|
|
|
+
|
|
|
result = await reviewer.check_semantic_logic(
|
|
|
- trace_id="test_trace_004",
|
|
|
- review_content="",
|
|
|
- state=mock_state,
|
|
|
- stage_name="basic_check"
|
|
|
+ trace_id="trace_err_003",
|
|
|
+ review_content="内容",
|
|
|
+ state=None,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert result.success is False
|
|
|
+ assert "boom" in result.error_message
|
|
|
+
|
|
|
+
|
|
|
+# ─── 链路集成测试：真实 prompt_loader + mock AI 调用 ────────────────────────────
|
|
|
+
|
|
|
+class TestChainIntegration:
|
|
|
+ """集成链路测试：使用真实 prompt_loader 验证完整链路（仅 mock AI 调用）"""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_full_chain_with_real_prompt_loader(self):
|
|
|
+ """
|
|
|
+ 使用真实 prompt_loader 加载 basic/semantic_logic_check 模板，
|
|
|
+ 仅 mock 底层 AI 调用，验证完整链路畅通。
|
|
|
+ """
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+ fake_ai_response = "无明显问题"
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke:
|
|
|
+ mock_invoke.return_value = fake_ai_response
|
|
|
+
|
|
|
+ result = await reviewer.check_semantic_logic(
|
|
|
+ trace_id="chain_001",
|
|
|
+ review_content="1. 工程概况\n本工程位于四川省，全长120公里。\n2. 施工安排\n先进行基础施工，再进行上部结构施工。",
|
|
|
)
|
|
|
-
|
|
|
+
|
|
|
assert result.success is True
|
|
|
assert result.details["name"] == "semantic_logic_check"
|
|
|
-
|
|
|
+ assert result.details["response"] == fake_ai_response
|
|
|
+
|
|
|
+ # 验证 AI 调用确实收到了正确格式的消息
|
|
|
+ call_kwargs = mock_invoke.call_args.kwargs
|
|
|
+ assert call_kwargs["function_name"] == "grammar_check"
|
|
|
+ messages = call_kwargs["messages"]
|
|
|
+ assert len(messages) >= 2 # system + user
|
|
|
+
|
|
|
+ # system message 应包含角色定义
|
|
|
+ system_msg = messages[0]
|
|
|
+ assert "语义逻辑审查" in system_msg.content or "role" in system_msg.content.lower()
|
|
|
+
|
|
|
+ # user message 应包含待审查内容
|
|
|
+ user_msg = messages[-1]
|
|
|
+ assert "工程概况" in user_msg.content or "120公里" in user_msg.content
|
|
|
+
|
|
|
@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
|
|
|
-
|
|
|
+ async def test_chain_with_state_progress(self):
|
|
|
+ """集成链路 + 进度推送"""
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+ state = _make_mock_state()
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke:
|
|
|
+ mock_invoke.return_value = '{"issue_point": "逻辑矛盾", "location": "第2条", "suggestion": "调整", "reason": "前后矛盾", "risk_level": "中风险"}'
|
|
|
+
|
|
|
result = await reviewer.check_semantic_logic(
|
|
|
- trace_id="test_trace_005",
|
|
|
- review_content=sample_review_content,
|
|
|
- review_references=sample_review_references
|
|
|
+ trace_id="chain_002",
|
|
|
+ review_content="前文采用A方法。后文说不能采用A方法。",
|
|
|
+ state=state,
|
|
|
+ stage_name="semantic_stage",
|
|
|
)
|
|
|
-
|
|
|
+
|
|
|
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
|
|
|
-
|
|
|
+ await asyncio.sleep(0.1)
|
|
|
+ assert state["progress_manager"].update_stage_progress.called
|
|
|
+
|
|
|
+
|
|
|
+# ─── 边界情况测试 ──────────────────────────────────────────────────────────────
|
|
|
+
|
|
|
+class TestEdgeCases:
|
|
|
+ """边界情况测试"""
|
|
|
+
|
|
|
@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, \
|
|
|
+ async def test_empty_content(self):
|
|
|
+ """空内容 — 链路正常完成"""
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke, \
|
|
|
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
|
|
|
-
|
|
|
+ mock_invoke.return_value = "内容为空，无法审查"
|
|
|
+ mock_get_prompt.return_value = _make_mock_prompt_template()
|
|
|
+
|
|
|
result = await reviewer.check_semantic_logic(
|
|
|
- trace_id="test_trace_006",
|
|
|
- review_content=sample_review_content
|
|
|
+ trace_id="edge_001",
|
|
|
+ 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"] == "未知类型"
|
|
|
-
|
|
|
+
|
|
|
+ assert result.success is True
|
|
|
+ assert result.details["name"] == "semantic_logic_check"
|
|
|
+
|
|
|
@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, \
|
|
|
+ async def test_long_content(self):
|
|
|
+ """长内容 — 链路正常完成"""
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+ long_text = "第{}条 施工技术要求详细说明...\n".format
|
|
|
+ content = "\n".join(long_text(i) for i in range(500))
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke, \
|
|
|
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
|
|
|
-
|
|
|
+ mock_invoke.return_value = "无明显问题"
|
|
|
+ mock_get_prompt.return_value = _make_mock_prompt_template()
|
|
|
+
|
|
|
result = await reviewer.check_semantic_logic(
|
|
|
- trace_id="test_trace_007",
|
|
|
- review_content=sample_review_content
|
|
|
+ trace_id="edge_002",
|
|
|
+ review_content=content,
|
|
|
)
|
|
|
-
|
|
|
- # 验证执行时间大于100ms
|
|
|
- assert result.execution_time >= 0.1
|
|
|
- assert result.execution_time < 1.0 # 应该不会太长
|
|
|
|
|
|
+ assert result.success is True
|
|
|
|
|
|
-class TestIntegration:
|
|
|
- """集成测试类"""
|
|
|
-
|
|
|
@pytest.mark.asyncio
|
|
|
- @pytest.mark.integration
|
|
|
- async def test_full_workflow(self):
|
|
|
- """测试完整工作流程（需要实际API可用）"""
|
|
|
- # 注意：此测试需要实际的API服务可用
|
|
|
- # 在CI/CD环境中可能需要跳过
|
|
|
- pytest.skip("需要实际API服务，跳过集成测试")
|
|
|
-
|
|
|
+ async def test_special_characters(self):
|
|
|
+ """特殊字符内容"""
|
|
|
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
|
|
|
-
|
|
|
+ content = "特殊字符：@#$%^&*(){}[]|\\:;\"'<>,.?/~` ±×÷≈≠≤≥∞"
|
|
|
|
|
|
-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, \
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke, \
|
|
|
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
|
|
|
-
|
|
|
+ mock_invoke.return_value = "无明显问题"
|
|
|
+ mock_get_prompt.return_value = _make_mock_prompt_template()
|
|
|
+
|
|
|
result = await reviewer.check_semantic_logic(
|
|
|
- trace_id="test_edge_001",
|
|
|
- review_content=long_content
|
|
|
+ trace_id="edge_003",
|
|
|
+ review_content=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, \
|
|
|
+ async def test_unicode_multilang(self):
|
|
|
+ """多语言 Unicode 内容"""
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+ content = "中文 / English / 日本語 / 한국어 / Русский / العربية / 🚧🏗️"
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke, \
|
|
|
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
|
|
|
-
|
|
|
+ mock_invoke.return_value = "无明显问题"
|
|
|
+ mock_get_prompt.return_value = _make_mock_prompt_template()
|
|
|
+
|
|
|
result = await reviewer.check_semantic_logic(
|
|
|
- trace_id="test_edge_002",
|
|
|
- review_content=special_content
|
|
|
+ trace_id="edge_004",
|
|
|
+ review_content=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, \
|
|
|
+ async def test_execution_time_tracking(self):
|
|
|
+ """验证执行时间被正确记录"""
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+
|
|
|
+ async def slow_response(*args, **kwargs):
|
|
|
+ await asyncio.sleep(0.15)
|
|
|
+ return "响应"
|
|
|
+
|
|
|
+ with patch.object(reviewer.model_client, 'get_model_generate_invoke', new_callable=AsyncMock) as mock_invoke, \
|
|
|
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
|
|
|
-
|
|
|
+ mock_invoke.side_effect = slow_response
|
|
|
+ mock_get_prompt.return_value = _make_mock_prompt_template()
|
|
|
+
|
|
|
result = await reviewer.check_semantic_logic(
|
|
|
- trace_id="test_edge_003",
|
|
|
- review_content=unicode_content
|
|
|
+ trace_id="edge_005",
|
|
|
+ review_content="测试",
|
|
|
)
|
|
|
-
|
|
|
- assert result.success is True
|
|
|
+
|
|
|
+ assert result.execution_time >= 0.15
|
|
|
+
|
|
|
+
|
|
|
+# ─── 实际 API 集成测试（需手动开启）─────────────────────────────────────────────
|
|
|
+
|
|
|
+class TestLiveAPI:
|
|
|
+ """实际 API 调用测试（标记为 integration，默认跳过）"""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ @pytest.mark.integration
|
|
|
+ async def test_live_api_call(self):
|
|
|
+ """真实调用 AI 接口（需服务可用）"""
|
|
|
+ pytest.skip("需要实际 API 服务，手动运行")
|
|
|
+
|
|
|
+ reviewer = SemanticLogicReviewer()
|
|
|
+ result = await reviewer.check_semantic_logic(
|
|
|
+ trace_id="live_001",
|
|
|
+ review_content="1. 工程概况\n本工程为高速公路桥梁项目。\n2. 施工顺序\n先施工上部结构，再进行基础施工。",
|
|
|
+ )
|
|
|
+
|
|
|
+ assert isinstance(result, ReviewResult)
|
|
|
+ # 不强制断言 success，依赖实际服务状态
|
|
|
+ assert result.execution_time is not None
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
- # 运行测试
|
|
|
pytest.main([__file__, "-v", "-s", "--tb=short"])
|
|
|
-
|
