BLOGE 0.9.8-RC1 在线版 · 事实校验 2026-09-15 · English

第 18 章 —— 测试你的 Graph

承诺： 读完本章后，你将知道如何在 graph 进入生产环境之前验证它声明的契约 —— 通过 GraphResult 做断言、用测试替身隔离 operator、借助 GraphTestRunner 执行整张 graph、在运行前用 DslTestHelper 捕获 DSL 错误，并用确定性时间控制验证时间相关行为。

---

学习目标

1. 把 GraphResult 作为首要测试面：断言成功、以类型安全方式读取输出，并在行为变化时检查节点状态或耗时。
2. 使用 MockOperator 隔离 operator 行为，让测试聚焦在 graph 接线而不是外部 I/O。
3. 使用 GraphTestRunner 在专用测试夹具中执行 graph，并验证执行顺序、跳过行为和状态变化。
4. 使用 DslTestHelper 编译 DSL 片段，在运行前捕获 parse / compile 错误，并让测试层级与要阻止的失败相匹配。
5. 使用 ManualTimeSource、DeterministicTimerService 和 TestGraphEngine 对 retry、delay 和 timer 行为做确定性时间控制。

前置条件

• 第 2 章 —— 你的第一个 Graph —— 你需要知道 graph 如何加载与执行。
• 第 6 章——韧性设计 —— retry、timeout 和 fallback 会影响“正确行为”的断言方式。
• 第 7 章 —— 设计良好的 Operator —— 在隔离 operator 之前，你需要先理解 operator 契约。

源示例

文件	展示内容
GraphResultObservabilityTest.java	GraphResult 的执行标识、耗时和兼容性行为
GraphResultSafeAccessTest.java	测试中安全读取类型化输出的模式
MockOperator.java	返回、抛错、延迟、记录调用的测试替身
GraphTestRunner.java	带顺序、状态和执行日志断言的 graph 测试夹具
DslTestHelper.java	.bloge 片段的 parse / compile 断言
TestGraphEngine.java	内置 ManualTimeSource 与 timer 控制的确定性测试引擎
TestGraphEngineTest.java	用逻辑时间推进替代墙钟等待的最小示例
DeterministicRetryTest.java	用手动时间推进断言 retry/backoff 的测试

---

为什么这很重要

一张 graph 在代码评审里看起来没问题，仍然可能以三种昂贵的方式出错：

• 分支调整后跳过了原本应该执行的节点；
• fallback 掩盖了你本来想暴露的失败；
• DSL 重构在脑海里成立，但在真实编译器里并不合法。

好的 BLOGE 测试会在最便宜的层级把这些问题暴露出来。你不需要为了验证一个依赖边、一条 fallback 路径或一个类型化输出契约而启动 Spring 或调用真实服务。

BLOGE 的测试思路很简单：断言 graph 的契约，而不是偶然的实现细节。 先从 GraphResult 开始；当你需要更聚焦时，再用 MockOperator、GraphTestRunner 或 DslTestHelper 放大问题。

---

一次测试全绿，不等于四层声明都成立

订单测试可以完全通过，但更宽的声明仍然未知。每次读测试结果时，都要同时读 它实际覆盖的边界：

声明	所需证据	本章能提供什么
DSL 合法	真实 parser/compiler 结果	DslTestHelper
graph 满足声明的契约	状态、顺序、输出、时间与失败断言	GraphTestRunner 与 GraphResult
业务故事已获批准并仍被满足	Owner 批准的 Scenario、Policy、Fixture 与 Oracle	本章不能建立
发布声明已绑定源码并受治理	密封证据、精确源码身份、Requirement 与 ClaimCapability	本章不能建立

因此，OrderGraphTest 通过只支持一句精确结论：“在这些测试输入和替身下， graph 产生了这些可观察节点事实。”它不支持“订单业务正确”或“该版本具备发布资格”。

---

心智模型

把测试看作三层同心安全网：

每一层回答的问题都不同：

• DSL 编译测试 问："这个 graph 定义合法吗？"
• Graph 测试 问："工作流满足这个测试声明的契约吗？"
• Operator 单元测试 问："这个 operator 的输入输出转换对吗？"

一旦测试失败，所处层级就会告诉你下一步该往哪里看。

---

第一个可运行的示例

这个示例展示了最常见的生产需求：运行一个 graph，断言它成功了，检查输出，并验证逐节点的耗时。以下代码是纯 JUnit 5，除了 bloge-core 之外没有外部依赖。 这 55 行 class 保留完整，目的是让 imports、graph 构造、执行和六类断言仍是一份 可复制测试，而不是彼此脱节的片段。

