Testing

The SDK provides two in-memory test environments that do not require a Cadence server:

• TestWorkflowEnvironment executes workflows deterministically, mocks activities, and advances virtual time.
• TestActivityEnvironment executes one activity with simulated metadata, heartbeats, cancellation, and timeouts.

Samples

Test samples:

Sample	Description	Code
Buffer overlap test	Exercises schedule overlap policies against a running server	test_buffer_overlap.py
Queue full test	Exercises schedule queue limits against a running server	test_queue_full.py
Activity environment tests	Covers execution, heartbeats, cancellation, methods, and timeouts without a server	test_activity_environment.py

Basic test

import pytest
from cadence import Registry, workflow
from cadence.workflow import execute_activity
from cadence.testing import TestWorkflowEnvironment
from datetime import timedelta

registry = Registry()

@registry.workflow()
class GreetingWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        return await execute_activity(
            "greet",
            str,
            name,
            start_to_close_timeout=timedelta(seconds=5),
        )

@pytest.mark.asyncio
async def test_greeting():
    env = TestWorkflowEnvironment(registry)
    env.on_activity("greet", result="Hello, World!")

    client = env.client
    await client.start_workflow(
        "GreetingWorkflow",
        "World",
        workflow_id="test-greeting",
        task_list="tl",
        execution_start_to_close_timeout=timedelta(minutes=1),
    )

    result = env.get_workflow_result(str, workflow_id="test-greeting")
    assert result == "Hello, World!"

Mocking activities

Mock an activity by name with a fixed return value:

env.on_activity("fetch_order", result={"id": "123", "status": "pending"})

Mock with a function to inspect arguments or return dynamic values:

def fake_fetch(order_id: str) -> dict:
    assert order_id == "123"
    return {"id": order_id, "status": "pending"}

env.on_activity("fetch_order", fn=fake_fetch)

The mock function receives the decoded activity arguments. It can be sync or async.

Checking workflow results

result = env.get_workflow_result(str, workflow_id="my-wf")

If the workflow failed, get_workflow_result re-raises the error. To inspect the error without re-raising:

error = env.get_workflow_error(workflow_id="my-wf")

Check whether a workflow has finished:

assert env.is_workflow_completed(workflow_id="my-wf")

Testing signals

Send a signal from the test after starting the workflow:

@pytest.mark.asyncio
async def test_approval():
    env = TestWorkflowEnvironment(registry)
    client = env.client

    execution = await client.start_workflow(
        "ApprovalWorkflow",
        workflow_id="approval-test",
        task_list="tl",
        execution_start_to_close_timeout=timedelta(minutes=1),
    )

    await client.signal_workflow(execution.workflow_id, "", "approve", True)

    result = env.get_workflow_result(bool, workflow_id="approval-test")
    assert result is True

Testing queries

await client.signal_workflow("my-wf", "", "set_status", "processing")
status = await client.query_workflow("my-wf", "", "get_status", result_type=str)
assert status == "processing"

Testing an activity

Pass an activity definition directly to TestActivityEnvironment.execute_activity:

from cadence import activity
from cadence.testing import TestActivityEnvironment

@activity.defn()
async def greet(name: str) -> str:
    return f"Hello, {name}!"

@pytest.mark.asyncio
async def test_greet_activity():
    with TestActivityEnvironment() as env:
        result = await env.execute_activity(greet, "Cadence")

    assert result == "Hello, Cadence!"

Synchronous activities and activity methods are supported. To execute an activity by name, register it first:

env = TestActivityEnvironment()
env.register_activity(greet)
result = await env.execute_activity("greet", "Cadence")

Heartbeats

Seed heartbeat details to simulate a retry, then inspect heartbeats emitted by the activity:

from cadence import activity

@activity.defn()
async def resumable_activity() -> int:
    previous_page, previous_count = activity.heartbeat_details(str, int)
    assert (previous_page, previous_count) == ("previous-page", 20)
    activity.heartbeat("current-page", 40)
    return 40

env = TestActivityEnvironment()
env.set_heartbeat_details("previous-page", 20)

result = await env.execute_activity(resumable_activity)

assert result == 40
assert env.heartbeat_count == 1
assert env.get_heartbeat_details(str, int) == ["current-page", 40]

Inside the activity, activity.heartbeat_details(str, int) returns the seeded values.

Cancellation and timeout

Call env.cancel() before or during execution to simulate an activity cancellation request. Async activities receive asyncio.CancelledError after they heartbeat; synchronous activities can observe activity.is_cancelled().

Use set_test_timeout to impose a wall-clock limit:

env = TestActivityEnvironment()
env.set_test_timeout(timedelta(milliseconds=100))

with pytest.raises(asyncio.TimeoutError):
    await env.execute_activity(slow_activity)

TestWorkflowEnvironment reference

Method / Property	Description
env.client	Client-compatible object for starting and interacting with workflows
env.on_activity(name, result=...)	Mock an activity with a fixed return value
env.on_activity(name, fn=...)	Mock an activity with a callable
env.get_workflow_result(type, workflow_id="")	Get the decoded result; raises if the workflow failed
env.get_workflow_error(workflow_id="")	Get the error if the workflow failed, or None
env.is_workflow_completed(workflow_id="")	Whether the workflow has finished
env.now()	Current virtual time
env.close()	Shut down the executor

TestWorkflowEnvironment is also a synchronous context manager:

with TestWorkflowEnvironment(registry) as env:
    ...

TestActivityEnvironment reference

Method / Property	Description
env.registry	Registry used to resolve activities by name
env.client	Client returned by activity.client() during async activity execution
env.register_activity(definition)	Register one activity
env.register_activities(instance)	Register decorated activity methods on an object
await env.execute_activity(activity, *args, **kwargs)	Execute a definition or registered activity name
env.set_heartbeat_details(*details)	Seed heartbeat details from a simulated previous attempt
env.heartbeat_count	Number of heartbeats in the last execution
env.get_heartbeat_details(*types)	Decode details from the latest heartbeat
env.cancel()	Request cancellation
env.set_test_timeout(duration)	Set a wall-clock execution timeout
env.close()	Shut down the activity thread pool

TestActivityEnvironment is also a synchronous context manager.