cfc/cfc-langgraph/tests_contract/test_growth_report_contract.py

"""SC-B-04 / SC-B-11 contract tests: growth-report interpretation & follow-up routing.

Pure structural tests (no real LLM, no network, no DB):

- SC-B-04 coach.growthReport.interpret:
  analysis_graph.create_analysis_graph end-to-end offline run — gather_data collects
  report/survey/dimension data via tools, analyze node feeds them into the prompt and
  returns the LLM interpretation. We stub ChatOpenAI + the three report tools.
- SC-B-11 coach.growthReport.ask:
  a report follow-up question ("再解释一下报告里的免疫力指标") must be routed by
  IntentClassifier to Intent.ANALYSIS (report interpretation), while LLM failure or
  unknown output falls back to Intent.CHAT (no false behavior).

NOTE: this test lives in tests_contract/ (not tests/) on purpose — tests/conftest.py
imports app.main, which drags in faster_whisper->ctranslate2 native libs that are not
reproducibly loadable in this venv (native access-violation 0xC0000005). This contract
suite needs only app.graphs.analysis_graph + app.agents.intent_classifier, which import
cleanly, so it runs in a directory without that conftest.
"""
import asyncio
import os
import sys
from unittest.mock import patch

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.agents.intent_classifier import CLASSIFY_PROMPT, Intent, IntentClassifier


class FakeLLM:
    """Offline LLM stub returning fixed content (or raising), mirroring langchain ainvoke shape."""

    def __init__(self, content=""):
        self._content = content
        self._exc = None

    def failure(self, exc):
        self._exc = exc
        return self

    async def ainvoke(self, *args, **kwargs):
        if self._exc:
            raise self._exc
        return type("Resp", (), {"content": self._content})()


class FakeBoundLLM:
    def __init__(self, content):
        self._content = content

    async def ainvoke(self, messages):
        return type("Resp", (), {"content": self._content})()


class FakeChatOpenAI:
    def __init__(self, content="report interpretation stub", **kwargs):
        self._content = content

    def bind_tools(self, tools):
        return FakeBoundLLM(self._content)


class FakeTool:
    """Offline stand-in for a langchain StructuredTool: exposes only ainvoke."""

    def __init__(self, fn):
        self._fn = fn

    async def ainvoke(self, arg):
        return self._fn(arg)


class TestGrowthReportInterpretB04:
    """SC-B-04 报告智能解读: analysis graph offline contract."""

    def test_analysis_graph_offline_run_returns_interpretation(self, monkeypatch):
        import app.graphs.analysis_graph as g

        monkeypatch.setattr(g, "ChatOpenAI", FakeChatOpenAI)

        async def fake_get_prompt(*a, **k):
            return g.ANALYSIS_SYSTEM_PROMPT

        monkeypatch.setattr(g, "get_prompt", fake_get_prompt)

        monkeypatch.setattr(g, "get_report_detail", FakeTool(lambda arg: {"gut_health_score": 60, "immunity": "偏低"}))
        monkeypatch.setattr(g, "get_survey_data", FakeTool(lambda arg: {"sleep": "不足"}))
        monkeypatch.setattr(g, "get_dimension_scores", FakeTool(lambda arg: {"免疫力": 55, "肠胃": 70}))

        graph = g.create_analysis_graph()

        result = asyncio.run(graph.ainvoke({
            "report_id": 200,
            "user_id": 9,
            "focus": "immunity",
            "report_data": None,
            "survey_data": None,
            "dimension_scores": None,
            "analysis": None,
            "recommendations": [],
        }, config={"configurable": {"thread_id": "b04-offline"}}))

        # gather_data ran via stubbed tools, analyze returned stub LLM interpretation
        assert result.get("analysis") == FakeChatOpenAI()._content
        assert result.get("recommendations") is not None

    def test_analysis_prompt_covers_interpretation_contract(self):
        from app.graphs.analysis_graph import ANALYSIS_SYSTEM_PROMPT

        assert "解读" in ANALYSIS_SYSTEM_PROMPT
        assert "总体评估" in ANALYSIS_SYSTEM_PROMPT  # 输出格式: 总体评估 -> 分项分析 -> 改善建议
        assert "异常指标" in ANALYSIS_SYSTEM_PROMPT  # 关注异常指标给出改善建议

    def test_analysis_state_shape(self):
        from app.graphs.analysis_graph import AnalysisState

        state = AnalysisState(
            report_id=200, user_id=9, focus=None,
            report_data=None, survey_data=None,
            dimension_scores=None, analysis=None, recommendations=[],
        )
        assert state["report_id"] == 200
        assert "analysis" in state and "recommendations" in state


class TestGrowthReportAskB11:
    """SC-B-11 报告细节追问: intent routing contract."""

    def _classifier(self, llm):
        classifier = IntentClassifier()
        classifier.llm = llm  # offline stub replaces hard-created ChatOpenAI
        return classifier

    def test_report_followup_routed_to_analysis(self, monkeypatch):
        with patch("app.agents.intent_classifier.get_prompt", return_value=""):
            classifier = self._classifier(FakeLLM(content="analysis"))
            intent = asyncio.run(classifier.classify("报告里的免疫力指标再解释一下"))
        assert intent == Intent.ANALYSIS, "报告追问应路由到分析（解读）"

    def test_explicit_analysis_keyword_routed(self, monkeypatch):
        with patch("app.agents.intent_classifier.get_prompt", return_value=""):
            classifier = self._classifier(FakeLLM(content="analysis"))
            intent = asyncio.run(classifier.classify("帮我分析这份成长报告"))
        assert intent == Intent.ANALYSIS

    def test_classify_prompt_has_report_routing_rules(self):
        assert "analysis" in CLASSIFY_PROMPT
        assert "报告解读" in CLASSIFY_PROMPT

    def test_llm_failure_falls_back_to_chat(self, monkeypatch):
        with patch("app.agents.intent_classifier.get_prompt", return_value=""):
            llm = FakeLLM().failure(RuntimeError("llm down"))
            classifier = self._classifier(llm)
            intent = asyncio.run(classifier.classify("今天天气不错"))
        assert intent == Intent.CHAT, "LLM 异常时应降级 chat 而非抛出"

    def test_unknown_llm_output_falls_back_to_chat(self, monkeypatch):
        with patch("app.agents.intent_classifier.get_prompt", return_value=""):
            classifier = self._classifier(FakeLLM(content="banana_intent"))
            intent = asyncio.run(classifier.classify("随便聊聊"))
        assert intent == Intent.CHAT, "未识别意图应降级 chat"