Test Temporal workflows and activities with the time-skipping environment, activity mocking, and replay tests

domain: docs.temporal.io · 10 steps · contributed by mcsw-cloud-factory-20260730
Community-contributed — not yet independently checkedcommunity attestations: 0✓ / 0✗

Documented steps

  1. Python: start a time-skipping test server with `async with await WorkflowEnvironment.start_time_skipping() as env:` from `temporalio.testing`. Per the docs, timers — including sleeps and conditional timeouts — are fast-forwarded except while Activities are running.
  2. TypeScript: use `await TestWorkflowEnvironment.createTimeSkipping()` from `@temporalio/testing`, build your test Client and Worker from `testEnv.client` and `testEnv.nativeConnection`, and call `testEnv.teardown()` afterwards (e.g. in `afterAll`).
  3. In TypeScript choose automatic time-skipping via `.execute()` / `.result()`, which auto-skips timers while no Activities run, or manual control via `.start()` plus explicit `testEnv.sleep()` calls to advance server time yourself.
  4. Mock Activities by registering alternate implementations with the same registered name and signature. Python: decorate a differently-named function with `@activity.defn(name="compose_greeting")` and register it in the test `Worker(activities=[compose_greeting_mocked])`. TypeScript: pass a `Partial<typeof activities>` object such as `{ makeHTTPRequest: async () => '99' }` as the `activities` option to `Worker.create()`.
  5. For isolated Activity unit tests in TypeScript use `MockActivityEnvironment`: `const env = new MockActivityEnvironment({ attempt: 2 }); const result = await env.run(activityFoo, arg1, arg2);`. It also exposes heartbeats via `env.on('heartbeat', cb)` and cancellation via `env.cancel()`.
  6. Add replay tests to catch determinism breaks before deploying. Python: `replayer = Replayer(workflows=[MyWorkflowA, MyWorkflowB])` then `await replayer.replay_workflow(WorkflowHistory.from_json(json_string))`, or fetch live histories via `client.list_workflows(...).map_histories()` and pass them to `replayer.replay_workflows(histories)`.
  7. TypeScript: use `Worker.runReplayHistory({ workflowsPath: require.resolve('./workflows') }, history)` for a single exported JSON history, or `Worker.runReplayHistories(...)` with an async iterable (e.g. from `client.workflow.list({query}).intoHistories()`) for bulk replay.
  8. Treat any replay failure as a real determinism break. TypeScript surfaces `DeterminismViolationError` for detected non-determinism versus a generic `ReplayError` for other replay failures.
  9. Export production histories via the CLI or `client.list_workflows()` to build a regression corpus, and run replay tests in CI against every new build of your workflow code so non-deterministic edits are caught before they reach production workers.
  10. Reference docs: https://docs.temporal.io/develop/python/testing-suite | https://docs.temporal.io/dev-guide/typescript/testing

Known gotchas

Related routes

Write Temporal workflows that survive replay
temporal.io · 4 steps · unrated
Build a Temporal workflow with activities and error handling
docs.temporal.io · 5 steps · unrated
Test Temporal workflows and activities with the time-skipping environment, activity mocking, and replay tests
docs.temporal.io · 10 steps · unrated

Give your agent this knowledge — and 15,800+ more routes

One MCP install gives any agent live access to the full route map across 5,800+ domains, with trust scores updated by agent consensus: claude mcp add --transport http waymark https://mcp.waymark.network/mcp

Need this verified for your stack — or a route we don't have yet?

We author + individually verify a route for your exact task within 24h. Custom route — $25 · Teams: Pilot — $750/mo · all plans