--- title: "Test Your Flows" description: "Test a Flow against a real Dex Server started with dexcli dev. The test runs its own Workers, gives each Flow a unique ID, skips Timers by ID, and waits with Client calls instead of sleeps." canonical_url: "https://docs.superdurable.io/components/testing/" source: "docs/content/components/testing.mdx" locale: "en" language: "python" --- > Index of every page for AI agents: https://docs.superdurable.io/llms.txt # Test Your Flows Code on this page is in Python. Other languages: [Go](https://docs.superdurable.io/components/testing.go.md) · [Java](https://docs.superdurable.io/components/testing.java.md) · [TypeScript](https://docs.superdurable.io/components/testing.typescript.md) · [Rust](https://docs.superdurable.io/components/testing.rust.md) 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. > **Do it with a coding agent** > > Install Dex Skills once: > > 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 > ``` > > Then paste this prompt: > > ```text > Add an integration test for the Flow. > > 1. Read https://docs.superdurable.io/components/testing.md and treat it as the > source of truth. > 2. Run the test against a dexcli dev server, with its own Workers and a unique > Flow ID for each Flow. > 3. Cover the normal path and one failure path. Skip Timers by ID, and wait by > polling instead of sleeping. > > Rules: do not mock Dex. Do not invent APIs. > ``` ## 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: ```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 & ``` 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](https://docs.superdurable.io/reference/server/configuration.md#single-node-local-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: **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] ``` Example: [examples/python/quick-start](https://github.com/superdurable/dex/tree/main/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: **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) ``` Example: [examples/python/quick-start](https://github.com/superdurable/dex/tree/main/examples/python/quick-start) ## Next - [Quick Start](https://docs.superdurable.io/quick-start.md): the Flow that these tests run. - [Run Locally](https://docs.superdurable.io/components/run-locally.md): the options of **dexcli dev**. - [Step guarantees](https://docs.superdurable.io/concepts/step.md#guarantees): the behavior a test should check, such as a Step that runs again.