Testing¶

testing ¶

In-process test harness for workflow authoring.

:class:WorkflowEnvironment drives a workflow to completion in a single Python process, without a running server or worker. It reuses the same :func:durable_workflow.workflow.replay machinery the worker uses, but resolves yielded commands against user-registered activity mocks and auto-fires timers / side-effects / search-attribute upserts so tests do not need a real clock or Redis.

Typical use::

def test_my_workflow():
    env = WorkflowEnvironment()
    env.register_activity_result("charge_card", {"id": "ch_1"})
    env.register_activity_result("send_receipt", None)
    result = env.execute_workflow(OrderWorkflow, "order-1", {"amount": 42})
    assert result == {"status": "complete", "charge_id": "ch_1"}

For regression-testing workflow code against production histories, use :func:replay_history — it hands the real durable history straight to the worker's replayer and surfaces any non-determinism as a raised exception.

WorkflowEnvironment ¶

WorkflowEnvironment(*, iteration_limit=1000)

Drives a workflow to completion against user-registered activity mocks.

register_activity_result ¶

register_activity_result(name, result)

Canned response: every call to name returns result.

register_activity ¶

register_activity(name, fn)

Callable mock: fn(*arguments) is invoked for each scheduled call.

Use this when the test needs the mock to vary with arguments (e.g. look up by order id) or to capture invocations.

register_child_workflow_result ¶

register_child_workflow_result(workflow_type, result)

Canned response for child workflow completions.

signal ¶

signal(name, args=None)

Queue a signal to be delivered before the next iteration.

Signals are drained in the order they were queued and injected into the workflow history as SignalReceived events; the replayer then dispatches each to its registered @workflow.signal handler.

execute_workflow ¶

execute_workflow(workflow_cls, *args, run_id='test-run')

Drive workflow_cls to a terminal state and return its result.

Raises :class:~durable_workflow.errors.WorkflowFailed if the workflow ended in the failed state. Activities that do not have a registered mock raise :class:KeyError so tests fail loudly on missing fixtures.

replay_history ¶

replay_history(workflow_cls, history_events, start_input=None, *, run_id='', payload_codec=None)

Replay a production history against current workflow code.

Hands the durable history directly to the worker's replayer. Raises any exception the workflow would raise during replay — for example a non-determinism failure when run yields a different command sequence from the one recorded in history.

This is the supported way to regression-test a workflow change against real production traffic: dump the history from Client.get_history, save the JSON, and replay it on every PR.

replay_history_file ¶

replay_history_file(workflow_cls, path, start_input=None, *, run_id='', payload_codec=None)

Convenience wrapper: load a JSON history file and replay it.

Accepts either a list of events at the top level or a dict with an events key (matching the shape of Client.get_history).