LQAgentPlatform/utils_test/Sensitive_Test/test_grammar_check_chain.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
语法审查 (GrammarCheckReviewer.check_grammar) 全链路测试

测试范围（仅限 sensitive_word_check.py 自身逻辑）：
  1. Prompt 模板：YAML 配置结构 → prompt_loader 加载 → 变量填充
  2. 模型调用：check_grammar → model_client.get_model_generate_invoke(function_name="sensitive_check")
  3. 结果封装：模型响应 → ReviewResult(details={name, response}, execution_time)
  4. 异常处理：超时/错误 → ReviewResult(success=False)
  5. 推送集成：state.progress_manager 异步推送

用法：
  cd <project_root>
  set PYTHONPATH=<project_root>
  pytest utils_test/Sensitive_Test/test_grammar_check_chain.py -v
"""

import asyncio
import sys, os, json, types
from pathlib import Path
from typing import Dict, Any, Optional
from dataclasses import dataclass
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

# -----------------------------------------------------------
# 项目根目录
# -----------------------------------------------------------
current_dir = Path(__file__).parent.absolute()
project_root = current_dir.parent.parent
sys.path.insert(0, str(project_root))

# -----------------------------------------------------------
# 只 Mock 导致导入链断裂的 modules，不引入多余依赖
# -----------------------------------------------------------
# directory_extraction.py 引用了不存在的 PydanticOutputParser
_mock_dir_ext = types.ModuleType(
    "core.construction_review.component.reviewers.utils.directory_extraction"
)
_mock_dir_ext.extract_basis_with_langchain_qwen = MagicMock()
_mock_dir_ext.BasisItems = MagicMock()
_mock_dir_ext.BasisItem = MagicMock()
sys.modules["core.construction_review.component.reviewers.utils.directory_extraction"] = (
    _mock_dir_ext
)

# langfuse 未安装
sys.modules.setdefault("langfuse", MagicMock())

# langchain_openai 未安装
_mock_lc_openai = types.ModuleType("langchain_openai")
_mock_lc_openai.ChatOpenAI = MagicMock()
_mock_lc_openai.OpenAIEmbeddings = MagicMock()
sys.modules.setdefault("langchain_openai", _mock_lc_openai)
sys.modules.setdefault("langchain_openai.chat_models", MagicMock())
sys.modules.setdefault("langchain_openai.embeddings", MagicMock())


# ============================================================
# 核心测试：GrammarCheckReviewer 链路
# ============================================================

class TestGrammarCheckChain:
    """sensitive_word_check.py 审查链路测试"""

    # --------------------------------------------------------
    # 1. Prompt 模板配置
    # --------------------------------------------------------
    def test_prompt_template_structure(self):
        """验证 YAML 模板包含必要字段和变量"""
        import yaml
        prompt_dir = (
            project_root
            / "core"
            / "construction_review"
            / "component"
            / "reviewers"
            / "prompt"
        )
        with open(prompt_dir / "basic_reviewers.yaml", "r", encoding="utf-8") as f:
            config = yaml.safe_load(f)

        cfg = config.get("sensitive_word_check", {})
        assert cfg, "sensitive_word_check 键不存在"
        assert "system_prompt" in cfg
        assert "user_prompt_template" in cfg
        tmpl = cfg["user_prompt_template"]
        assert "{review_content}" in tmpl
        assert "{review_references}" in tmpl

    def test_prompt_loader_loads_template(self):
        """prompt_loader 能正常加载 sensitive_word_check 模板"""
        from core.construction_review.component.reviewers.utils.prompt_loader import (
            prompt_loader,
        )
        prompts = prompt_loader.list_available_prompts("basic")
        assert "sensitive_word_check" in prompts

        template = prompt_loader.get_prompt_template(
            "basic",
            "sensitive_word_check",
            review_content="测试内容abc123",
            review_references="测试参考",
        )
        messages = template.format_messages()
        assert len(messages) == 2
        assert messages[0].type == "system"
        assert messages[1].type == "human"
        assert "测试内容abc123" in messages[1].content
        assert "测试参考" in messages[1].content

    # --------------------------------------------------------
    # 2. 模型调用参数构建
    # --------------------------------------------------------
    def test_check_grammar_calls_model_with_function_name(self):
        """验证 check_grammar 以 function_name=sensitive_check 调用模型"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        reviewer.model_client.get_model_generate_invoke = AsyncMock(return_value="无明显问题")

        async def run():
            await reviewer.check_grammar(
                trace_id="trace_001", review_content="测试内容。"
            )
            kwargs = reviewer.model_client.get_model_generate_invoke.call_args[1]
            assert kwargs.get("function_name") == "sensitive_check", (
                f"期望 sensitive_check，实际: {kwargs.get('function_name')}"
            )
            assert kwargs.get("trace_id") == "trace_001"
            messages = kwargs.get("messages", [])
            assert len(messages) == 2
            assert "测试内容。" in messages[1].content

        asyncio.run(run())

    # --------------------------------------------------------
    # 3. 正常审查结果封装
    # --------------------------------------------------------
    def test_success_result_structure(self):
        """验证成功时返回正确的 ReviewResult"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )
        from core.construction_review.component.reviewers.base_reviewer import ReviewResult

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        mock_resp = json.dumps({"issue_point": "测试问题"}, ensure_ascii=False)
        reviewer.model_client.get_model_generate_invoke = AsyncMock(return_value=mock_resp)

        async def run():
            result = await reviewer.check_grammar(
                trace_id="t_succ", review_content="测试。"
            )
            assert isinstance(result, ReviewResult)
            assert result.success is True
            assert result.error_message is None
            assert result.details.get("name") == "sensitive_word_check"
            assert result.details.get("response") == mock_resp
            assert isinstance(result.execution_time, (int, float))
            assert result.execution_time >= 0

        asyncio.run(run())

    def test_model_json_response_preserved(self):
        """模型 JSON 响应完整保留"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        resp = json.dumps({
            "issue_point": "绝对化用语",
            "location": "第一章",
            "suggestion": "修改建议",
            "risk_level": "中风险",
        }, ensure_ascii=False)
        reviewer.model_client.get_model_generate_invoke = AsyncMock(return_value=resp)

        async def run():
            result = await reviewer.check_grammar(
                trace_id="t_json", review_content="绝对不会出现问题。"
            )
            assert result.success is True
            assert "绝对化用语" in result.details["response"]

        asyncio.run(run())

    def test_execution_time_recorded(self):
        """耗时被正确记录"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()

        async def slow(*a, **kw):
            await asyncio.sleep(0.05)
            return "无明显问题"

        reviewer.model_client.get_model_generate_invoke = AsyncMock(side_effect=slow)

        async def run():
            result = await reviewer.check_grammar(
                trace_id="t_time", review_content="测试。"
            )
            assert result.execution_time >= 0.04

        asyncio.run(run())

    # --------------------------------------------------------
    # 4. 异常处理
    # --------------------------------------------------------
    def test_timeout_returns_error(self):
        """超时 → success=False"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        reviewer.model_client.get_model_generate_invoke = AsyncMock(
            side_effect=TimeoutError("模型调用超时")
        )

        async def run():
            result = await reviewer.check_grammar(
                trace_id="t_to", review_content="测试。"
            )
            assert result.success is False
            assert result.error_message is not None
            assert "超时" in result.error_message
            assert result.details.get("name") == "sensitive_word_check"
            assert result.execution_time is not None

        asyncio.run(run())

    def test_api_error_returns_error(self):
        """API 异常 → success=False"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        reviewer.model_client.get_model_generate_invoke = AsyncMock(
            side_effect=RuntimeError("API 500")
        )

        async def run():
            result = await reviewer.check_grammar(
                trace_id="t_api", review_content="测试。"
            )
            assert result.success is False
            assert "语法检查失败" in result.error_message

        asyncio.run(run())

    def test_empty_content_handled(self):
        """空内容不崩溃"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        reviewer.model_client.get_model_generate_invoke = AsyncMock(return_value="无明显问题")

        async def run():
            result = await reviewer.check_grammar(
                trace_id="t_empty", review_content=""
            )
            assert result.success is True

        asyncio.run(run())

    # --------------------------------------------------------
    # 5. Progress Manager 推送
    # --------------------------------------------------------
    def test_progress_push_on_success(self):
        """成功时推送进度"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        reviewer.model_client.get_model_generate_invoke = AsyncMock(return_value="无明显问题")

        pm = AsyncMock()
        pm.update_stage_progress = AsyncMock()
        state = {"progress_manager": pm, "callback_task_id": "cb_001"}

        async def run():
            result = await reviewer.check_grammar(
                trace_id="t_ps",
                review_content="测试。",
                state=state,
                stage_name="test_stage",
            )
            assert result.success is True
            # asyncio.create_task 是 fire-and-forget，需要 yield 让 task 执行
            await asyncio.sleep(0)
            pm.update_stage_progress.assert_awaited_once()
            kw = pm.update_stage_progress.call_args[1]
            assert kw["callback_task_id"] == "cb_001"
            assert kw["stage_name"] == "test_stage"
            assert kw["status"] == "processing"
            assert "sensitive_word_check" in kw["message"]
            assert len(kw["issues"]) == 1
            assert kw["issues"][0]["name"] == "sensitive_word_check"
            assert kw["issues"][0]["success"] is True

        asyncio.run(run())

    def test_progress_push_on_failure(self):
        """失败时也推送"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        reviewer.model_client.get_model_generate_invoke = AsyncMock(
            side_effect=RuntimeError("err")
        )

        pm = AsyncMock()
        pm.update_stage_progress = AsyncMock()
        state = {"progress_manager": pm, "callback_task_id": "cb_002"}

        async def run():
            result = await reviewer.check_grammar(
                trace_id="t_pf",
                review_content="测试。",
                state=state,
                stage_name="test",
            )
            assert result.success is False
            await asyncio.sleep(0)  # yield 让 create_task 执行
            pm.update_stage_progress.assert_awaited_once()
            issues = pm.update_stage_progress.call_args[1]["issues"]
            assert issues[0]["success"] is False

        asyncio.run(run())

    def test_no_state_skips_push(self):
        """不传 state 不推送"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        reviewer.model_client.get_model_generate_invoke = AsyncMock(return_value="无明显问题")

        async def run():
            result = await reviewer.check_grammar(
                trace_id="t_ns", review_content="测试。"
            )
            assert result.success is True

        asyncio.run(run())

    # --------------------------------------------------------
    # 6. 模块导出
    # --------------------------------------------------------
    def test_module_exports(self):
        """验证模块导出全局实例"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            sensitive_word_check_reviewer,
            GrammarCheckReviewer,
        )
        assert isinstance(sensitive_word_check_reviewer, GrammarCheckReviewer)

    def test_default_model_client(self):
        """默认 model_client 指向全局实例"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )
        from foundation.ai.agent.generate.model_generate import generate_model_client

        inst = GrammarCheckReviewer()
        assert inst.model_client is generate_model_client

    # --------------------------------------------------------
    # 7. 全链路集成
    # --------------------------------------------------------
    def test_full_chain_success(self):
        """全链路成功：加载 prompt → 调用模型 → ReviewResult"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        reviewer.model_client.get_model_generate_invoke = AsyncMock(
            return_value="无明显问题"
        )

        async def run():
            result = await reviewer.check_grammar(
                trace_id="chain_ok",
                review_content="本方案编制依据GB50300-2013。",
            )
            assert reviewer.model_client.get_model_generate_invoke.awaited
            assert result.success is True
            assert result.execution_time is not None

        asyncio.run(run())

    def test_full_chain_error(self):
        """全链路异常：模型抛错 → ReviewResult(success=False)"""
        from core.construction_review.component.reviewers.sensitive_word_check import (
            GrammarCheckReviewer,
        )

        reviewer = GrammarCheckReviewer()
        reviewer.model_client = MagicMock()
        reviewer.model_client.get_model_generate_invoke = AsyncMock(
            side_effect=TimeoutError("超时")
        )

        async def run():
            result = await reviewer.check_grammar(
                trace_id="chain_err", review_content="测试。"
            )
            assert result.success is False
            assert result.error_message is not None

        asyncio.run(run())


# ============================================================
# 入口
# ============================================================
if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-v", "--capture=no"]))