LQAgentPlatform/utils_test/Semantic_Logic_Test/test_semantic_logic.py

"""
语义逻辑审查模块 — 链路测试
测试 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, PropertyMock

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,
)
from core.construction_review.component.reviewers.base_reviewer import ReviewResult


# ─── 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)


# ─── 单元测试：成功路径 ────────────────────────────────────────────────────────

class TestCheckSemanticLogicSuccess:
    """检查语义逻辑 — 成功路径"""

    @pytest.mark.asyncio
    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_invoke.return_value = fake_response
            mock_get_prompt.return_value = mock_tmpl

            result = await reviewer.check_semantic_logic(
                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 result.details["response"] == fake_response
            assert result.error_message is None
            assert result.execution_time > 0

    @pytest.mark.asyncio
    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_invoke.return_value = fake_response
            mock_get_prompt.return_value = mock_tmpl

            result = await reviewer.check_semantic_logic(
                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,
            )

            assert result.success is True
            # 不应抛异常


# ─── 单元测试：错误路径 ────────────────────────────────────────────────────────

class TestCheckSemanticLogicError:
    """检查语义逻辑 — 错误路径"""

    @pytest.mark.asyncio
    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_invoke.side_effect = Exception("模型服务连接超时")
            mock_get_prompt.return_value = mock_tmpl

            result = await reviewer.check_semantic_logic(
                trace_id="trace_err_001",
                review_content="内容",
                state=state,
                stage_name="basic_check",
            )

            assert isinstance(result, ReviewResult)
            assert result.success is False
            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_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_invoke.side_effect = Exception("boom")
            mock_get_prompt.return_value = _make_mock_prompt_template()

            result = await reviewer.check_semantic_logic(
                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_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="chain_002",
                review_content="前文采用A方法。后文说不能采用A方法。",
                state=state,
                stage_name="semantic_stage",
            )

            assert result.success is True
            await asyncio.sleep(0.1)
            assert state["progress_manager"].update_stage_progress.called


# ─── 边界情况测试 ──────────────────────────────────────────────────────────────

class TestEdgeCases:
    """边界情况测试"""

    @pytest.mark.asyncio
    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_invoke.return_value = "内容为空，无法审查"
            mock_get_prompt.return_value = _make_mock_prompt_template()

            result = await reviewer.check_semantic_logic(
                trace_id="edge_001",
                review_content="",
            )

            assert result.success is True
            assert result.details["name"] == "semantic_logic_check"

    @pytest.mark.asyncio
    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_invoke.return_value = "无明显问题"
            mock_get_prompt.return_value = _make_mock_prompt_template()

            result = await reviewer.check_semantic_logic(
                trace_id="edge_002",
                review_content=content,
            )

            assert result.success is True

    @pytest.mark.asyncio
    async def test_special_characters(self):
        """特殊字符内容"""
        reviewer = SemanticLogicReviewer()
        content = "特殊字符：@#$%^&*(){}[]|\\:;\"'<>,.?/~` ±×÷≈≠≤≥∞"

        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 = _make_mock_prompt_template()

            result = await reviewer.check_semantic_logic(
                trace_id="edge_003",
                review_content=content,
            )

            assert result.success is True

    @pytest.mark.asyncio
    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_invoke.return_value = "无明显问题"
            mock_get_prompt.return_value = _make_mock_prompt_template()

            result = await reviewer.check_semantic_logic(
                trace_id="edge_004",
                review_content=content,
            )

            assert result.success is True

    @pytest.mark.asyncio
    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_invoke.side_effect = slow_response
            mock_get_prompt.return_value = _make_mock_prompt_template()

            result = await reviewer.check_semantic_logic(
                trace_id="edge_005",
                review_content="测试",
            )

            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"])