Home Artificial Intelligence Python Mocking in Production Recap of Vital Topics Problem Setup Testing Conclusion

Python Mocking in Production Recap of Vital Topics Problem Setup Testing Conclusion

0
Python Mocking in Production
Recap of Vital Topics
Problem Setup
Testing
Conclusion

Advanced mocking techniques introduced via an example

Towards Data Science
Photo by Joyce Hankins on Unsplash

Unit testing is an art. While the query of what to check is crucial (you’ll be able to’t test all the things), on this post we’ll take a better have a look at some advanced testing and mocking techniques in Python. While a few of these were already introduced in previous postings (Part 1, Part 2, Part 3), this post differs by showing their interplay on a real-world example, in addition to goes above the scope of previous posts and adds more insights.

Amongst others, we’ll discuss:

  • patching functions — with the extra requirements of patching async functions and modifying their behaviour
  • asserting that functions are called as expected
  • writing custom matchers for partially matching call arguments

Before diving into the actual contents of this post, we first give a fast recap of the best way to use pytest. Afterwards, we do the identical for asyncio, since our example further on makes use of this. For those who are already conversant in these topics, please be at liberty to skip to the subsequent section.

pytest

Unit testing must be a part of any somewhat skilled software product. It helps avoid bugs and thus increases efficiency and quality of the code. While Python natively ships with unittest, pytest is prefered by many developers.

To start, let’s have a look at one example from a previous post:

import pytest

class 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 will run this test via python -m pytest test_filename.py. When doing so, pytest discovers all tests in that file following some conventions (e.g. all functions named test_…) and executes them. In our case, we defined a fixture returning an instance of MySummationClass. Fixtures might be used to e.g. avoid repetitive initialisation and to moduralize code. We then call that instance’s sum() method, and check that the result equals the expected one via assert.

Mocking

Often, during testing we encounter functions we either can’t or don’t wish to run — e.g. because they’d take too long or have undesired unintended effects. For this purpose, we are able to mock them out.

Let’s consider an example from the previous post:

import time
from unittest.mock import Mock, patch

def 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’re using a decorator to patch expensive_function with mock_expensive_function, this manner replacing the unique function’s future time by a function with similar properties, chosen by us.

asyncio

Lastly, let’s briefly recap asyncio: asyncio is a multi-threading library whose primary area of application is I/O sure applications — that’s applications, which spend a big portion of their time waiting for inputs or outputs. asyncio actually uses a single thread for this, and leaves it as much as the developer to define when coroutines can yield execution and hand over to others.

Let’s re-use the motivating example from the previous post:

import asyncio

async def sleepy_function():
print("Before sleeping.")
await asyncio.sleep(1)
print("After sleeping.")

async def fundamental():
await asyncio.gather(*[sleepy_function() for _ in range(3)])

asyncio.run(fundamental())

If we were to run sleepy_function conventionally 3 times in a row, this may take 3s. Nevertheless, with asyncio this program finishes in 1s: gather schedules the execution of three function calls, and inside sleepy_function the keyword await yields control back to the fundamental loop, which has time to execute other code (here: other instances of sleepy_function) while sleeping for 1s.

Now, equipped with sufficient prior knowledge, let’s dive deeper into the actual contents of this post. Particularly, on this section we first define the programming problem serving as playground for unit testing.

For organising the project, we used poetry, and likewise followed other best practises, corresponding to using typing, formatters and linters.

Our example models generating some messages and sending them via some client (e.g. email): within the fundamental file, we first instantiate the client via a factory function, then generate some messages, and lastly send these messages asynchronously using the client.

The project consists of the next files, which you can even find on github:

pyproject.toml

[tool.poetry]
name = "Pytest Example"
version = "0.1.0"
description = "A somewhat larger pytest example"
authors = ["hermanmichaels "]

