Testing
Pykka actors can be tested using the regular Python
testing tools like pytest,
unittest
, and unittest.mock
.
To test actors in a setting as close to production as possible, a typical pattern is the following:
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.
In the test, assert on the actor’s state or the return value from the
ask()
.In the test teardown, stop the actor to properly clean up before the next test.
An example
Let’s look at an example actor that we want to test:
import pykka
class ProducerActor(pykka.ThreadingActor):
def __init__(self, consumer):
super().__init__()
self.consumer = consumer
def produce(self):
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:
import pytest
from producer import ProducerActor
@pytest.fixture()
def consumer_mock(mocker):
return mocker.Mock()
@pytest.fixture()
def producer(consumer_mock):
# 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, producer):
# 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.