test_tutorial_05.py

"""教程 05 验证脚本：无 API Key，可选真实 LangChain 路径。"""

from __future__ import annotations

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

from pydantic import ValidationError


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" / "05-langchain-practice"


def load_example(filename: str):
    path = EXAMPLES / filename
    module_name = f"tutorial05_{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


lcel = load_example("01_lcel_runnable.py")
history = load_example("02_conversation_history.py")
callbacks = load_example("03_callbacks_and_tracing.py")
components = load_example("04_custom_components.py")
customer_service = load_example("05_customer_service_chain.py")


def test_lcel_business_functions() -> None:
    normalized = lcel.normalize_input({"question": "  VPN   无法连接  "})
    assert normalized["question"] == "VPN 无法连接"

    ticket = lcel.run_without_langchain({"question": "查询 TICKET-1001"})
    assert ticket["intent"] == "ticket"
    assert "网络组" in ticket["answer"]

    policy = lcel.run_without_langchain({"question": "退钱需要多久？"})
    assert policy["intent"] == "policy"
    assert "7 天" in policy["answer"]

    try:
        lcel.normalize_input({"question": "  "})
    except ValueError:
        pass
    else:
        raise AssertionError("空问题未被拒绝")
    print("✓ 归一化、分类、检索、分支和输入拒绝路径正确")


def test_local_conversation_history() -> None:
    manager = history.LocalConversationManager()
    first = manager.invoke("user-a", "帮我查询 TICKET-1001")
    second = manager.invoke("user-a", "它什么时候有结果？")
    isolated = manager.invoke("user-b", "它什么时候有结果？")

    assert "TICKET-1001" in first
    assert "TICKET-1001" in second
    assert "TICKET-1001" not in isolated
    assert len(manager.histories["user-a"]) == 4
    assert len(manager.histories["user-b"]) == 2
    print("✓ 同一会话可引用上一轮，不同会话完全隔离")


def test_trace_recorder() -> None:
    recorder = callbacks.TraceRecorder()
    result = callbacks.run_without_langchain({"question": "查询工单"}, recorder)
    assert result["intent"] == "ticket"
    assert [event.event for event in recorder.events] == ["start", "end"]
    assert recorder.events[-1].elapsed_ms is not None

    error_recorder = callbacks.TraceRecorder()
    try:
        callbacks.run_without_langchain({"question": ""}, error_recorder)
    except ValueError:
        pass
    else:
        raise AssertionError("异常路径没有抛出错误")
    assert [event.event for event in error_recorder.events] == ["start", "error"]
    assert "不能为空" in (error_recorder.events[-1].error or "")
    print("✓ 成功路径记录耗时，失败路径记录错误")


def test_custom_components_logic() -> None:
    documents = components.keyword_search("VPN 网络连接失败", k=2)
    assert documents
    assert documents[0].metadata["source"] == "vpn-runbook"

    intent = components.parse_ticket_intent(
        '{"category":"network","priority":"P2","summary":"VPN 连接失败"}'
    )
    assert intent.category == "network"

    try:
        components.parse_ticket_intent(
            '{"category":"unknown","priority":"P9","summary":"x"}'
        )
    except ValidationError:
        pass
    else:
        raise AssertionError("非法枚举和过短摘要未被拒绝")

    try:
        components.parse_ticket_intent("not-json")
    except json.JSONDecodeError:
        pass
    else:
        raise AssertionError("非法 JSON 未被拒绝")
    print("✓ 关键词检索、严格 Parser 和 Pydantic 校验正确")


def test_local_customer_service() -> None:
    assistant = customer_service.LocalCustomerAssistant()
    first = assistant.invoke("customer-a", "查询订单 ORD1001")
    follow_up = assistant.invoke("customer-a", "它什么时候到？")
    isolated = assistant.invoke("customer-b", "它什么时候到？")
    refund = assistant.invoke("customer-b", "如何退款？")

    assert "顺丰" in first
    assert "ORD1001" in follow_up
    assert "请提供订单号" in isolated
    assert "7 天" in refund
    print("✓ 订单指代、政策问答和客户隔离路径正确")


