LQAgentPlatform/utils_test/Semantic_Logic_Test/test_semantic_logic.py

"""
语义逻辑审查模块单元测试
测试 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"])