import com.leanowtech.bloge.core.dsl.GraphBuilder;
import com.leanowtech.bloge.core.engine.GraphEngine;
import com.leanowtech.bloge.core.engine.GraphResult;
import com.leanowtech.bloge.core.model.NodeStatus;
import com.leanowtech.bloge.core.operator.Operator;
import com.leanowtech.bloge.core.spi.DefaultOperatorRegistry;
import org.junit.jupiter.api.Test;

import static org.junit.jupiter.api.Assertions.*;

class OrderGraphTest {

    @Test
    void orderGraph_completesSuccessfully() {
        Operator<Void, String>   validate = (in, ctx) -> "valid";
        Operator<String, String> price    = (in, ctx) -> "$42.00";
        Operator<String, String> confirm  = (in, ctx) -> "ORD-001";

        var gb = new GraphBuilder("order");
        var graph = gb
                .node("validate", validate)
                .node("price", price).dependsOn("validate")
                    .input((results, ctx) -> results.get("validate", String.class))
                .node("confirm", confirm).dependsOn("price")
                    .input((results, ctx) -> results.get("price", String.class))
                .build();

        var engine = GraphEngine.builder()
                .registry(new DefaultOperatorRegistry())
                .build();

        GraphResult result = engine.executeWithOperators(graph, null, gb.operators());

        // 1. 整体成功
        result.requireSuccess();                        // 如果有错误则抛异常

        // 2. 类型化输出提取
        assertEquals("ORD-001", result.getOutput("confirm", String.class));

        // 3. 当节点可能不存在时的安全访问
        assertTrue(result.findOutput("confirm", String.class).isPresent());
        assertTrue(result.findOutput("missing", String.class).isEmpty());

        // 4. 逐节点状态
        assertEquals(NodeStatus.COMPLETED, result.getStatus("validate"));
        assertTrue(result.nodeSucceeded("price"));

        // 5. 逐节点耗时（执行后始终非 null）
        assertNotNull(result.nodeTimings().get("validate"));
        assertFalse(result.nodeTimings().get("validate").isZero());

        // 6. 执行标识
        assertNotNull(result.executionId());
    }
}

源码： GraphResultObservabilityTest.java 和 GraphResultSafeAccessTest.java 演示了上述所有访问方法。

---

拆解分析

GraphResult —— 你的首要断言目标

GraphResult 是一个包含九个字段的不可变 record。在测试中最常使用的访问器：

方法	返回类型	使用场景
isSuccess()	boolean	快速通过/失败检查
requireSuccess()	this 或抛出 GraphExecutionException	在测试中快速失败 —— 异常消息列出所有错误
getOutput(nodeId, type)	T	你确定节点已完成
findOutput(nodeId, type)	Optional<T>	节点可能被跳过或失败
getOutputOrDefault(nodeId, type, default)	T	需要一个兜底值
getOutputIfSuccess(nodeId, type)	Optional<T>	只在整个 graph 成功时才关注输出
nodeSucceeded(nodeId)	boolean	检查单个节点，无需提取输出
statusMap()	Map<String, NodeStatus>	遍历所有状态
nodeTimings()	Map<String, Duration>	性能断言
errors()	List<NodeError>	查看具体的失败详情
suspendedNodes()	Map<String, String>	nodeId → suspendKey，表示挂起的节点
compensationResults()	List<CompensationResult>	graph 失败后的补偿结果
nodeSchemas()	Map<String, SchemaDescriptor>	逐节点输出 schema

bloge-test 模块

bloge-test 模块提供了三个专为 graph 测试设计的工具：

MockOperator<I, O> —— 一个记录每次调用的测试替身。

var fetch = MockOperator.<String, String>returning("Alice");
// ... 执行 graph ...
assertEquals(1, fetch.callCount());
assertEquals("test-input", fetch.lastInput());

工厂方法：returning(value)、throwing(exception)、delaying(duration, value)、recording()、of(function)。

源码：MockOperator.java

GraphTestRunner —— 包装 GraphEngine，内部使用一个 ExecutionListener 来捕获 START:、COMPLETE:、FAILED: 和 SKIPPED: 日志条目。

var runner = new GraphTestRunner(Map.of(
    "validate", validate,
    "price",    price
));
runner.execute(graph);

runner.assertNodeExecuted("validate");
runner.assertNodeSkipped("unreachable");
runner.assertExecutionOrder("validate", "price");

List<String> log = runner.executionLog();
// ["START:validate", "COMPLETE:validate", "START:price", "COMPLETE:price"]

