Testing¶
Test failures, retries and compensation¶
This failure/retry behavior and callable child mocks require Python SDK 2.10.0 or later. Earlier packages let activity mock exceptions escape and do not provide these child callables or retry observations.
Synchronous mock exceptions become terminal history events. The production
replayer then raises ActivityFailed or ChildWorkflowFailed at the authored
yield, so the workflow's handlers and compensation code run normally:
from durable_workflow import workflow
from durable_workflow.errors import ActivityFailed
from durable_workflow.testing import WorkflowEnvironment
from durable_workflow.workflow import ActivityRetryPolicy
@workflow.defn(name="checkout-test")
class Checkout:
def run(self, ctx):
yield ctx.schedule_activity("reserve", [])
try:
return (yield ctx.schedule_activity(
"charge", [], retry_policy=ActivityRetryPolicy(max_attempts=3)
))
except ActivityFailed as error:
yield ctx.schedule_activity("release", [])
return {"compensated": True, "failure_type": error.exception_type}
def decline():
raise ValueError("payment unavailable")
env = WorkflowEnvironment()
env.register_activity_result("reserve", None)
env.register_activity("charge", decline)
env.register_activity_result("release", None)
assert env.execute_workflow(Checkout) == {
"compensated": True, "failure_type": "ValueError"
}
assert env.mock_calls[1].attempt_count == 3
assert env.mock_calls[1].status == "failed"
Each logical call invokes its mock once per attempt. Later workflow replays
reuse its final history event and do not invoke the old mock again. The last
attempt's error reaches the workflow when its retry budget is exhausted.
No configured policy means one attempt, matching the external-worker default.
Explicit ActivityRetryPolicy, ChildWorkflowRetryPolicy and mapping policies
use their positive integer max_attempts. Non-retryable type lists match
either a short error name or its qualified Python class name.
An activity mock raising NonRetryableError stops immediately with
ActivityFailed.non_retryable=True. Its reported type is the outer
NonRetryableError; a supplied cause does not replace that type. Child
exceptions follow the existing workflow failure mapper. ChildWorkflowFailed
exposes the child class/message; the published child-history contract does not
carry an activity-style non_retryable field. Use the child retry policy's
type exclusions to stop retries for a named child error.
Register a child callable with
env.register_child_workflow("credit-check", check_credit). It receives the
child arguments as positional arguments and can return a value or raise an
ordinary exception. register_child_workflow_result() still registers a
canned success. A later child result or callable registration replaces the
previous one. Missing mocks remain actionable KeyError fixture errors.
env.mock_calls contains MockCallRecord objects in logical execution order,
with kind, name, run_number, attempt_count, status and the last
observed failure. These observations span a continue-as-new chain and reset
on the next execute_workflow() call. They are test records, separate from
durable history. A later successful attempt keeps the last observed failure
available while setting the call's status to completed.
WorkflowEnvironment(mock_attempt_limit=100) caps each logical call at 100
actual attempts by default. If more retries remain, the harness raises
RuntimeError with the last callback error as its cause and records
attempt_limit_exceeded. It does not deliver a fabricated exhausted failure
to the workflow. Raise the cap deliberately for a larger bounded test.
Retry backoff does not sleep or advance a virtual clock. Mocks must be synchronous callables. They do not test durable queue scheduling, actual retry delays, timeout/lease enforcement, worker processes, cancellation or backend recovery; qualify those behaviors with a real Server and workers.
Mock a Nexus service operation¶
These Nexus mock registration methods require Python SDK 2.10.0 or later.
Register a result or a synchronous callable by endpoint, service and operation name. The mock runs once for each authored call, even when later workflow tasks replay that call. A later registration for the same three names replaces the previous mock.
from durable_workflow import workflow
from durable_workflow.errors import NexusOperationFailed
from durable_workflow.testing import WorkflowEnvironment
@workflow.defn(name="create-invoice")
class InvoiceWorkflow:
def run(self, ctx, customer):
try:
operation = yield ctx.call_nexus_service(
"billing", "invoices", "create", [customer]
)
return operation.result
except NexusOperationFailed as error:
return {"refused": error.service_error_type}
env = WorkflowEnvironment()
env.register_nexus_operation_result(
"billing", "invoices", "create", {"id": "invoice-1"}
)
assert env.execute_workflow(InvoiceWorkflow, "customer-1") == {"id": "invoice-1"}
def decline(customer):
raise NexusOperationFailed(
"Payment declined", service_error_type="PaymentDeclined", reason="insufficient_funds"
)
env.register_nexus_operation("billing", "invoices", "create", decline)
assert env.execute_workflow(InvoiceWorkflow, "customer-1") == {"refused": "PaymentDeclined"}
Callable mocks receive the operation arguments as positional arguments. An
ordinary exception becomes NexusOperationFailed, with the original exception
class in service_error_type and its message preserved. Raising
NexusOperationFailed directly also preserves its reason and typed failure
metadata. Workflow code can catch the failure and yield compensation commands.
An unmocked operation raises an actionable KeyError from the harness.
call_nexus_service() normally waits for completion and returns a
NexusOperationResult whose result contains the mocked value.
start_nexus_operation() normally waits for acceptance: it returns an accepted,
started operation with a synthetic service-call ID and no completed result.
Explicit acceptance waits and asynchronous modes follow the same distinction.
These mocks run in-process without service endpoints or network requests. They do not test namespace authorization, delayed remote completion, service retries, timeouts or cancellation. Use a real Server and service workers for those checks.
In-process test harness for workflow authoring.
WorkflowEnvironment drives a workflow to completion in a single
Python process, without a running server or worker. It reuses the same
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"}
When a workflow yields ContinueAsNew,
the harness records a WorkflowContinuedAsNew event on the completing
run, resets history, and starts a new run with the command's arguments.
WorkflowEnvironment.runs exposes the per-run record (input,
workflow type, history events, and terminal command) so tests can assert
on the whole chain. The chain length is bounded by
continue_as_new_limit; exceeding it raises RuntimeError.
For regression-testing workflow code against production histories, use
replay_history — it hands the real durable history straight to
the worker's replayer and surfaces any non-determinism as a raised
exception.
dataclass
¶
One logical activity/child mock call in the latest execution chain.
attempt_count includes each actual callback invocation. failure
retains the last observed failure even if a later attempt succeeds.
These are harness observations, not durable runtime history events.
dataclass
¶
One run within an execute_workflow chain.
A single-run workflow produces one record with terminal of
type CompleteWorkflow or
FailWorkflow. A workflow that
yields ContinueAsNew produces
one record per link; the terminal of each non-final link is the
ContinueAsNew command itself and the final link terminates as
usual.
Drives a workflow to completion against user-registered activity mocks.
:param iteration_limit: upper bound on per-run replay iterations
(guard against workflows that never yield a terminal command).
:param continue_as_new_limit: upper bound on the number of
continue-as-new links in one execute_workflow chain. The
first run counts as link 1; a workflow that continues five
times runs six links. Exceeding the limit raises
RuntimeError.
:param mock_attempt_limit: positive per-call cap on activity and child
mock attempts. A larger scheduled budget fails loudly at this cap;
it is not reported to workflow code as exhausted durable retries.
property
¶
Per-run records produced by the most recent execute_workflow call.
Empty before the first execute_workflow call. Reset at the
start of each call.
property
¶
Logical mock calls, with actual attempt counts, in execution order.
Reset by execute_workflow. The returned list spans all runs in a
continue-as-new chain; run_number identifies the owning run.
Canned response: every call to name returns result.
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.
Canned response for child workflow completions.
Synchronous child-body mock, called once per scheduled attempt.
Exceptions become ChildWorkflowFailed through production replay.
A later child result/callable registration replaces the previous mock.
Mock a service response for the named operation.
Completion waits receive result through NexusOperationResult.result.
Acceptance waits receive a started operation without a completed result.
A later registration for the same three names replaces this mock.
Call synchronous fn(*arguments) once per mocked operation.
Exceptions become service failures delivered as NexusOperationFailed
at the workflow yield. Raise that type explicitly to supply a service
error type, reason or other failure metadata. A later registration for
the same three names replaces this mock.
Make workflow_cls resolvable by name for continue-as-new.
continue_as_new(workflow_type="other-name") looks up the
next workflow class by its @workflow.defn name. The class
passed to execute_workflow is registered automatically;
call this method for additional workflow classes the chain may
continue into.
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 dispatches each to its registered @workflow.signal
handler.
Pass run to target a specific link in a continue-as-new
chain (run=1 is the first run, run=2 the run that
starts after the first ContinueAsNew, and so on). Without
run the signal is consumed at the current front of the
chain — on the first run before the first iteration, and on
later iterations in whatever link happens to be running.
Drive workflow_cls to a terminal state and return its result.
Raises WorkflowFailed if the
workflow ended in the failed state. Activities that do not
have a registered mock raise KeyError so tests fail
loudly on missing fixtures.
If the workflow yields
ContinueAsNew, the harness
starts a new run with the new input and, if the command named a
workflow_type, the matching registered workflow class. The
chain is bounded by continue_as_new_limit and returns the
terminal result of the final run.
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.
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).