[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 asyncio

from 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 fundamental() -> None:
asyncio.run(send_messages())

if __name__ == "__main__":
fundamental()

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 will run this program via python message_sending.py, which — as stated above — first instantiates a MessageClient, then generates an inventory of dummy messages via generate_messages, and eventually sends these using asyncio. Within the last step, we create tasks message_client.send(message) for each message, after which run these asynchronously via gather.

With that, let’s come to testing. Here, our goal is to create some scenarios, and to be certain that the right messages are being send out via the message client. Naturally, our easy demo setting is simply too simplistic to cover this, but imagine the next: in the true, you’re using the client to send out messages to customers. Depending on certain events (e.g. product bought / sold), there will likely be different messages created. You thus wish to simulate these different scenarios (e.g. mock someone buying a product), and check, that the precise emails are being generated and sent out.

Sending actual emails during testing might be not desired though: it could put stress on the e-mail server, and would require certain setup steps, corresponding to entering credentials, etc. Thus, we would like to mock out the message client, particularly it’s send function. Throughout the test we then simply put some expectations on this function, and confirm it was called as expected (e.g. with the precise messages). Here, we won’t mock generate_messages: while actually possible (and desired in some unit tests), the thought here is to not touch the message generating logic — while obviously very simplistic here, in an actual system the messages can be generated based on certain conditions, which we would like to check (one could thus call this more of an integration test, than an isolated unit test).

Test Function was Called Once

For a primary try, let’s change generate_messages to only create a single message. Then, we expect the send() function to be called once, which we’ll test here.

That is how the corresponding test looks:

from unittest.mock import AsyncMock, Mock, call, patch

import 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 dissect this in additional details: test_send_messages is our fundamental test function. We patched the function generate_message_client, as a way to not use the true (email) client returned in the unique function. Concentrate to “where to patch”: generate_message_client is defined in message_utils, but because it is imported via from message_utils import generate_message_client, we now have to focus on message_sending because the patch goal.

We’re not done yet though, as a result of asyncio. If we continued without adding more details to the mocked message client, we’d get an error just like the next:

TypeError: An asyncio.Future, a coroutine or an awaitable is required … ValueError(‘The longer term belongs to a distinct loop than ‘ ‘the one specified because the loop argument’).

The explanation for that is that in message_sending we call asyncio.gather on message_client.send. Nevertheless, as of now, the mocked message client, and consequently its send message, are simply Mock objects, which can’t be scheduled asynchronously. So as to get around this, we introduced the fixture message_client_mock. On this, we define a Mock object called message_client_mock, after which define its send method as an AsyncMock object. Then, we assign this as return_value to the generate_message_client function.

Note that pytest natively actually doesn’t support asyncio, but needs the package pytest-asyncio, which we installed within the pyproject.toml file.

Test Function was Called Once With Specific Argument

As a next step, we not only want to envision send was called once, as expected — but in addition ensure it was called with the precise arguments — i.e. the precise message.

For this, we first overload the equals operator for Message:

def __eq__(self, other: object) -> bool:
if not isinstance(other, Message):
return NotImplemented
return self.body == other.body

Then, at the tip of the test, we use the next expectation:

message_client_mock.send.assert_called_once_with(Message("Message 1"))

Partially Matching Arguments

Sometimes, we’d wish to do some “fuzzy” matching — that’s, don’t check for the precise arguments a function was called with, but check some portion of them. To stick with our user story of sending emails: imagine, the actual emails accommodates plenty of text, of which some is somewhat arbitrary and specific (e.g. a date / time).

To do that, we implement a brand new proxy class, which implements the __eq__ operator w.r.t. Message. Here, we simply subclass string, and check it being contained in message.body:

class MessageMatcher(str):
def __eq__(self, other: object):
if not isinstance(other, Message):
return NotImplemented
return self in other.body

We will then assert that the sent message e.g. accommodates a “1”:

message_client_mock.send.assert_called_once_with(MessageMatcher("1"))

Checking Arguments for Multiple Calls

Naturally, only having the ability to check that a function was called once will not be very helpful. If we would like to envision a function was called N times with different arguments, we are able to use assert_has_calls. This expects an inventory of of type call, with each element describing one expected call:

message_client_mock.send.assert_has_calls(
[call(Message("Message 1")), call(MessageMatcher("2"))]
)

This brings us to the tip of this text. After recapping the fundamentals of pytest and asyncio, we dove right into a real-world example and analysed some advanced testing, and particularly mocking, techniques.

We saw the best way to test and mock async functions, the best way to assert they’re called as expected, and the best way to chill out equality checks for the expected arguments.

I hope you enjoyed reading this tutorial — thanks for tuning in!

LEAVE A REPLY

Please enter your comment!
Please enter your name here