多后端测试的根本矛盾

LangGraph 的 checkpointer 设计是"接口归框架,实现归用户"。这个决策带来了巨大灵活性,但也制造了一个测试上的根本矛盾:框架支持 Memory、SQLite(同步/异步/AES 加密)、Postgres(同步/pipeline/连接池/异步各变体)、Redis 等十余种后端配置,而这些后端必须在语义层面表现完全一致——用同一个 thread_id 存进去的状态,用任何后端都要能原样取出来。

最直接的做法是为每个后端写一套测试。但这条路很快就走不通了:十几个后端意味着十几倍的测试维护成本,一旦协议规范有变动,每套测试都要改一遍,而且改漏了也不会立刻报错。

另一种常见方案是用类继承——定义一个 BaseCheckpointerTest,把公共测试逻辑放在基类方法里,各后端的测试类继承并重写 make_checkpointer() 工厂方法。这在初期看起来整洁,但随着测试维度增加,问题就暴露出来了。假设你需要测试 checkpointer 和 store 的组合,或者 checkpointer 和 cache 的组合,类继承就会陷入多重继承的泥潭,组合出来的类层次树越来越难以追踪。类继承擅长表达"是什么",却拙于表达"和什么组合"。

LangGraph 采用了不同的解法:pytest fixture 参数化。

Pytest Fixture 参数化:把"支持哪些后端"变成运行时决策

看一下 libs/langgraph/tests/conftest.py 中的核心 fixture:

NO_DOCKER = os.getenv("NO_DOCKER", "false") == "true"
@pytest.fixture(
    scope="function",
    params=[
        "memory",
        "sqlite",
        "sqlite_aes",
    ]
    if NO_DOCKER
    else [
        "memory",
        "memory_migrate_sends",
        "sqlite",
        "sqlite_aes",
        "postgres",
        "postgres_pipe",
        "postgres_pool",
    ],
)
def sync_checkpointer(
    request: pytest.FixtureRequest,
) -> Iterator[BaseCheckpointSaver]:
    checkpointer_name = request.param
    if checkpointer_name == "memory":
        with _checkpointer_memory() as checkpointer:
            yield checkpointer
    elif checkpointer_name == "sqlite":
        with _checkpointer_sqlite() as checkpointer:
            yield checkpointer
    elif checkpointer_name == "postgres":
        with _checkpointer_postgres() as checkpointer:
            yield checkpointer
    # ...

这个 fixture 有几个设计决策值得细看。

第一,params 列表在定义时根据 NO_DOCKER 环境变量静态确定。这意味着后端列表的选择发生在 pytest 收集阶段,而不是每个测试函数运行时。如果 Docker 未启动,直接用 NO_DOCKER=true 运行,Postgres 相关的参数根本不会进入 params 列表,pytest 不会尝试连接数据库,测试也不会因为连接失败而报错,而是压根不生成这些测试用例。本地开发完全不受影响。

第二,scope="function" 确保每个测试函数都获得一个独立的 checkpointer 实例。这是故意的:共享 checkpointer 会让不同测试之间产生状态污染,而状态污染是最难调试的问题之一,因为失败现象取决于测试执行顺序。

第三,测试函数本身对后端完全无感知:

def test_time_travel(sync_checkpointer: BaseCheckpointSaver) -> None:
    # 这里的代码对 sync_checkpointer 是 Memory 还是 Postgres 一无所知
    # 也不需要知道
    graph = builder.compile(checkpointer=sync_checkpointer)
    # ...
graph LR
    subgraph fixture["sync_checkpointer fixture (params=[memory, sqlite, postgres, ...])"]
        P1["memory"]
        P2["sqlite"]
        P3["postgres"]
        P4["postgres_pool"]
    end
    fixture -->|"自动展开"| T1["test_time_travel[memory]"]
    fixture -->|"自动展开"| T2["test_time_travel[sqlite]"]
    fixture -->|"自动展开"| T3["test_time_travel[postgres]"]
    fixture -->|"自动展开"| T4["test_time_travel[postgres_pool]"]

pytest 会自动将这个函数扩展为 test_time_travel[memory]、test_time_travel[sqlite]、test_time_travel[postgres] 等多个测试用例。新增一个后端只需要在 params 列表里加一行,并在 fixture 的 if-elif 里加一个分支,所有依赖 sync_checkpointer 的测试函数都自动覆盖了新后端。

更重要的是多维度组合。libs/langgraph/tests/conftest.py 里同时定义了 sync_checkpointer、async_checkpointer、sync_store、async_store、cache 这几个参数化 fixture。测试函数可以同时声明多个参数化 fixture:

async def test_graph_with_store_and_cache(
    async_checkpointer: BaseCheckpointSaver,
    async_store: BaseStore,
    cache: BaseCache,
) -> None:
    ...

pytest 会取这三个 fixture 的笛卡尔积,生成所有合法的后端组合测试。类继承在这里完全束手无策——你没有办法用继承优雅地表达三个维度的全组合,但 fixture 参数化自然就支持。

checkpoint-conformance:把测试变成协议文档

libs/checkpoint-conformance 是一个独立的包,专门存放 checkpointer 协议的合规测试。它存在的理由非常清晰:LangGraph 框架的 checkpointer 接口是公开的,任何人都可以实现自己的后端——可能是 DynamoDB、MongoDB 或者自定义的分布式存储。这些第三方实现需要有一种机制来验证自己是否正确。