Unit testing

Pykka actors can be tested using the regular Python testing tools like pytest, unittest, and unittest.mock.

General approach

To test actors in a setting as close to production as possible, a typical pattern is the following:

1. In the test setup, start an actor together with any actors/collaborators it depends on. The dependencies will often be replaced by mocks to control their behavior.
2. In the test, ask() or tell() the actor something.
3. In the test, assert on the actor's state or the return value from the ask() call.
4. In the test teardown, stop the actor to properly clean up before the next test.

Example

Let's look at an example actor that we want to test:

examples/testing.py

from typing import Any

import pykka


class ProducerActor(pykka.ThreadingActor):
    def __init__(self, consumer: pykka.ActorProxy[Any]) -> None:
        super().__init__()
        self.consumer = consumer

    def produce(self) -> None:
        new_item = {"item": 1, "new": True}
        self.consumer.consume(new_item)

We can test this actor with pytest by mocking the consumer and asserting that it receives a newly produced item:

examples/testing_test.py

from __future__ import annotations

from typing import TYPE_CHECKING, Any

import pytest
from testing import ProducerActor

if TYPE_CHECKING:
    from collections.abc import Generator
    from unittest.mock import Mock

    from pytest_mock import MockerFixture

    from pykka import ActorProxy


@pytest.fixture
def consumer_mock(mocker: MockerFixture) -> Any:
    return mocker.Mock()


@pytest.fixture
def producer(consumer_mock: Mock) -> Generator[ActorProxy[ProducerActor]]:
    # Step 1: The actor under test is wired up with
    # its dependencies and is started.
    proxy = ProducerActor.start(consumer_mock).proxy()

    yield proxy

    # Step 4: The actor is stopped to clean up before the next test.
    proxy.stop()


def test_producer_actor(
    consumer_mock: Mock,
    producer: ActorProxy[ProducerActor],
) -> None:
    # Step 2: Interact with the actor.
    # We call .get() on the last future returned by the actor to wait
    # for the actor to process all messages before asserting anything.
    producer.produce().get()

    # Step 3: Assert that the return values or actor state is as expected.
    consumer_mock.consume.assert_called_once_with({"item": 1, "new": True})

If this way of setting up and tearing down test resources is unfamiliar to you, it is strongly recommended to read up on pytest's great fixture feature.