Unit testing is an art. While the question of what to test is crucial (you can’t test everything), in this post we’ll take a closer look at some advanced testing and simulation techniques in Python. While some of these have already been introduced in previous publications (Part 1, Part 2, part 3), this post differs by showing their interaction in a real-world example, as well as going beyond the scope of previous posts and adding more information.
Among others, we will talk about:
- patch functions — with the additional requirements of patching asynchronous functions and modifying their behavior
- assert that functions are called as expected
- write custom matchers for partially matching call arguments
Before we delve into the actual content of this post, we first do a quick rundown of how to use pytest. Then we do the same for asyncio, as our example below makes use of this. If you’re already familiar with these topics, feel free to skip ahead to the next section.
pytest
Unit tests should be part of any somewhat professional software product. It helps to avoid errors and therefore increases the efficiency and quality of the code. While Python ships natively with unit test, pytest it is preferred by many developers.
To get started, let’s look at an example from a previous post:
import pytestclass MySummationClass:
def sum(self, x, y):
return x + y
@pytest.fixture
def my_summation_class():
return MySummationClass()
def test_sum(my_summation_class):
assert my_summation_class.sum(2, 3) == 5
We can run this test through python -m pytest test_filename.py. By doing so, pytest discovers all the tests in that file by following some conventions (for example, all functions named test_…) and runs them. In our case, we define a accessory returning an instance of MySummationClass. Props can be used to, for example, avoid repetitive initialization and to modularize code. So we call that instance sum() method, and verify that the result is the same as expected through assert.
Mocking
Often during testing we come across features that we can’t or don’t want to run, for example because they would take too long or have unwanted side effects. For this purpose, we can make fun of them.
Let’s consider an example from the previous post:
import time
from unittest.mock import Mock, patchdef expensive_function() -> int:
time.sleep(10)
return 42
def function_to_be_tested() -> int:
expensive_operation_result = expensive_function()
return expensive_operation_result * 2
@patch("sample_file_name.expensive_function")
def test_function_to_be_tested(mock_expensive_function: Mock) -> None:
mock_expensive_function.return_value = 42
assert function_to_be_tested() == 84
we are using a decorator Patch expensive_function with mock_expensive_functionthus replacing the long running time of the original function with a function with similar properties, chosen by us.
asyncio
Finally, let’s briefly recap asyncio: asyncio is a multi-threaded library whose main area of application is I/O-bound applications, that is, applications that spend a large part of their time waiting for input or output. asyncio actually uses a single thread for this, and leaves it up to the developer to define when coroutines can give execution and deliver to others.
Let’s reuse the motivating example from the previous post:
import asyncioasync def sleepy_function():
print("Before sleeping.")
await asyncio.sleep(1)
print("After sleeping.")
async def main():
await asyncio.gather(*[sleepy_function() for _ in range(3)])
asyncio.run(main())
if we had to run sleepy_function conventionally three times in a row, this would take 3s. However, with asyncio this program ends in 1s: gather programs the execution of three function calls, and inside sleepy_function the key word await returns control to the main loop, which has time to execute other code (here: other instances of sleepy_function) while sleeping for 1 s.
Now, equipped with enough background knowledge, let’s dive into the actual content of this post. In particular, in this section we first define the programming problem that serves as the playing field for unit tests.
To configure the project, we use poetryand also followed others better practiceshow to use typingformatters and linters.
Our example models generate some messages and send them via some client (e.g. email): In the main file, we first instantiate the client via a factory function, then generate some messages, and lastly , we send these messages asynchronously using the client.
The project consists of the following files, which you can also find at github:
pyproject.toml
[tool.poetry]
name = "Pytest Example"
version = "0.1.0"
description = "A somewhat larger pytest example"
authors = ["hermanmichaels <[email protected]>"][tool.poetry.dependencies]
python = "3.10"
mypy = "0.910"
pytest = "7.1.2"
pytest-asyncio = "0.21.0"
black = "22.3.0"
flake8 = "4.0.1"
isort = "^5.10.1"
message_sending.py
import asynciofrom message_utils import Message, generate_message_client
def generate_messages() -> list[Message]:
return [Message("Message 1"), Message("Message 2")]
async def send_messages() -> None:
message_client = generate_message_client()
messages = generate_messages()
await asyncio.gather(*[message_client.send(message) for message in messages])
def main() -> None:
asyncio.run(send_messages())
if __name__ == "__main__":
main()
message_utils.py
from dataclasses import dataclass@dataclass
class Message:
body: str
class MessageClient:
async def send(self, message: Message) -> None:
print(f"Sending message: {message}")
def generate_message_client() -> MessageClient:
return MessageClient()
We can run this program through python message_sending.pywhich, as stated above, first instantiates a MessageClientthen generates a list of dummy messages via generate_messages, and eventually sends them using asyncio. In the last step, we create tasks. message_client.send(message) for each message, and then run them asynchronously via gather.
With that, let’s move on to the test. Here our goal is to create some scenarios and ensure that the correct messages are sent through the message client. Naturally, our simple demo setup is too simplistic to cover this, but imagine this: In reality, you’re using the client to send messages to clients. Depending on certain events (eg product bought/sold), different messages will be created. So you want to simulate these different scenarios (for example, pretend someone buys a product) and verify that the correct emails are generated and sent.
However, you probably don’t want to send actual emails during the test: it would put stress on the email server and require certain configuration steps like entering credentials etc. Therefore, we want to simulate the message client, in particular it is send function. During testing, we simply put some expectations on this function and verify that it was called as expected (for example, with the correct messages). Here, we will not mock generate_messages: while possible (and desired in some unit tests), the idea here is to leave the message generation logic untouched; although it’s obviously very simplistic here, in a real system messages would be generated based on certain conditions, which we want to test for. (So you could call this more of an integration test than an isolated unit test.)
The test function was called once
For a first try, let’s change generate_messages to create a single message. So we wait for the send() function to be called once, which we’ll test here.
This is what the corresponding test looks like:
from unittest.mock import AsyncMock, Mock, call, patchimport pytest as pytest
from message_sending import send_messages
from message_utils import Message
@pytest.fixture
def message_client_mock():
message_client_mock = Mock()
message_client_mock_send = AsyncMock()
message_client_mock.send = message_client_mock_send
return message_client_mock
@pytest.mark.asyncio
@patch("message_sending.generate_message_client")
async def test_send_messages(
generate_message_client: Mock, message_client_mock: Mock
):
generate_message_client.return_value = message_client_mock
await send_messages()
message_client_mock.send.assert_called_once()
Let’s break this down in more detail: test_send_messages is our main testing function. We patched the function generate_message_client, so as not to use the actual customer (email) returned in the original function. Pay attention to “where to patch”: generate_message_client is defined in message_utilsbut since it is imported via from message_utils import generate_message_clientwe have to aim message_sending as the target of the patch.
However, we are not done yet, due to asyncio. If we continued without adding more details to the mock message client, we would get an error similar to the following:
TypeError: An asyncio.Future, a coroutine, or an awaitable is required… ValueError(‘The future belongs to a different ‘ ‘loop than the one specified as the loop’s argument’).
The reason for this is that in message_sending we call asyncio.gather in message_client.send. However, from now on, the simulated message client and consequently its send message, they are simply Mock objects, which cannot be scheduled asynchronously. To avoid this, we introduced the accessory message_client_mock. In this, we define a Mock named object message_client_mockand then define your send method as a AsyncMock object. So, we assign this as return_value toward generate_message_client function.
Note that pytest natively doesn’t actually support asyncio, but you need the package pytest-asynciowhich we install in the pyproject.toml file.
The test function was called once with a specific argument
As a next step, we want to not only check send was called once, as expected, but also make sure it was called with the correct arguments, i.e. the correct message.
For this, we first overload the equality operator to Message:
def __eq__(self, other: object) -> bool:
if not isinstance(other, Message):
return NotImplemented
return self.body == other.body
Then, at the end of the test, we use the following expectation:
message_client_mock.send.assert_called_once_with(Message("Message 1"))
Partially matching arguments
Sometimes we may want to do some “fuzzy” matching, that is, not check the exact arguments a function was called with, but check a part of them. To stay with our email sending user story: Imagine, real emails contain a lot of text, some of which is somewhat arbitrary and specific (for example, a date/time).
To do this, we implement a new proxy class, which implements the __eq__ wrt operator Message. Here, we simply create a subclass of string and check that it is contained in message.body:
class MessageMatcher(str):
def __eq__(self, other: object):
if not isinstance(other, Message):
return NotImplemented
return self in other.body
So we can affirm that the message sent, for example, contains a “1”:
message_client_mock.send.assert_called_once_with(MessageMatcher("1"))
Argument checking for multiple calls
Naturally, just being able to check that a function was called once isn’t very useful. If we want to check that a function was called N times with different arguments, we can use assert_has_calls. This expects a list of type callwith each element describing an expected call:
message_client_mock.send.assert_has_calls(
[call(Message("Message 1")), call(MessageMatcher("2"))]
)
This brings us to the end of this article. After recapping the basics of pytest and asyncio, we dive into a real world example and discuss some advanced testing and, in particular, simulation techniques.
We saw how to test and mock asynchronous functions, how to assert that they are called as expected, and how to relax equality checks for expected arguments.
I hope you enjoyed reading this tutorial. Thanks for tuning in!
