Test and run a Worker

Add a unit test

The Temporal Python SDK includes functions that help you test your Workflow executions. Let's add tests to the application to make sure the Workflow works as expected.

Create a new folder in your project directory called tests:

mkdir tests

Create an empty __init__.py file within that directory:

touch tests/__init__.py

Create the /test_run_workflow.py file in the tests directory and add the following content to test the Workflow:

touch tests/test_run_workflow.py

import uuid

import pytest

from temporalio import activity
from temporalio.worker import Worker
from temporalio.testing import WorkflowEnvironment

from activities import say_hello
from workflows import SayHello

@pytest.mark.asyncio
async def test_execute_workflow():
    task_queue_name = str(uuid.uuid4())
    async with await WorkflowEnvironment.start_time_skipping() as env:

        async with Worker(
            env.client,
            task_queue=task_queue_name,
            workflows=[SayHello],
            activities=[say_hello],
        ):
            assert "Hello, World!" == await env.client.execute_workflow(
                SayHello.run,
                "World",
                id=str(uuid.uuid4()),
                task_queue=task_queue_name,
            )

This code snippet imports the uuid and pytest packages, along with Activity and Worker from the Temporal SDK. It then imports WorkflowEnvironment from the Temporal SDK so you can create an environment for testing. It then imports your Activity and Workflow.

The test function test_execute_workflow creates a WorkflowEnvironment so it can run the tests. It then creates a random Task Queue name and initiates the Worker with env.client.execute_workflow. It then checks if the result of the Workflow Execution is Hello, World! when the input parameter is World.

This code tests the Workflow and invokes the actual say_hello Activity. However, you may want to test your Workflows and mock out the Activity so you can see how your Workflow responds to different inputs and results.

Add the following code to create a test that uses a mocked say_hello Activity:

# ...


@activity.defn(name="say_hello")
async def say_hello_mocked(name: str) -> str:
    return f"Hello, {name} from mocked activity!"


@pytest.mark.asyncio
async def test_mock_activity():
    task_queue_name = str(uuid.uuid4())
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client,
            task_queue=task_queue_name,
            workflows=[SayHello],
            activities=[say_hello_mocked],
        ):
            assert "Hello, World from mocked activity!" == await env.client.execute_workflow(
                SayHello.run,
                "World",
                id=str(uuid.uuid4()),
                task_queue=task_queue_name,
            )

This creates a function called say_hello_mocked which the Workflow test will use as the mock Activity function. The test_mock_activity test then checks that the outcome of the Workflow is "Hello, World from mocked activity!" for the passed input parameter World, using the same type of test setup as the previous test function.

Run the following command from the project root to start the tests:

pytest

This command will search for files in your tests folder that match the pattern test_*.py or *_test.py.

You'll see output similar to the following from your test run indicating that the test was successful:

===================== test session starts =====================
platform darwin -- Python 3.10.9, pytest-7.2.0, pluggy-1.0.0
rootdir: /hello-world-python-getting-started
plugins: asyncio-0.20.3, anyio-3.6.2
asyncio: mode=strict
collected 2 items

tests/test_run_workflow.py ..                             [100%]
===================== 2 passed in 10.29s ======================

You can also pass the command line option --workflow-environment at runtime to change the test environment.

Most test suites reuse the local environment across tests. You can explore fixtures in Pytest to set this up.

You've built a test suite and you've successfully tested your Workflow. You can reuse the conftest.py file you've built in future Temporal Python projects.

You have a working Temporal Application and tests that make sure the Workflow executes as expected. Next, you'll configure a Worker to execute your Workflow.

Configure a Worker

A Worker hosts Workflow and Activity functions and executes them. The Temporal Cluster tells the Worker to execute a specific function from information it pulls from the Task Queue. After the Worker runs the code, it communicates the results back to the Temporal Cluster.

When you start a Workflow, you tell the server which Task Queue the Workflow and Activities use. A Worker listens and polls on the Task Queue, looking for work to do.

To configure a Worker process using the Python SDK, you'll connect to the Temporal Cluster and give it the name of the Task Queue to poll.

You'll connect to the Temporal Cluster using a Temporal Client, which provides a set of APIs to communicate with a Temporal Cluster. You'll use Clients to interact with existing Workflows or to start new ones.

In this tutorial you'll create a small standalone Worker program so you can see how all of the components work together.

Create the file run_worker.py in the root of your project and add the following code to connect to the Temporal Server, instantiate the Worker, and register the Workflow and Activity:

import asyncio

from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.worker import Worker

from activities import say_hello
from workflows import SayHello

async def main():
    client = await Client.connect("localhost:7233", namespace="default")
    # Run the worker
    worker = Worker(
        client, task_queue="hello-task-queue", workflows=[SayHello], activities=[say_hello]
    )
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

This program connects to the Temporal Cluster using client.Connect. In this example, you only need to provide a target host and a Namespace. Since you're running this locally, you use localhost:7233 for the target host, and you specify the optional default Namespace name, default.

You then create a new Worker instance by specifying the client, the Task Queue to poll, and the Workflows and Activities to monitor. Then you run the worker.

You've created a program that instantiates a Worker to process the Workflow. Now you need to start the Workflow.