cfc/cfc-langgraph/tests/test_graph.py

import json
from unittest.mock import patch, MagicMock
from src.schemas.questionnaire import GenerateRequest, Questionnaire, Question, Option


def _mock_llm_response():
    """返回合规 JSON 的 mock LLM 响应"""
    mock_msg = MagicMock()
    mock_msg.content = json.dumps({
        "version": 1,
        "questions": [
            {
                "id": "q1",
                "dimension": "trust",
                "direction": "positive",
                "weight": 1.0,
                "text": "您对小明的信任程度如何？",
                "options": [
                    {"id": "a", "score": 0},
                    {"id": "b", "score": 1},
                    {"id": "c", "score": 2},
                ],
            }
        ]
    }, ensure_ascii=False)
    return mock_msg


def _mock_llm_invalid_response():
    mock_msg = MagicMock()
    mock_msg.content = "这不是JSON"
    return mock_msg


class TestQuestionnaireSchema:
    def test_valid_questionnaire(self):
        q = Questionnaire(version=1, questions=[
            Question(id="q1", dimension="trust", text="test",
                     options=[Option(id="a", score=0)])
        ])
        assert q.validate_structure() is True

    def test_empty_questions_fails_validation(self):
        q = Questionnaire(version=1, questions=[])
        assert q.validate_structure() is False

    def test_question_without_options_or_scale_fails(self):
        q = Questionnaire(version=1, questions=[
            Question(id="q1", dimension="trust", text="test")
        ])
        assert q.validate_structure() is False


class TestGraph:
    def test_graph_valid_output(self):
        with patch("src.graphs.questionnaire.get_llm") as mock_get_llm:
            mock_llm = MagicMock()
            mock_llm.invoke.return_value = _mock_llm_response()
            mock_get_llm.return_value = mock_llm

            from src.graphs.questionnaire import build_questionnaire_graph
            graph = build_questionnaire_graph()
            result = graph.invoke({
                "request": GenerateRequest(member_name="小明", relationship_type="child"),
                "raw_response": "",
                "questionnaire": None,
                "error": None,
            })
            assert result["error"] is None
            assert result["questionnaire"] is not None
            assert len(result["questionnaire"]["questions"]) == 1

    def test_graph_invalid_llm_response_rejected(self):
        with patch("src.graphs.questionnaire.get_llm") as mock_get_llm:
            mock_llm = MagicMock()
            mock_llm.invoke.return_value = _mock_llm_invalid_response()
            mock_get_llm.return_value = mock_llm

            from src.graphs.questionnaire import build_questionnaire_graph
            graph = build_questionnaire_graph()
            result = graph.invoke({
                "request": GenerateRequest(member_name="小明", relationship_type="child"),
                "raw_response": "",
                "questionnaire": None,
                "error": None,
            })
            assert result["error"] is not None
            assert result["questionnaire"] is None