Skip to main content
Super Durable
Menu

Test Your Flows

View as Markdown

Test a Flow against a real Dex Server. A mock cannot show that a Flow survives a Worker restart, that its inputs encode and decode, or that a wait wakes it. The SDKs have no in-memory server for this. Unit-test only plain helper functions, such as a function that builds a request body.

dexcli dev starts a server in a few seconds, so each test run can start its own.

Start a server for the tests

Start dexcli dev on ports and a data directory that only the tests use, so the tests never touch your long-running local server:

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 &

Wait until its output says Dex development environment is ready, then run the tests with DEX_FLOW_SERVICE_ADDRESS=127.0.0.1:19802. The processCrashSafe setting skips fsync, which makes a test server faster. It survives a killed process but not a host crash, so use it only for tests. See Durability.

The examples in the repository run their tests this way in CI. Each language folder has a run-integ-tests.sh script that starts dexcli dev, waits for it, and runs the tests.

Write a test

The Quick Start's restart test shows the pattern. It starts its own Worker, starts a Flow with a unique ID, stops that Worker after the first Step, starts a second Worker, and checks that only the second Step ran on it:

quick-startPython
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]

Example: examples/python/quick-start

Follow the same rules in your own tests:

  • Give every test a unique Flow ID. Tests can then run at the same time on one server, and a Flow left over from an earlier run does not interfere.
  • Run the Worker inside the test. Build it from the same Registry constructor that your application uses, so the test checks the real registration.
  • Wait with Client calls, not sleeps. wait_for_step_completion waits for one Step, and wait_for_flow waits for the Flow to close.
  • Use fakes that record every call. The test's mock API keeps each call, so the test can see a repeat and count it.
  • Give a crash test its own Worker. Workers that register the same Flow type share its tasks. A test that stops a Worker must not stop one that other tests use.

Skip a Timer

Skip Timers in tests, such as the Quick Start's 60-second one. Give the Timer a condition_id in the Flow, then skip it from the test with skip_timer. The call fails until the Step waits on the Timer, so the test retries it until a deadline:

quick-startPython
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)

Example: examples/python/quick-start

Next