test_tutorial_04.py

"""教程 04 离线验证脚本：无需 API Key。"""

from __future__ import annotations

import importlib.util
import io
import sys
from pathlib import Path


if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8")


ROOT = Path(__file__).parent
EXAMPLES = ROOT / "examples" / "04-agent-workflow"


def load_example(filename: str):
    """从数字开头的示例文件名加载模块。"""
    path = EXAMPLES / filename
    module_name = f"tutorial04_{path.stem}"
    spec = importlib.util.spec_from_file_location(module_name, path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"无法加载示例：{path}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[module_name] = module
    spec.loader.exec_module(module)
    return module


def run_test(name: str, test_func) -> bool:
    print(f"\n【{name}】")
    print("-" * 60)
    try:
        test_func()
        print(f"✅ {name}通过")
        return True
    except Exception as exc:
        print(f"❌ {name}失败：{exc}")
        return False


react = load_example("01_react_agent.py")
plan_execute = load_example("02_plan_execute.py")
state_machine = load_example("03_langgraph_state_machine.py")
approval = load_example("04_persistence_human_in_loop.py")
ticket_workflow = load_example("05_ticket_workflow.py")


def test_safe_tools() -> None:
    assert react.safe_calculate("(3 + 4) * 2") == "14"
    assert react.safe_calculate("10 / 4") == "2.5"
    try:
        react.safe_calculate("__import__('os').getcwd()")
    except react.ToolError:
        pass
    else:
        raise AssertionError("危险表达式没有被拒绝")
    print("✓ 四则运算正确，函数调用被拒绝")


def test_react_agent() -> None:
    agent = react.ReActAgent(react.RuleBasedReasoner(), react.TOOLS)
    result = agent.run("请查询 TICKET-1001 的处理状态")
    assert result.stopped_reason == "finished"
    assert len(result.trace) == 1
    assert result.trace[0].action == "lookup_ticket"
    assert "网络组" in result.answer

    knowledge = agent.run("VPN 无法连接应该怎么办？")
    assert "刷新客户端配置" in knowledge.answer
    print("✓ 工单查询与知识检索路径正确")


def test_react_iteration_limit() -> None:
    class EndlessReasoner:
        def decide(self, query, trace, tool_names):
            return react.AgentDecision(
                thought="继续查询",
                kind="action",
                action="search_knowledge",
                action_input="不存在",
            )

    result = react.ReActAgent(
        EndlessReasoner(), react.TOOLS, max_iterations=2
    ).run("测试循环上限")
    assert result.stopped_reason == "max_iterations"
    assert len(result.trace) == 2
    print("✓ 达到 2 次上限后停止")


def test_plan_execute() -> None:
    executor = plan_execute.PlanExecutor(plan_execute.Planner(), plan_execute.TOOLS)
    state = executor.run("分析 TICKET-1001 并通知负责团队")
    assert len(state.completed_steps) == 3
    assert all(record.success for record in state.completed_steps)
    assert "通知已发送" in state.final_answer

    failed = executor.run("分析 TICKET-9999 并通知负责团队")
    assert len(failed.completed_steps) == 1
    assert not failed.completed_steps[0].success
    print("✓ 正常计划完成，关键事实失败后正确短路")


def make_graph_state(**overrides):
    state = state_machine.create_initial_state(
        "TICKET-2001", "VPN 故障", "单个用户 VPN 无法连接"
    )
    state.update(overrides)
    return state


def test_nodes_and_routes() -> None:
    state = make_graph_state()
    category = state_machine.classify_ticket(state)
    assert category["category"] == "network"

    assert (
        state_machine.route_after_assessment(
            make_graph_state(category="network", priority="P2")
        )
        == "resolve"
    )
    assert (
        state_machine.route_after_assessment(
            make_graph_state(category="network", priority="P1")
        )
        == "human_review"
    )

    retry = make_graph_state(status="retrying", retry_count=1, max_retries=2)
    assert state_machine.route_after_resolution(retry) == "resolve"
    retry["retry_count"] = 2
    assert state_machine.route_after_resolution(retry) == "human_review"
    print("✓ 分类、优先级和重试上限路由正确")


def test_ticket_validation_and_routing() -> None:
    valid = ticket_workflow.initial_state(
        {
            "ticket_id": "TICKET-4001",
            "title": "VPN 断开",
            "description": "用户无法连接公司 VPN",
            "requester": "张三",
        }
    )
    validated = ticket_workflow.validate_input(valid)
    assert validated["valid"] is True

    invalid = ticket_workflow.initial_state(
        {
            "ticket_id": "bad",
            "title": "x",
            "description": "短",
            "requester": "",
        }
    )
    rejected = ticket_workflow.validate_input(invalid)
    assert rejected["valid"] is False
    assert rejected["status"] == "rejected"

    valid.update(category="network", priority="P2", knowledge="运行手册")
    assert ticket_workflow.choose_handler(valid) == "auto_resolve"
    valid["priority"] = "P1"
    assert ticket_workflow.choose_handler(valid) == "assign_human"
    print("✓ Pydantic 校验与自动/人工分流正确")


def test_compiled_graphs() -> None:
    if state_machine.StateGraph is None:
        print("⚠️ 未安装 LangGraph，跳过编译图路径测试")
        return

    graph = state_machine.build_workflow()
    result = graph.invoke(
        state_machine.create_initial_state(
            "TICKET-2001", "VPN 间歇断开", "单个用户 VPN 间歇断开"
        )
    )
    assert result["status"] == "closed"
    assert result["retry_count"] == 2

    ticket_graph = ticket_workflow.build_workflow()
    ticket_result = ticket_graph.invoke(
        ticket_workflow.initial_state(
            {
                "ticket_id": "TICKET-4001",
                "title": "VPN 间歇断开",
                "description": "我的 VPN 每隔十分钟会间歇断开一次",
                "requester": "张三",
            }
        )
    )
    assert ticket_result["status"] == "resolved"
    assert ticket_result["attempts"] == 2
    assert ticket_result["trace"][-1] == "notify"
    print("✓ LangGraph 条件分支和重试循环执行正确")


def test_human_interrupt() -> None:
    if approval.StateGraph is None:
        print("⚠️ 未安装 LangGraph，跳过人工暂停/恢复测试")
        return

    graph = approval.build_workflow()
    config = {"configurable": {"thread_id": "tutorial-04-test-approval"}}
    graph.invoke(approval.initial_state("TICKET-3001", "执行生产退款"), config)
    snapshot = graph.get_state(config)
    assert "human_review" in snapshot.next

    graph.update_state(
        config,
        {"approved": True, "review_comment": "测试批准"},
    )
    result = graph.invoke(None, config)
    assert result["status"] == "completed"
    assert "review:approved" in result["trace"]
    print("✓ 工作流可以暂停、写入审批结果并恢复")


if __name__ == "__main__":
    print("=" * 80)
    print("教程 04 代码验证")
    print("=" * 80)

    tests = [
        ("安全工具", test_safe_tools),
        ("ReAct 执行循环", test_react_agent),
        ("ReAct 循环上限", test_react_iteration_limit),
        ("Plan-Execute", test_plan_execute),
        ("节点与条件路由", test_nodes_and_routes),
        ("工单校验与分流", test_ticket_validation_and_routing),
        ("LangGraph 完整路径", test_compiled_graphs),
        ("人工暂停与恢复", test_human_interrupt),
    ]
    results = [(name, run_test(name, func)) for name, func in tests]

    print("\n" + "=" * 80)
    print("测试总结")
    print("=" * 80)
    for name, passed in results:
        print(f"{'✅ PASS' if passed else '❌ FAIL'} - {name}")

    if not all(passed for _, passed in results):
        sys.exit(1)
    print("\n🎉 所有可用测试通过！教程 04 代码验证完成。")