How to test your Temporal application code
Ensuring that your Temporal Application is robust requires the right testing strategies.
In this guide, we’ll dive into the various ways to test your Workflow and Activity code using the tools and frameworks provided by Temporal or your language’s idiomatic features.
What are the Temporal testing environments?
Temporal’s SDKs come with a testing suite.
In the Python SDK, Temporal uses the testing package, where it can help set up testing environments for your Workflow or Activity.
The Workflow testing Environment comes with a start_time_skipping
function, in the case of long-running Workflows.
Testing Activities
Activities can be tested with the help of a mocked environment provided by the temporalio.testing.ActivityEnvironment
class. This environment allows:
- Customizing returned values with the
info
property. - Handling
activity.heartbeat()
calls using theon_heartbeat
property. - Simulating activity cancellations with the
cancel()
method. - Emulating worker shutdown during activity execution with the
worker_shutdown()
method.
Running an Activity with a Mocked Context
If an Activity requires context, you need to mock it during testing.
Utilize the ActivityEnvironment
class for this purpose.
Monitoring Heartbeats
Being able to verify heartbeats sent by an Activity is crucial.
The on_heartbeat()
property of the ActivityEnvironment
class lets you do just that.
Here’s a simple Python test example illustrating this:
Testing Workflows
Mocking Activities
While unit testing Workflows, you might need to mock Activity invocations.
During integration testing, provide mocked activity implementations to the Worker.
Here’s an example in Python:
For long-running Workflows, Temporal’s test framework lets you skip time, and complete your tests in seconds rather than the Workflow’s specified amount.
For example, if you have a Workflow sleep for a day, or have an Activity failure with a long retry interval, you don’t need to wait the entire length of the sleep period to test whether the sleep function works.
Instead, test the logic that happens after the sleep by skipping forward in time and complete your tests in a timely manner.
Automatically Skipping Time
The Temporal SDK allows you to automatically skip time during tests. To automatically advance time, use the start_time_skipping()
method. For running a full local Temporal Server, utilize the start_local()
method, and for an existing Temporal Server, opt for the from_client()
method.
Manually Skipping Time
To manually control time during tests, use the start_time_skipping()
method:
Time-Skipping Server
Temporal offers a unique time-skipping server. It’s useful for testing Workflows that either have long sleep durations or where you want to observe the effect of timeouts on the system. To start this server, use:
It’s important to note that this server is a product of our Java SDK, compiled natively via GraalVM.
Recommended Testing Frameworks
For the Temporal SDK in Python, pytest is a recommended testing framework. It offers features like environment setup/teardown fixtures, test discovery, and parameterized testing.
Note: Always remember that while testing is essential, the type of test (end-to-end, integration, or unit) should be chosen based on the specific scenario and requirements of your Temporal application.