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

"""作业 3 转换校验与图级行为测试。

分工回顾：所有 stage_code（含拍照失败 → 0005、连续无回复 → 0004）
都由（假）LLM 提议，``transitions.is_legal_transition`` 按白名单校验；
代码只做事件解析与无回复计数等上下文簿记。

分组：
- 校验器语义：不依赖白名单内容，现在就应该通过；
- 提议为空/非法时保持阶段：现在就应该通过；
- 白名单内容：``LEGAL_TRANSITIONS`` 填写完成前会失败，属预期状态。
"""

import pytest

import src.agent.graph as graph_mod
from src.agent.graph import build_graph
from src.agent.transitions import (
    TERMINAL_STAGES,
    TransitionSpec,
    is_legal_transition,
)


def invoke(graph, thread_id, text, **initial):
    payload = {"session_id": thread_id, "input_text": text}
    payload.update(initial)
    return graph.ainvoke(
        payload,
        config={"configurable": {"thread_id": thread_id}},
    )


# ---------- 校验器语义（不依赖白名单内容） ----------


def test_same_stage_is_always_legal():
    assert is_legal_transition("2000", "2000") is True


def test_unlisted_move_is_rejected():
    # 白名单未登记的转换一律拒绝（越级跳步 / 跨阶段乱跳）
    assert is_legal_transition("2000", "2003") is False
    assert is_legal_transition("1002", "3001") is False


@pytest.mark.parametrize("stage", sorted(TERMINAL_STAGES))
def test_terminal_stage_never_transitions_out(stage):
    monkey_whitelist = [TransitionSpec(stage, frozenset({"1002"}))]
    import src.agent.transitions as transitions_mod

    original = transitions_mod.LEGAL_TRANSITIONS
    transitions_mod.LEGAL_TRANSITIONS = monkey_whitelist
    try:
        assert is_legal_transition(stage, "1002") is False
    finally:
        transitions_mod.LEGAL_TRANSITIONS = original


def test_whitelist_allows_listed_target(monkeypatch):
    import src.agent.transitions as transitions_mod

    monkeypatch.setattr(
        transitions_mod,
        "LEGAL_TRANSITIONS",
        [TransitionSpec("2000", frozenset({"2001"}), source_ref="test")],
    )
    assert is_legal_transition("2000", "2001") is True
    assert is_legal_transition("2000", "2002") is False  # 越级跳步拒绝
    assert is_legal_transition("2001", "2002") is False  # 未登记拒绝


# ---------- 提议为空/非法时保持阶段（现在即应通过） ----------


@pytest.mark.asyncio
async def test_valid_reply_resets_no_reply_count():
    graph = build_graph()
    await invoke(graph, "t-reset", "【用户无回复】", stage_code="1002")
    await invoke(graph, "t-reset", "追尾了")
    third = await invoke(graph, "t-reset", "【用户无回复】")
    assert third["stage_code"] == "1002"


@pytest.mark.asyncio
async def test_illegal_proposal_keeps_stage(monkeypatch):
    # LLM 提议非法跳步（2001 → 2003）：校验拒绝，状态不被污染
    graph = build_graph()
    monkeypatch.setattr(graph_mod, "_fake_llm_propose", lambda s, e: "2003")
    result = await invoke(graph, "t-illegal", "【拍摄完成】", stage_code="2001")
    assert result["stage_code"] == "2001"


@pytest.mark.asyncio
async def test_plain_continue_keeps_photo_stage():
    graph = build_graph()
    result = await invoke(graph, "t-keep", "继续", stage_code="2001")
    assert result["stage_code"] == "2001"


# ---------- 白名单内容（填好前失败，属预期） ----------


@pytest.mark.asyncio
async def test_photo_failed_once_goes_to_transfer():
    # 假 LLM 对拍照失败提议 0005；需白名单登记拍照阶段 → 0005
    # （本测试走 2011 → 0005）
    graph = build_graph()
    result = await invoke(
        graph,
        "t-failed",
        "【客户端连续3次拍摄识别失败：图片过于模糊】",
        stage_code="2011",
    )
    assert result["stage_code"] == "0005"


@pytest.mark.asyncio
async def test_second_no_reply_goes_to_0004():
    # 第二次无回复时假 LLM 提议 0004；需白名单登记 1002 → 0004
    graph = build_graph()
    first = await invoke(graph, "t-noreply", "【用户无回复】", stage_code="1002")
    assert first["stage_code"] == "1002"  # 第一次保持并计数
    second = await invoke(graph, "t-noreply", "【用户无回复】")
    assert second["stage_code"] == "0004"


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("source", "target"),
    [
        ("2000", "2001"),  # 单车拍照链
        ("2001", "2002"),
        ("2002", "2003"),
        ("2003", "2005"),
        ("2010", "2011"),  # 双车拍照链
        ("2011", "2012"),
        ("2012", "2013"),
        ("2013", "2014"),
        ("2014", "2015"),
    ],
)
async def test_photo_done_advances_photo_chain(source, target):
    # 假 LLM 提议 chain 下一格；白名单未登记该转换时保持原阶段
    graph = build_graph()
    result = await invoke(graph, f"t-{source}", "【拍摄完成】", stage_code=source)
    assert result["stage_code"] == target
    assert result["response_text"].startswith(f"<state>{target}</state>")


@pytest.mark.asyncio
@pytest.mark.parametrize("marker", ["【开始】", "【继续办理】"])
async def test_session_start_enters_collect_stage(marker):
    graph = build_graph()
    result = await invoke(graph, f"t-start-{marker}", marker, stage_code="1001")
    assert result["stage_code"] == "1002"