testing-and-debugging

164 notes in this subject, read out of this brain and free to use. This is page 2 of 3.

Spring Boot TestWorkflowEnvironment bean wiring

When spring.temporal.test-server.enabled is set to true in Spring Boot, you can autowire TestWorkflowEnvironment, WorkflowClient, and ConfigurableApplicationContext beans in your unit tests. The application context must be started in a @BeforeEach method before tests run.

Temporal testing Maven dependency

Add the temporal-testing dependency to Maven pom.xml with groupId io.temporal, artifactId temporal-testing, version 1.33.0, and scope test.

Temporal testing Gradle dependency

Add temporal-testing:1.33.0 to Gradle build.gradle dependencies as testImplementation 'io.temporal:temporal-testing:1.33.0'.

Testing helper method to detect Continue-As-New time

A helper method can check both Workflow.getInfo().isContinueAsNewSuggested() and a test-only maxHistoryLength variable. If maxHistoryLength is greater than 0 and Workflow.getInfo().getHistoryLength() exceeds maxHistoryLength, trigger Continue-As-New. In production, rely on Workflow.getInfo().isContinueAsNewSuggested().

Production environment debugging tools for PHP Workflows

In production environments, PHP Workflows can be debugged using the Web UI and Temporal CLI.

Development environment debugging tools for PHP Workflows

In PHP SDK development environments, you can debug Workflows using normal development tools such as logging and a debugger, as well as the Web UI and Temporal CLI.

Production Worker performance debugging in PHP

Worker performance can be debugged and tuned in production using metrics and the Worker performance guide.

Replay workflow executions from server using WorkflowReplayer

Use \Temporal\Testing\Replay\WorkflowReplayer to replay Workflow Executions. This class can replay Workflows fetched from Temporal Server. If a Workflow is non-deterministic, a NonDeterministicWorkflowException will be thrown. This requires Advanced Visibility to be enabled.

Replay workflow from JSON file

