跳转到内容

测试策略:从单元测试到大规模集成测试

测试策略:从单元测试到大规模集成测试

学习目标

本章要解决什么问题:

  • LangGraph 的 pytest 测试组织方式
  • 同步/异步双版本测试的维护策略
  • 可插拔测试固定装置(conftest)的设计
  • 大规模集成测试的场景覆盖方法

项目实践

测试目录结构

libs/langgraph/tests/
├── conftest_checkpointer.py # 可插拔的 checkpointer 测试固定装置
├── conftest_store.py # 可插拔的 store 测试固定装置
├── conftest.py # 通用 pytest 配置
├── agents.py # 测试 Agent 辅助类
├── messages.py # 测试消息辅助类
├── any_str.py # 模糊匹配器
├── any_int.py # 模糊匹配器
├── test_pregel.py # Pregel 核心测试(同步)
├── test_pregel_async.py # Pregel 核心测试(异步)
├── test_channels.py # Channel 单元测试
├── test_interruption.py # 中断测试
├── test_retry.py # 重试策略测试
├── test_time_travel.py # 时间旅行测试(同步)
├── test_time_travel_async.py # 时间旅行测试(异步)
├── test_large_cases_async.py # 大规模集成测试
├── test_remote_graph.py # 远程图测试
├── test_runtime.py # 运行时测试
├── test_serde_allowlist.py # 序列化白名单测试
├── test_stream_*.py # 多种流模式测试
├── test_subgraph_persistence.py # 子图持久化测试
└── test_subgraph_persistence_async.py

可插拔测试固定装置

conftest_checkpointer.py 定义了通用的 checkpointer 测试模板:

@pytest.fixture
def checkpointer_memory():
yield InMemorySaver()
@pytest.fixture
def checkpointer_sqlite():
yield SqliteSaver(conn)

每个后端包(checkpoint-sqlitecheckpoint-postgres)继承 checkpoint-conformance 的测试定义,注入自己的 checkpointer 实例。

好处:测试逻辑只写一次,所有后端共享验证。

同步/异步双版本测试

LangGraph 的核心执行路径同时支持同步和异步,因此大量测试有双版本:

# test_pregel.py(同步)
def test_invoke():
result = app.invoke({"input": "hello"})
# test_pregel_async.py(异步)
async def test_ainvoke():
result = await app.ainvoke({"input": "hello"})

维护成本:每次新增测试需要同步和异步两个版本,但保证了两种 API 的一致性。

模糊匹配器

any_str.pyany_int.py 定义了测试辅助匹配器:

# 不关心具体的 checkpoint_id 值,只要是非空字符串即可
assert result["checkpoint_id"] == AnyStr()

这在测试 checkpoint 相关功能时非常有用——ID 是随机生成的,但测试需要断言字段存在。

大规模集成测试

test_large_cases_async.py 包含复杂的端到端场景:

  • 多节点并行执行
  • 子图嵌套
  • 中断 + 恢复
  • 流式输出验证

这些测试验证多个子系统协同工作的正确性。

测试运行方式

Terminal window
cd libs/langgraph && make test # 运行所有测试
cd libs/langgraph && TEST=tests/test_pregel.py make test # 运行特定文件

陷阱与对策

测试固定装置不匹配

问题:在 checkpoint-sqlite 中使用 memory saver 的固定装置,测试不生效。

规避:确保每个后端包继承正确的 conftest fixture。

异步测试的事件循环冲突

问题:多个异步测试共享同一个事件循环,状态互相影响。

规避:使用 pytest-asyncioscope="function" 模式,每个测试独立事件循环。

设计取舍

优势

  • 一致性:所有后端共享同一套测试逻辑
  • 全面:同步/异步双版本覆盖两种 API
  • 可扩展:模糊匹配器和辅助类降低测试编写成本

代价

  • 维护成本:同步/异步双版本测试的工作量翻倍
  • 执行时间:测试量大,CI 运行时间长

替代方案

  • 属性测试(Property-based Testing):用 hypothesis 生成随机输入,但难以覆盖复杂的执行流程

参考来源

  • 源码验证: libs/langgraph/tests/ — 测试目录
  • 源码验证: libs/checkpoint-conformance/ — 一致性测试框架