源码：GraphTestRunner.java

DslTestHelper —— 编译 DSL 字符串并断言解析/编译错误。

var helper = new DslTestHelper(registry);
Graph graph = helper.compile("""
    graph test {
        node a : MyOperator { input { key = ctx.value } }
    }
    """);

ParseException err = helper.expectParseError("graph { bad }");
assertTrue(err.getMessage().contains("Expected"));

源码：DslTestHelper.java

对挂起节点、补偿和 schema 做断言

GraphResult 提供了三个额外字段，为挂起/恢复工作流、saga 风格的补偿和 schema 契约测试开辟了新的断言面。

suspendedNodes() —— 一个 Map<String, String>，其中 key 是 nodeId，value 是该节点交回的 suspend key。用它来验证节点确实正确挂起了，且其 key 与你的恢复逻辑预期一致。

GraphResult result = engine.execute(graph, context);

// "approval" 节点应该已挂起
assertTrue(result.isSuspended());
assertEquals("pending-manager-approval",
             result.suspendedNodes().get("approval"));

// 其他节点不应挂起
assertFalse(result.suspendedNodes().containsKey("validate"));

compensationResults() —— 一个 List<CompensationResult>，在配置了补偿 operator 的 graph 失败后收集。每个条目告诉你补偿是否运行了、是否成功。

GraphResult result = engine.execute(failingGraph, context);
assertFalse(result.isSuccess());

List<CompensationResult> compensations = result.compensationResults();
assertFalse(compensations.isEmpty());

// 验证支付节点已成功补偿
CompensationResult paymentComp = compensations.stream()
        .filter(c -> c.nodeId().equals("processPayment"))
        .findFirst().orElseThrow();
assertTrue(paymentComp.isSuccess());
assertEquals("RefundOperator", paymentComp.operatorRef());

// 验证失败的补偿被暴露出来，而不是被吞掉
CompensationResult inventoryComp = compensations.stream()
        .filter(c -> c.nodeId().equals("reserveInventory"))
        .findFirst().orElseThrow();
assertFalse(inventoryComp.isSuccess());
assertNotNull(inventoryComp.error());

nodeSchemas() —— 一个 Map<String, SchemaDescriptor>，包含逐节点的输出 schema。用它做 schema 契约测试，在不运行完整 graph 的情况下验证每个节点的输出结构。

GraphResult result = engine.execute(graph, context);

// 验证 "fetchUser" 节点暴露了预期的 schema
SchemaDescriptor userSchema = result.nodeSchemas().get("fetchUser");
assertNotNull(userSchema);
assertInstanceOf(StructuredSchema.class, userSchema);

StructuredSchema structured = (StructuredSchema) userSchema;
assertTrue(structured.fieldNames().containsAll(
        List.of("id", "name", "email", "vipLevel")));

这三个字段在未填充时默认为空集合，因此已有测试无需修改即可继续通过。

测试时间相关的工作流

时间是最容易让测试套件变慢、变脆，甚至两者兼而有之的因素之一。

BLOGE 的测试模块专门提供了确定性时间控制：

• ManualTimeSource 只会在测试显式推进时前进
• DeterministicTimerService 会在时间推进时同步触发已调度的定时器
• TestGraphEngine 把这两者接成一个现成可用的 GraphEngine

var registry = new DefaultOperatorRegistry();
var testEngine = TestGraphEngine.create(registry);

var op = MockOperator.<Object, String>delaying(Duration.ofMinutes(10), "done");
var builder = new GraphBuilder("delay-test");
Graph graph = builder.node("slow", op).build();

var resultRef = new AtomicReference<GraphResult>();
var done = new CountDownLatch(1);

Thread.startVirtualThread(() -> {
    resultRef.set(testEngine.executeWithOperators(graph, new GraphContext(), builder.operators()));
    done.countDown();
});

testEngine.awaitPendingSleepers(1);      // 先等延迟真正注册完成
testEngine.advanceTime(Duration.ofMinutes(10));

assertTrue(done.await(10, TimeUnit.SECONDS));
assertEquals("done", resultRef.get().getOutput("slow", String.class));

这里最关键的细节是：只有在 sleeper 或 timer 已经注册完成之后，才推进逻辑时间。 这就是 awaitPendingSleepers(...) 存在的意义。

ManualTimeSource.advance(...) 还会同步触发 timer 回调，所以一次时间推进就可能同时释放 sleeper、触发持久 timer，并让整张 graph 在没有任何墙钟等待的情况下继续跑完。

当前最适合用确定性时间验证的场景