Replay a Workflow from a JSON Event History file using WorkflowReplayer.replayFromJSON() method. The lastEventId parameter is optional and limits the maximum number of replayed Events: $replayer->replayFromJSON( workflowType: 'MyWorkflow', path: 'history.json', lastEventId: 42, // optional );

Replay workflow from history object

Download an Event History using WorkflowClient and replay it from a History object: $history = $this->workflowClient->getWorkflowHistory( execution: $run->getExecution(), )->getHistory(); (new WorkflowReplayer())->replayHistory($history);

Types of automated tests in Temporal PHP

Temporal PHP applications support three types of automated tests: end-to-end tests which run a Temporal Server and Worker with all Workflows and Activities, interacting with them from a Client; integration tests which cover anything between end-to-end and unit testing, such as running Activities with mocked Context, running Workers with mock Activities, or running Workflows with mocked SDK imports; and unit tests which run a piece of Workflow or Activity code with mocked dependencies. Integration tests are generally recommended as the majority of your test suite.

Test server supports skipping time

The test server in Temporal PHP supports skipping time, which allows testing both end-to-end and integration tests with Workers. This enables long-running Workflows that would normally take months or years to complete to finish in seconds during testing.

ActivityMocker for mocking Activities in PHP tests

To mock Activities when testing Workflows in PHP, use the ActivityMocker class from the test framework. An Activity can be tested with a mock Activity environment that provides a way to mock the Activity context, listen to Heartbeats, and cancel the Activity. This allows testing an Activity in isolation without creating a Worker.

ActivityMocker RoadRunner configuration

To mock Activities in PHP tests, add Key-Value storage configuration to tests/.rr.test.yaml with the following: kv: test: driver: memory config: interval: 10

WorkerFactory setup for Activity mocking

To enable Activity mocking in PHP tests, use WorkerFactory from the Temporal\Testing namespace instead of the default worker implementation: use Temporal\Testing\WorkerFactory; $factory = WorkerFactory::create(); $worker = $factory->newWorker(); $worker->registerWorkflowTypes(MyWorkflow::class); $worker->registerActivity(MyActivity::class); $factory->run();

Activity mocking with expectCompletion in PHP

Mock an Activity to return a value using ActivityMocker.expectCompletion() method. For example: $this->activityMocks->expectCompletion('SimpleActivity.doSomething', 'world'); will mock the Activity named SimpleActivity.doSomething to return the string 'world'.

Activity mocking with expectFailure in PHP

Mock an Activity to throw an exception using ActivityMocker.expectFailure() method. For example: $this->activityMocks->expectFailure('SimpleActivity.echo', new \LogicException('something went wrong')); will mock the Activity to fail with the specified exception.

ActivityMocker lifecycle in test cases

Instantiate ActivityMocker in the setUp() method of a test case and call clear() on it in the tearDown() method to reset the mocked Activities between tests.

Set up Temporal test server with bootstrap in PHP

Create tests/bootstrap.php to set up the Temporal test environment: declare(strict_types=1); require __DIR__ . '/../vendor/autoload.php'; use Temporal\Testing\Environment; $environment = Environment::create(); $environment->start(); register_shutdown_function(fn () => $environment->stop());

Conditional test server startup with environment variable

To only run the Temporal test server when needed, wrap the Environment setup in a conditional check for the RUN_TEMPORAL_TEST_SERVER environment variable: if (getenv('RUN_TEMPORAL_TEST_SERVER') !== false) { $environment = Environment::create(); $environment->start('./rr serve -c .rr.silent.yaml --workflow-id tests'); register_shutdown_function(fn() => $environment->stop()); }

PHPUnit bootstrap and Temporal address configuration

Add the bootstrap file and TEMPORAL_ADDRESS environment variable to phpunit.xml: <phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="https://schema.phpunit.de/9.3/phpunit.xsd" bootstrap="tests/bootstrap.php" > <php> <env name="TEMPORAL_ADDRESS" value="127.0.0.1:7233" /> </php> </phpunit>

Time skipping in Temporal test framework

Time is a global property of a TestWorkflowEnvironment instance. Skipping time applies to all currently running tests. If different tests need different time behaviors, run tests in series or with separate test server instances. For example, run all tests with automatic time skipping in parallel, then tests with manual time skipping in series, then tests without time skipping in parallel.

Workflow replay to detect non-determinism

Replay recreates the exact state of a Workflow Execution from its Event History and succeeds only if the Workflow Definition is compatible with the provided history from a deterministic point of view. When testing Workflow Definition changes, replay a representative set of recent open and closed Workflows from each Task Queue as part of CI checks.

Replay workflow from server example

Example of fetching and replaying Workflow Executions from Temporal Server: $executions = $workflowClient->listWorkflowExecutions( "WorkflowType='MyWorkflow' AND TaskQueue='MyTaskQueue'" ); foreach ($executions as $executionInfo) { try { $replayer->replayFromServer( workflowType: $executionInfo->type->name, execution: $executionInfo->execution, ); } catch (\Temporal\Testing\Replay\Exception\ReplayerException $e) { // Handle a replay error. } }

DotNet worker MaxCachedWorkflows 0 for testing

Example of disabling workflow caching in a DotNet worker for plugin testing: using var worker = new TemporalWorker( client, new TemporalWorkerOptions("task-queue") { MaxCachedWorkflows = 0 });

Ruby worker max_cached_workflows 0 for testing

Example of disabling workflow caching in a Ruby worker for plugin testing: worker = Temporalio::Worker.new( client: client, task_queue: 'task-queue', max_cached_workflows: 0 )

Go worker SetStickyWorkflowCacheSize 0 for testing

Example of disabling workflow caching in a Go worker for plugin testing: worker.SetStickyWorkflowCacheSize(0) w := worker.New(c, "task-queue", worker.Options{})

Python worker max_cached_workflows 0 for testing

Example of disabling workflow caching in a Python worker for plugin testing: worker = Worker(client, task_queue="task-queue", max_cached_workflows=0)

Avoid side effects to global variables in plugin tests

It is harder to test against side effects to global variables, so this practice is best avoided entirely in plugin development.

Test for duplicate side effects in plugins by counting events or activity IDs

When testing for duplicate side effects in plugins, it may not be sufficient to simply increment a counter once per effect as activities may be retried. Instead, count ActivityTaskScheduled events of the expected activity type in event history (one per intended call, independent of retries), or accumulate activity IDs in a concurrency-safe set and assert on its size (different scheduled activities get different IDs; retries of the same scheduled activity share one).

Test plugin side effects by disabling Workflow caching

To ensure a plugin does not depend on local side effects, disable Workflow caching (set max_cached_workflows to 0) so that the Workflow replays from the top each time it progresses. This tests whether the plugin properly handles Workflows resuming in different processes and replaying from the beginning.

Test plugin changes with replay testing

When making changes to a plugin after it has already shipped to users, set up replay testing on each important change to ensure you are not causing non-determinism errors for users.

TypeScript worker maxCachedWorkflows 0 for testing

Example of disabling workflow caching in a TypeScript worker for plugin testing: const worker = await Worker.create({ connection, taskQueue: 'task-queue', maxCachedWorkflows: 0, });

Java WorkerFactory workflow cache size 0 for testing

Example of disabling workflow caching in a Java worker for plugin testing: WorkerFactory factory = WorkerFactory.newInstance( client, WorkerFactoryOptions.newBuilder().setWorkflowCacheSize(0).build()); Worker worker = factory.newWorker("task-queue");

Debug Workflows in development environment

When developing Workflows with the Python SDK, you can use normal development tools including logging and a debugger to see what's happening in your Workflow. In addition, the Web UI and Temporal CLI provide visibility into Workflows.

Worker performance debugging and tuning

Worker performance can be debugged and tuned using metrics and the Worker performance guide. SDK metrics must be configured to monitor Worker performance.

Workflow Replay for production debugging

Replay is one of the tools available for debugging production Workflows. It allows you to replay events and verify Workflow behavior.

Debug Workflows in production environment

Production Workflows can be debugged using the Web UI, Temporal CLI, Replay, Tracing, and Logging. Each tool provides different insights into Workflow execution in production.

Example: Testing Workflow end-to-end

```python import uuid import pytest from temporalio import activity from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker from activities import greet from workflows import SayHelloWorkflow @pytest.mark.asyncio async def test_say_hello_workflow(): """Execute the workflow end-to-end with its real activity.""" task_queue_name = str(uuid.uuid4()) async with await WorkflowEnvironment.start_local(ui=True, ui_port=8233) as env: async with Worker( env.client, task_queue=task_queue_name, workflows=[SayHelloWorkflow], activities=[greet], ): result = await env.client.execute_workflow( SayHelloWorkflow.run, "Temporal", id=str(uuid.uuid4()), task_queue=task_queue_name, ) assert result == "Hello Temporal" ``` This example shows how to set up a local Workflow environment, create a Worker with real Activities, and execute a Workflow end-to-end.

Time skipping support in test server

The test server in the Python SDK supports skipping time. Use the test server for both end-to-end and integration tests with Workers because it enables time skipping.

WorkflowEnvironment.from_client for existing Temporal Server

Use the from_client() method to work with an existing Temporal Server.

Manual time skipping with WorkflowEnvironment.sleep

To implement time skipping manually, use the sleep method inside the WorkflowEnvironment. This will manually advance time by the duration you specify.

Pytest recommended testing framework for Python SDK

Pytest is a recommended framework for testing with the Temporal Python SDK. When using pytest, consider using the -s flag (--show-capture=no) to see logs live. Pytest provides fixtures to stand up and tear down test environments, useful test discovery, and easy parameterized tests.

ActivityEnvironment for testing Activities

To test an Activity in isolation, use the ActivityEnvironment class. This class allows you to run any callable inside an Activity context and test the behavior of your code under various conditions. It provides a way to mock the Activity context, listen to Heartbeats, and cancel the Activity.

Testing Activity Heartbeats with on_heartbeat

To test Heartbeats in an Activity, use the on_heartbeat property of the ActivityEnvironment class. This property sets a custom function that is called every time the activity.heartbeat() function is called within the Activity.

Example: Testing Activity Heartbeats

```python @activity.defn async def activity_with_heartbeats(param: str): activity.heartbeat(f"param: {param}") activity.heartbeat("second heartbeat") env = ActivityEnvironment() heartbeats = [] # Set the `on_heartbeat` property to a callback function that will be called for each Heartbeat sent by the Activity. env.on_heartbeat = lambda *args: heartbeats.append(args[0]) # Use the run method to start the Activity, passing in the function that contains the Heartbeats and any necessary parameters. await env.run(activity_with_heartbeats, "test") # Verify that the expected Heartbeats are received by the callback function. assert heartbeats == ["param: test", "second heartbeat"] ``` This example shows how to use ActivityEnvironment to test an Activity with Heartbeats by capturing them in a callback function.

WorkflowEnvironment.start_local for testing Workflows

Use WorkflowEnvironment.start_local() to configure a local environment for running and testing Workflows end-to-end. You can pass ui=True to view the Temporal UI during testing.

Mocking Activities in Workflow tests

When testing Workflow logic in isolation, you can mock Activities by providing mock Activity implementations to the Worker. The mocked Activity implementation should have the same signature as the real implementation (including input and output types) and the same name (using the name parameter in @activity.defn if needed). When the Workflow invokes the Activity, it invokes the mocked implementation instead of the real one.

Example: Mocking Activities in Workflow tests

```python import uuid from temporalio.client import Client from temporalio.worker import Worker # Import your Activity Definition and real implementation from hello.hello_activity import ( ComposeGreetingInput, GreetingWorkflow, compose_greeting, ) # Define your mocked Activity implementation @activity.defn(name="compose_greeting") async def compose_greeting_mocked(input: ComposeGreetingInput) -> str: return f"{input.greeting}, {input.name} from mocked activity!" async def test_mock_activity(client: Client): task_queue_name = str(uuid.uuid4()) # Provide the mocked Activity implementation to the Worker async with Worker( client, task_queue=task_queue_name, workflows=[GreetingWorkflow], activities=[compose_greeting_mocked], ): # Execute your Workflow as usual assert "Hello, World from mocked activity!" == await client.execute_workflow( GreetingWorkflow.run, "World", id=str(uuid.uuid4()), task_queue=task_queue_name, ) ``` This example demonstrates how to provide a mocked Activity implementation to a Worker for testing Workflow logic in isolation.

Time is global in WorkflowEnvironment

Time is a global property of an instance of WorkflowEnvironment. Skipping time (either automatically or manually) applies to all currently running tests. If different tests need different time behaviors, run tests in a series or with separate instances of the test server.

WorkflowEnvironment.start_time_skipping for automatic time skipping

Use the start_time_skipping() method to start a test server process and skip time automatically. In time-skipping mode, Timers (including sleeps and conditional timeouts) are fast-forwarded except when Activities are running.

WorkflowEnvironment.start_local for full local Temporal Server

Use the start_local() method for testing a full local Temporal Server. There is no time skipping in this environment, so any Timers will wait the actual amount of time.

Example: Manual time skipping

```python from temporalio.testing import WorkflowEnvironment async def test_manual_time_skipping(): async with await WorkflowEnvironment.start_time_skipping() as env: # Your code here # You can use the env.sleep(seconds) method to manually advance time await env.sleep(3) # This will advance time by 3 seconds # Your code here ``` This example shows how to manually advance time in a test using the sleep method of WorkflowEnvironment.

Workflow Execution replay for testing changes

Replay recreates the exact state of a Workflow Execution by replaying it from the beginning of its Event History. Replay succeeds only if the Workflow Definition is compatible with the provided history from a deterministic point of view. When testing changes to Workflow Definitions, download Event Histories of representative sets of recent open and closed Workflows from each Task Queue and run them through replay as part of CI checks.

Replayer class for replaying Workflow Executions

To replay Workflow Executions, use the replay_workflows or replay_workflow methods of the Replayer class, passing one or more Event Histories as arguments. If any Event History is non-deterministic, an error is thrown. You can set the fail_fast option to false to wait until all histories have been replayed instead of failing immediately.

Example: Replaying Workflow Executions from server

```python workflows = client.list_workflows(f"TaskQueue=foo and StartTime > '2022-01-01T12:00:00'") histories = workflows.map_histories() replayer = Replayer( workflows=[MyWorkflowA, MyWorkflowB, MyWorkflowC] ) await replayer.replay_workflows(histories) ``` This example shows how to download Event Histories from the server and replay them (requires Advanced Visibility enabled as of server v1.18).

Example: Replaying Workflow Execution from JSON string

```python replayer = Replayer(workflows=[YourWorkflow]) await replayer.replay_workflow(WorkflowHistory.from_json(history_json_str)) ``` This example shows how to load a single Workflow history from a JSON string and replay it. Event Histories exported by Temporal Web UI or Temporal CLI can be passed as JSON strings or as Python dictionaries via json.load().

Protobuf encoding in Event Histories

When fetching Event Histories directly from the server or exporting them, the data can be protobuf-encoded (bytes). The Replayer often works with decoded histories (like a dict). If TypeError exceptions occur related to dict vs. bytes mismatches during replay, ensure the Event History is properly decoded before passing it to the Replayer.

Python replay testing example with verify mode

Example Python code for replay testing in verify mode: fetch workflows from a task queue within a time period using client.list_workflows() with a TaskQueue and StartTime filter, map to histories using map_histories(), create a Replayer with workflows, and call await replayer.replay_workflows(histories). If any Workflows fail to replay, an error is thrown.

Replay testing validates determinism before deployment

Replay testing is the best way to verify that code changes won't cause non-determinism errors once deployed. Replay testing takes one or more existing Event Histories that ran against a previous version of Workflow code and runs them against current Workflow code, verifying compatibility.

Give your agent this brain