ZNJJ-api-server/test/agent/test_confirm.py

"""作业 5 验收测试：信息核实节点（LLM 决策版）。

测试用脚本化 FakeLLM 模拟“行为良好的模型”：按用户最新输入
返回对应的带前缀回复；字段提取调用按提示词返回 JSON。
断言仍是行为级的：stage 推进、person_fields 落库、person_index
切换、模糊不计入失败。
"""

import json

import pytest

from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm
from src.agent.llm import FakeLLMService


def latest_input(message: str) -> str:
    return message.split("【用户最新输入】")[-1].strip()


def make_llm(dialogue_script, extraction_script=None):
    """dialogue_script(latest_input) -> 带前缀回复；
    extraction_script(extraction_system_prompt) -> JSON 字符串。"""

    def responder(system, user):
        if system.startswith("【字段提取】"):
            if extraction_script:
                return extraction_script(system)
            return "{}"
        return dialogue_script(latest_input(user))

    return FakeLLMService(responder)


async def invoke(graph, thread_id, text, llm=None, **initial):
    payload = {"session_id": thread_id, "input_text": text}
    payload.update(initial)
    return await graph.ainvoke(
        payload,
        config={"configurable": {"thread_id": thread_id}},
        context=GraphDeps(llm=llm or workflow_fake_llm()),
    )


# ---------- 单车核实 ----------


@pytest.mark.asyncio
async def test_entry_consumes_pending_identity_question():
    # 用户确认车主身份 → 模型直接问姓名（身份问题的提问发生在
    # 拍照节点进入 3001 的同一回复里，这里不得重复）
    llm = make_llm(
        lambda q: "<state>3001</state>好的，请问您的姓名是什么？"
    )
    graph = build_graph()
    result = await invoke(
        graph, "t-entry", "是的，我是车主", llm=llm,
        stage_code="3001", client_mode="direct",
    )

    assert result["stage_code"] == "3001"
    assert "姓名" in result["response_text"]
    assert "车主" not in result["response_text"]


@pytest.mark.asyncio
async def test_single_confirm_full_chain_to_0000():
    def dialogue(q):
        if q == "是的":
            return "<state>3001</state>好的，请问您的姓名是什么？"
        if q == "张三":
            return "<state>3001</state>张三您好，请问您的身份证号码后四位是否正确？"
        if q == "是对的":
            return "<state>3001</state>请问您的手机号码后四位是否正确？"
        return "<state>0000</state>好的，信息已全部确认无误。感谢您的配合。"

    def extraction(system):
        if "【用户输入】张三" in system:
            return json.dumps({"xm1": "张三"}, ensure_ascii=False)
        return "{}"

    llm = make_llm(dialogue, extraction)
    graph = build_graph()

    r1 = await invoke(graph, "t-chain", "是的", llm=llm, stage_code="3001")
    assert "姓名" in r1["response_text"]
    r2 = await invoke(graph, "t-chain", "张三", llm=llm)
    assert "身份证" in r2["response_text"]
    assert r2["person_fields"]["xm1"] == "张三"
    r3 = await invoke(graph, "t-chain", "是对的", llm=llm)
    assert "手机" in r3["response_text"]
    r4 = await invoke(graph, "t-chain", "没错", llm=llm)
    assert r4["stage_code"] == "0000"
    assert r4["response_text"].startswith("<state>0000</state>")


@pytest.mark.asyncio
async def test_deny_collects_correct_value():
    llm = make_llm(
        lambda q: "<state>3001</state>抱歉，请告诉我您正确的身份证号码后四位。",
        lambda system: "{}",
    )
    graph = build_graph()
    result = await invoke(
        graph, "t-deny", "不对", llm=llm,
        stage_code="3001", person_fields={"xm1": "张三"}, client_mode="direct",
    )

    assert result["stage_code"] == "3001"
    assert "身份证" in result["response_text"]
    assert result["person_fields"].get("sfzmwh1") in (None, "")


@pytest.mark.asyncio
async def test_vague_answer_stays_without_counting_failure():
    llm = make_llm(
        lambda q: "<state>3001</state>请问您是否是这辆车的车主？请回答“是”或“不是”。",
        lambda system: "{}",
    )
    graph = build_graph()
    result = await invoke(
        graph, "t-vague", "嗯", llm=llm,
        stage_code="3001", clarify_count=0, client_mode="direct",
    )

    assert result["stage_code"] == "3001"
    assert "车主" in result["response_text"]
    assert (result.get("clarify_count") or 0) == 0


# ---------- 双车核实 ----------


@pytest.mark.asyncio
async def test_dual_person_index_switches_after_handoff():
    llm = make_llm(
        lambda q: "<state>3002</state>您好。请问这辆车是由您驾驶的吗？",
        lambda system: json.dumps({"person_index": 2}),
    )
    graph = build_graph()
    result = await invoke(
        graph, "t-dual", "移交完毕", llm=llm,
        stage_code="3002", person_index=1,
        person_fields={"xm1": "张三"}, client_mode="direct",
    )

    assert result["stage_code"] == "3002"
    assert result["person_index"] == 2
    assert "驾驶" in result["response_text"]


@pytest.mark.asyncio
async def test_dual_second_person_fields_isolated():
    def dialogue(q):
        if q == "是的":
            return "<state>3002</state>好的，驾驶员您好。请问您的姓名是什么？"
        return "<state>3002</state>请问您的身份证号码后四位是否正确？"

    def extraction(system):
        if "【用户输入】李四" in system:
            return json.dumps({"xm2": "李四"}, ensure_ascii=False)
        return "{}"

    llm = make_llm(dialogue, extraction)
    graph = build_graph()

    await invoke(
        graph, "t-dual2", "是的", llm=llm,
        stage_code="3002", person_index=2,
        person_fields={"xm1": "张三", "sjhm1": "13800001111"},
        client_mode="direct",
    )
    result = await invoke(graph, "t-dual2", "李四", llm=llm)

    assert result["person_fields"]["xm2"] == "李四"
    assert result["person_fields"]["xm1"] == "张三"


@pytest.mark.asyncio
async def test_dual_completion_goes_to_0000():
    llm = make_llm(
        lambda q: "<state>0000</state>好的，两位驾驶员的信息均已核实完毕。感谢配合。",
        lambda system: "{}",
    )
    graph = build_graph()
    result = await invoke(
        graph, "t-dual-done", "没错", llm=llm,
        stage_code="3002", person_index=2,
        person_fields={"xm1": "张三", "xm2": "李四"},
        client_mode="direct",
    )

    assert result["stage_code"] == "0000"