---
title: "测试你的 Flow"
description: "用 dexcli dev 启动的真实 Dex Server 测试 Flow。测试运行自己的 Worker，给每个 Flow 一个唯一的 ID，按 ID 跳过 Timer，并用 Client 调用等待，而不是 sleep。"
canonical_url: "https://docs.superdurable.io/zh-Hans/components/testing/"
source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/components/testing.mdx"
locale: "zh-Hans"
language: "python"
---

> 供 AI Agent 使用的全部页面索引：https://docs.superdurable.io/zh-Hans/llms.txt

# 测试你的 Flow

本页代码为 Python。其他语言：[Go](https://docs.superdurable.io/zh-Hans/components/testing.go.md) · [Java](https://docs.superdurable.io/zh-Hans/components/testing.java.md) · [TypeScript](https://docs.superdurable.io/zh-Hans/components/testing.typescript.md) · [Rust](https://docs.superdurable.io/zh-Hans/components/testing.rust.md)

用真实的 Dex Server 测试 Flow。只有真实的 server 才能证明 Flow 能在 Worker 重启后继续运行、它的输入能正确编码和解码、等待能唤醒它，mock 做不到这些。SDK 也没有为此提供内存中的 server。只对普通的辅助函数做单元测试，例如构造请求体的函数。

**dexcli dev** 几秒内就能启动一个 server，所以每次测试运行都可以启动自己的 server。

> **用编程 Agent 来做**
>
> 先安装一次 Dex Skills：
>
> Claude Code:
>
> ```bash
> claude plugin marketplace add superdurable/dex-skills
> claude plugin install superdurable-dex@superdurable
> ```
>
> Codex:
>
> ```bash
> codex plugin marketplace add superdurable/dex-skills
> codex plugin add superdurable-dex@superdurable
> ```
>
> 然后粘贴这段提示词：
>
> ```text
> 为 <FlowType> Flow 加一个集成测试。
>
> 1. 阅读 https://docs.superdurable.io/zh-Hans/components/testing.md，把它当作唯一依据。
> 2. 在 dexcli dev 启动的 server 上运行测试，使用测试自己的 Worker，每个 Flow 使用唯一的 Flow ID。
> 3. 覆盖正常路径和一条失败路径。按 ID 跳过 Timer，用轮询等待，不要用 sleep。
>
> 规则：不要 mock Dex。不要编造 API。
> ```

<a id="start-a-server-for-the-tests"></a>

## 为测试启动 server

在只给测试使用的端口和数据目录上启动 **dexcli dev**，这样测试永远碰不到你长期运行的本地 server：

```bash
printf 'dev:\n  durabilityMode: processCrashSafe\n' > zd-overrides.yaml
dexcli dev -bind-address 127.0.0.1 -dex-port 19802 -web-port 19902 -open=false \
  -zd-data-dir "$(mktemp -d)" -zd-config zd-overrides.yaml &
```

等它的输出显示 **Dex development environment is ready**，然后用 `DEX_FLOW_SERVICE_ADDRESS=127.0.0.1:19802` 运行测试。`processCrashSafe` 设置跳过 fsync，让测试 server 更快。它能扛住进程被杀，但扛不住主机崩溃，所以只在测试中使用。见 [Durability](https://docs.superdurable.io/zh-Hans/reference/server/configuration.md#single-node-local-durability)（英文）。

仓库里的示例在 CI 中就是这样运行测试的。每个语言目录都有一个 `run-integ-tests.sh` 脚本，它启动 **dexcli dev**，等待它就绪，然后运行测试。

<a id="write-a-test"></a>

## 编写测试

快速开始的重启测试展示了这个模式。它启动自己的 Worker，用唯一的 ID 启动一个 Flow，在第一个 Step 之后停止这个 Worker，再启动第二个 Worker，然后检查在第二个 Worker 上只运行了第二个 Step：

**Python**

```python
def test_restarted_worker_ships_without_charging_again(tmp_path: Path) -> None:
    order_id = f"order-{uuid.uuid4().hex}"
    first_shop = ShopApi()
    second_shop = ShopApi()
    client_flow = OrderFlow(ShopApi())
    with connected_client(client_flow, tmp_path / "client") as client:
        with running_worker(OrderFlow(first_shop), tmp_path / "worker-1"):
            client.start_flow(client_flow, order_id, order_id)
            client.wait_for_step_completion(
                order_id,
                StepExecutionId("Charge"),
                WaitForStepCompletionOptions(request_timeout=WAIT_TIMEOUT),
            )
        assert _calls_for(order_id, first_shop.charged) == [order_id]
        assert _calls_for(order_id, first_shop.shipped) == []

        with running_worker(OrderFlow(second_shop), tmp_path / "worker-2"):
            _skip_ship_delay(client, order_id)
            result = client.wait_for_flow(order_id, WAIT_TIMEOUT)

    assert result.status == FlowStatus.COMPLETED
    assert _calls_for(order_id, second_shop.charged) == []
    assert _calls_for(order_id, second_shop.shipped) == [order_id]
```

例子: [examples/python/quick-start](https://github.com/superdurable/dex/tree/main/examples/python/quick-start)

在你自己的测试中遵循同样的规则：

- **给每个测试一个唯一的 Flow ID。** 这样多个测试可以同时在一个 server 上运行，之前运行留下的 Flow 也不会干扰。
- **在测试里运行 Worker。** 用应用里同一个 Registry 构造函数来构建它，这样测试检查的是真实的注册。
- **用 Client 调用等待，而不是 sleep。** `wait_for_step_completion` 等待一个 Step，`wait_for_flow` 等待 Flow 关闭。
- **使用记录每次调用的 fake。** 测试的 mock API 保存每一次调用，所以测试能看到重复的调用并计数。
- **给崩溃测试单独的 Worker。** 注册同一个 Flow type 的 Worker 会共享它的任务。停止 Worker 的测试，只能停止自己的 Worker，而不是其他测试正在用的。

<a id="skip-a-timer"></a>

## 跳过 Timer

在测试中跳过 Timer，例如快速开始里那个 60 秒的 Timer。在 Flow 中给这个 Timer 一个 `condition_id`，然后在测试中用 `skip_timer` 跳过它。在 Step 开始等待这个 Timer 之前，这个调用会失败，所以测试会一直重试，直到超过截止时间：

**Python**

```python
def _skip_ship_delay(client: Client, order_id: str) -> None:
    """Skips the 60-second Timer once Ship waits on it."""
    deadline = time.monotonic() + WAIT_TIMEOUT.total_seconds()
    while True:
        try:
            client.skip_timer(
                order_id,
                StepExecutionId("Ship"),
                TimerId.by_condition_id(SHIP_DELAY_TIMER),
            )
            return
        except DexServiceError:
            if time.monotonic() >= deadline:
                raise
            time.sleep(0.25)
```

例子: [examples/python/quick-start](https://github.com/superdurable/dex/tree/main/examples/python/quick-start)

<a id="next"></a>

## 下一步

- [快速开始](https://docs.superdurable.io/zh-Hans/quick-start.md)：这些测试运行的 Flow。
- [本地运行](https://docs.superdurable.io/zh-Hans/components/run-locally.md)：**dexcli dev** 的选项。
- [Step 的保证](https://docs.superdurable.io/zh-Hans/concepts/step.md#guarantees)：测试应该检查的行为，例如 Step 再运行一次。