这套测试夹具尤其适合下面这些时间相关行为：

• 延迟 operator 和 retry backoff
• 普通 graph 中的 loop / wait timer
• durable lease 过期与定时恢复基础设施
• 执行器显式接入 timer service 时，状态机里的 timer 驱动路径

但也有一个值得明确写出来的现实约束：当前 session 空闲超时路径仍然使用它自己的调度运行机制，而不是直接挂在 ManualTimeSource 上。因此，确定性单元测试目前最适合 graph 与状态机级 timer；session 空闲超时依然应该测试，但更适合用聚焦 smoke test 和策略级断言，而不是假设它已经拥有完整的逻辑时钟控制。

---

限定租约围栏测试范围 —— @WithExecutionLeaseFencing

当两个引擎共享一个持久化 store 时，同一次执行的执行租约应该只被其中一个 持有。如果租约过期、第二个引擎接手了，原来的引擎就不能再被允许提交陈旧的 checkpoint。运行时通过给每次 store 写入加 fencing token 来强制这一点， 但测试需要一种方式来证明这套强制是正确接上的。

@WithExecutionLeaseFencing 要与 bloge-test 的 ExecutionLeaseFencingExtension 配合使用。它只负责一件聚焦的事：

1. 在当前测试范围内临时修改默认 fencing policy；
2. 可选择在缺少 ambient lease context 时启用严格拒绝；
3. 测试结束后恢复此前的默认值。

它不会创建 engine、注入 store、让 lease 过期，也不会自动断言每次写入。 这些准备与断言仍由测试本身负责。

@ExtendWith(ExecutionLeaseFencingExtension.class)
@WithExecutionLeaseFencing
class CheckoutDurabilityTest {

    @Test
    @WithExecutionLeaseFencing(enabled = true, strictMode = true)
    void missingLeaseContextRejectsDurableWrite() {
        InMemoryExecutionStore store = preparedStoreWithExecution("checkout-1");

        assertThrows(StaleFencingEpochException.class, () ->
            store.updateStatus("checkout-1", ExecutionStatus.COMPLETED, 1));
    }
}

这个精简示例里的 helper 只创建 execution record。真正的 takeover 测试还要： 第一次 claim、release、由第二个 owner 再 claim，然后在第一次 claim 的 ExecutionLeaseContext 中尝试旧写入并断言被拒绝。证明旧 owner 被围栏挡住的 是这段时序，而不是 annotation 本身。

---

快照测试

只断言某一个最终输出值，你会丢掉整次执行的形状。快照测试把每一次节点 生命周期事件都捕获成一个稳定的 GraphExecutionSnapshot，然后整体断言。这 是写"这张图还和以前一样跑"这种回归测试的正确工具。

SnapshotCapturingListener listener = new SnapshotCapturingListener();

GraphEngine engine = GraphEngine.builder()
    .registry(registry)
    .listeners(List.of(listener))
    .build();

GraphResult result = engine.execute(checkoutGraph, ctx);

GraphExecutionSnapshot snapshot = GraphExecutionSnapshot.capture(result, listener);

GraphSnapshotAssert.assertMatchesBaseline(
    snapshot,
    "checkout-happy-path.snapshot.json"
);

GraphSnapshotAssert 用 JSON baseline 比较已捕获的 snapshot：

• assertMatchesBaseline(...) 对 completed/skipped 节点使用集合语义，避免合法并行完成顺序造成假失败。
• assertMatchesBaselineStrict(...) 还要求 completed 节点顺序一致。
• assertEquals(...) 与 assertEqualsStrict(...) 比较两个内存 snapshot。
• updateBaseline(...) 写出可审阅 baseline；它应由人明确触发，而不是测试自动刷新。

只在整段可观察执行形状确实是稳定合同时使用 snapshot。对窄规则和边界用例， 单点断言通常更容易说明到底哪里错了。

---

常见陷阱

❌ 只断言最终输出

GraphResult result = runner.execute(graph);
assertEquals("ORD-001", result.getOutput("confirm", String.class));

这个断言只能证明一件事：终点节点返回了你期望的值。它不能证明 graph 确实干净地成功了、上游节点没有被跳过，或者某条你关心的失败没有被韧性策略掩盖。

更稳妥的模式是分层断言：

GraphResult result = runner.execute(graph);
result.requireSuccess();
assertTrue(result.nodeSucceeded("validate"));
assertTrue(result.nodeSucceeded("price"));
assertEquals("ORD-001", result.getOutput("confirm", String.class));

先断言整体成功，再检查这个场景真正重要的少数节点事实，最后再断言最终输出。