def test_real_lcel_chain() -> None:
    if lcel.RunnableLambda is None:
        print("⚠️ 未安装 LangChain，跳过真实 LCEL 测试")
        return

    chain = lcel.build_chain()
    result = chain.invoke({"question": "查询 TICKET-1001"})
    assert result["intent"] == "ticket"
    assert "网络组" in result["answer"]

    batch = chain.batch(
        [
            {"question": "退款政策是什么？"},
            {"question": "VPN 无法连接怎么办？"},
        ]
    )
    assert [item["intent"] for item in batch] == ["policy", "knowledge"]
    print("✓ RunnableParallel、RunnableBranch、invoke 和 batch 正确")


def test_real_message_history() -> None:
    if history.RunnableWithMessageHistory is None:
        print("⚠️ 未安装 LangChain，跳过真实 History 测试")
        return

    bot, store = history.build_chatbot()
    bot.invoke({"input": "查询 TICKET-1001"}, history.session_config("user-a"))
    follow_up = bot.invoke(
        {"input": "它什么时候有结果？"}, history.session_config("user-a")
    )
    isolated = bot.invoke(
        {"input": "它什么时候有结果？"}, history.session_config("user-b")
    )

    assert "TICKET-1001" in follow_up.content
    assert "TICKET-1001" not in isolated.content
    assert len(store["user-a"].messages) == 4
    assert len(store["user-b"].messages) == 2
    print("✓ RunnableWithMessageHistory 自动写入并按 session 隔离")


def test_real_callbacks() -> None:
    if callbacks.RunnableLambda is None:
        print("⚠️ 未安装 LangChain，跳过真实 Callback 测试")
        return

    handler = callbacks.WorkflowCallback()
    chain = callbacks.build_chain()
    chain.invoke({"question": "查询工单"}, config={"callbacks": [handler]})
    assert any(event.event == "start" for event in handler.recorder.events)
    assert any(event.event == "end" for event in handler.recorder.events)

    error_handler = callbacks.WorkflowCallback()
    try:
        chain.invoke({"question": ""}, config={"callbacks": [error_handler]})
    except ValueError:
        pass
    else:
        raise AssertionError("真实 Chain 异常未向调用方传播")
    assert any(event.event == "error" for event in error_handler.recorder.events)
    print("✓ 嵌套 Runnable 的成功与错误 Callback 均触发")


def test_real_custom_components() -> None:
    if components.KeywordRetriever is None:
        print("⚠️ 未安装 LangChain，跳过真实自定义组件测试")
        return

    documents = components.build_retriever().invoke("VPN 网络连接失败")
    assert documents[0].metadata["source"] == "vpn-runbook"

    intent = components.build_parser().invoke(
        '{"category":"network","priority":"P2","summary":"VPN 连接失败"}'
    )
    assert intent.priority == "P2"
    print("✓ 自定义 BaseRetriever 与 BaseOutputParser 可通过 invoke 调用")


def test_real_customer_service_chain() -> None:
    if customer_service.RunnableWithMessageHistory is None:
        print("⚠️ 未安装 LangChain，跳过真实客服 Chain 测试")
        return

    assistant, store = customer_service.build_assistant()
    metrics = customer_service.MetricsCallback()
    config_a = customer_service.session_config("customer-a", metrics)
    assistant.invoke({"input": "查询订单 ORD1001"}, config_a)
    follow_up = assistant.invoke({"input": "它什么时候到？"}, config_a)
    isolated = assistant.invoke(
        {"input": "它什么时候到？"},
        customer_service.session_config("customer-b", metrics),
    )

    assert "ORD1001" in follow_up.content
    assert "请提供订单号" in isolated.content
    assert len(store["customer-a"].messages) == 4
    assert metrics.started > 0
    assert metrics.completed > 0
    assert metrics.failed == 0
    print("✓ LCEL、检索、History 与 Callback 完整组合正确")


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

    tests = [
        ("LCEL 业务函数", test_lcel_business_functions),
        ("本地会话历史", test_local_conversation_history),
        ("Trace Recorder", test_trace_recorder),
        ("自定义组件逻辑", test_custom_components_logic),
        ("本地客服系统", test_local_customer_service),
        ("真实 LCEL Chain", test_real_lcel_chain),
        ("真实 Message History", test_real_message_history),
        ("真实 Callback", test_real_callbacks),
        ("真实自定义组件", test_real_custom_components),
        ("真实客服 Chain", test_real_customer_service_chain),
    ]
    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🎉 所有可用测试通过！教程 05 代码验证完成。")