Eric Wang
8323d53a70
Architecture (mirrors workflow/20261016 chatNode structure): - events.py: client markers parsed to typed ClientEvent (sources verified against workflow JSON); plain text like '拍好了' is not an event - transitions.py: LEGAL_TRANSITIONS whitelist drafted from chatNode prompts and Condition edges (16 entries, each with source_ref); is_legal_transition rejects unlisted/terminal-violating moves - graph.py: multi-node graph (collect/single_photo/dual_photo/... matching workflow chatNodes), route_by_stage conditional edges, _fake_llm_propose simulates LLM stage-code proposals (assignment 4 replaces with real LLM); code only keeps context bookkeeping (no_reply_count) and validation - state.py: expanded domain fields (pending_question, accident/ person fields, clarify/no-reply counts, turn_patch, state_version) Key finding: 单车拍照#6/双车拍照#7 are browser_addon branches ending at 0000, so 2005/2015 allow both 300x (direct) and 0000. Verified: pytest 87 passed (24 transition/event tests newly green). Known limit: stage proposals come from a fake LLM; per-stage scripts and real LLM integration are assignments 4-5.
156 lines
5.2 KiB
Python
156 lines
5.2 KiB
Python
"""作业 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"
|