---

引导式重写

把第一个示例中的三节点 graph 按下面三步加强测试：

1. 用 MockOperator.returning("$42.00") 替换真实的 price operator，并断言它只被调用一次。
2. 使用 GraphTestRunner 执行 graph，并断言顺序为 validate → price → confirm。
3. 故意破坏 DSL —— 删除一个节点的 operator 类型 —— 然后用 DslTestHelper.expectCompileError(...) 断言这个失败，而不是等到运行时才发现。

完成后，你应该拥有针对同一张小 graph 的三层覆盖：编译、执行和 operator 行为。

---

思维检查

1. 在失败时提供清晰节点错误摘要、并让测试立即失败的最快方法是什么？

   （对 GraphResult 调用 requireSuccess()。）

2. 什么情况下应该使用 findOutput(nodeId, type) 而不是 getOutput(nodeId, type)？

   （当该节点在当前场景下可能被合法跳过、失败或不存在时。）

3. MockOperator 相比手写 lambda，更擅长解决什么问题？

   （它会记录调用信息，并提供可复用的返回 / 抛错 / 延迟行为，减少测试样板代码。）

4. 哪类问题最适合由 DslTestHelper 捕获？

   （.bloge 源中的 parse / compile 问题 —— 比如缺失 operator、无效绑定、schema 错误或语法错误。）

5. 为什么 GraphTestRunner 比"只看最终输出"更有用？

   （因为它让你断言执行顺序、跳过行为和逐节点结果，而不仅是终点值。）

---

实验

目标： 在你信任一张小型订单 graph 之前，先为它建立测试护栏。

1. 用 GraphBuilder 构建一张 validate → price → confirm 的 graph。
2. 三个 operator 都使用 MockOperator。
3. 使用 GraphTestRunner 执行 graph，并断言：
   • 三个节点都执行了；
   • 没有节点被跳过；
   • 执行顺序是 validate、price、confirm。
4. 再次执行同一张 graph，并对返回的 GraphResult 断言：
   • requireSuccess() 通过；
   • confirm 返回预期订单号；
   • price 具有非空耗时。
5. 写一个会失败的 DSL 片段，并断言 DslTestHelper 报告了你预期的编译错误。

进阶： 给 price 加上 fallback，强制主逻辑抛异常，并断言 graph 仍然是按你想要的原因成功的。

---

桥接：从 graph 断言到业务验证

GraphResult、GraphTestRunner 和 DslTestHelper 能说明 graph 是否按测试预期执行，却不会绑定业务负责人批准的 Scenario、用 Policy 保护必需 Cases、用 Fixture 控制声明的 effect，也不会发布 source-bound evidence。当问题从“这张 graph 是否运行”变为“这个业务故事是否仍满足批准的合同”时，继续阅读第 23 章——业务正确性不等于测试通过。

---

实验验收卡

• 预期与观察： 绿色 graph test 只支持它声明的 graph contract。
• 失败与恢复： 删除 node status 断言制造假绿；恢复 output、status 和 evidence 断言。
• 证明边界： 证明受控运行事实，不证明外部系统或业务答案。
• 练习合同： 一个绿色测试；只删一个断言；交付前后 claim 卡；每条结论能指回断言即停止。

回顾

• GraphResult 是断言工作流行为的第一入口。它还提供了 suspendedNodes()、compensationResults() 和 nodeSchemas()，分别用于 挂起/恢复、saga 补偿和 schema 契约断言。
• MockOperator 通过替换无关 I/O，让测试保持聚焦。
• GraphTestRunner 提供 graph 级断言：顺序、执行与跳过行为。
• DslTestHelper 让你以编译器的真实视角测试 .bloge 源文件。
• TestGraphEngine、ManualTimeSource 和 DeterministicTimerService 让时间相关测试既快速又可重复。
• 强健的 BLOGE 测试不是堆一个巨大的集成测试，而是分层建立安全网。

---

下一步

在 第 19 章 —— 生产环境中的可观测性 中，你将从“验证声明的 graph 契约”切换到“理解 graph 在真实运行中正在发生什么”。

---

参考链接

• bloge-test README —— 测试辅助总览
• GraphResult —— 不可变执行结果 record
• MockOperator —— 可复用测试替身
• GraphTestRunner —— graph 执行夹具与断言
• DslTestHelper —— DSL parse / compile 断言
• GraphResultSafeAccessTest.java —— 安全访问示例
• 核心架构 —— 与 graph 执行断言相关的引擎内部机制

Coding Agent: Open the versioned task guide.