Async Python Mock Testing

May 1, 2020 (4y ago)

Archive

The standard library does not really provide helpful practical information over the topic, nor does Pytest docs, so I figured, I'll write a quick blog post on how to do it. There are third party libraries of course for this subject but I'm not a fan of using third parties, unless absolutely necessary, plus you should understand how the subject works before abstracting it.

I've previously explained how the whole async concept works under the hood in Python. While not all the details are essential for this task, it's beneficial for grasping the nuances of asynchronous logic before attempting to test it.

With that out of the way, lets start with the two main ways you can patch an async object.

Monkey Patch

Given a function some_result of the class Foo

class Foo:
    @staticmethod
    async def some_result(previous_status: bool) -> Result:
        return Result.OK if previous_status else Result.Err

With this enumerable as a return type:

1import enum 
2
3class Result(enum.Enum):
4    Err = enum.auto()
5    OK = enum.auto()
6

The goal is to mock the behavior of the some_result() method in the Foo class under different conditions. The method takes a boolean which is previous_status that we might not control, and returns a Result. Depending on the value of previous_status, the method should return either OK or Err.

For this we'll need to monkey patch it.

1import pytest
2from _pytest.monkeypatch import MonkeyPatch
3
4@pytest.fixture
5def mock_get_result(monkeypatch: MonkeyPatch) -> None:
6    async def _always_ok(_previous_status: bool) -> Result: 
7        return Result.OK
8    monkeypatch.setattr(target=Foo, name="get_result", value=_always_ok)
9

Pytest allows us to define fixtures, which are functions that set up or provide resources for our tests, so we don't keep repeating the logic again and again for each test case. In this case we only have one fixture.

mock_foo_result_ok monkey patches the some_result() method of the Foo class. it replaces it with the inner function _always_ok, meaning when we pass this fixture as an argument to any test function whenever we use some_result() in that particular test, it comes patched already to always return whatever we want it to, in this case it is the Ok variant. And by the way with Pytest, when dealing with asynchronous test cases, you have to mark them as @pytest.mark.asyncio and also install an async plugin, for me I'm using pytest-asyncio. There are many others of course.

@pytest.mark.asyncio
async def test(mock_get_result) -> None:
    r1 = await Foo.get_result(True) # either True or Flase 
    r2 = await Foo.get_result(False) # it does not matter

    assert r1 == Result.OK # passes 
    assert r2 == Result.OK # passes 

AsyncMock Patch

Say I have a library called playground where the project structure is as follows (I use poetry btw)

(.venv) ➜  playground 
.
├── README.md
├── __init__.py
├── foo.py
├── poetry.lock
├── pyproject.toml
└── test.py

The foo module has a class that contains another class which has a method that uses httpx to request API data asynchronously, this is a dummy example of course but you get the picture, we're trying to test the functionality of an object (function) that's two modules deep.

1import httpx
2
3class Foo:
4    class Bar:
5        @staticmethod
6        async def fetch(url: str) -> httpx.Response :
7            async with httpx.AsyncClient() as client:
8                response = await client.get(url)
9                return response
10

How do you mock the fetch method ?

For this we will need AsyncMock (Python 3.8+), which is an enhanced version of the synchronous Mock object, supporting async functions which returns awaitables instead of normal functions.

from unittest.mock import AsyncMock

We'll also need patch() to patch the object in hand.

from unittest.mock import AsyncMock, patch

I'll be mocking the status code of httpx.Response object along with the JSON data that the .json() method returns.

1import pytest
2
3from unittest.mock import AsyncMock, patch
4from playground.foo import Foo
5
6FAKE_JSON = {'hey, this is some fake':'json'}
7
8@pytest.mark.asyncio
9async def test() -> None:
10    with patch(
11        "playground.foo.Foo.Bar.fetch"
12    ) as patched_fetch:
13        mock_object = AsyncMock()
14        mock_object.status_code = 200
15        mock_object.json = lambda:  FAKE_JSON
16
17        patched_fetch.return_value = mock_object
18
19        result = await Foo.Bar.fetch('https://api.example.com')
20        assert result.status_code == 200
21        assert result.json() == FAKE_JSON
22

Let's go step by step over what's happening here:

The patch function allows the use of context managers, so as long as you're in the context perse, whatever object you patched stays patched, you must provide the full import path of the function or else it will not work. So if you're patching a function from library X you might do:

@pytest.mark.asyncio
async def test() -> None:
    with patch(
        "library_x.foo.class.method"
    ) as patched_method:

With it, you can alter the return value of methods, hence the:

patched_fetch.return_value = mock_object

But what's the deal with the mock object? the return value is a mocked version of the response, it's saying: here I have a Response object that I'm trying to patch with a mocked version of it, that behaves like I need it to, we need a basic mock object that we can build upon, that's asynchronous in nature.

mock_object = AsyncMock()

You can set any properties and any methods on this object as you see fit. If you're setting properties you can directly set them as such:

mock_object.my_property = 'anything'

But if it's a Callable that you're mocking that's not exactly a @property, you must provide a function for it, you can use a lambda if the function is simple, or define a normal function and assign it, if it's complex. In the example above I mocked the .json() method of the Response class with a lambda function that takes nothing and returns a dictionary holding the data I want to receive, which is FAKE_JSON. Notice I didn't do the same for the .status_code as it is a property method so you can set it directly like you would set a normal attribute.

With that out of the way, we can then test if it actually works.

result = await Foo.Bar.fetch('https://api.fake-example.com')
assert result.status_code == 200
assert result.json() == {'patched':'json'}

And it does!

============================= test session starts ==============================
collecting ... collected 1 item

test.py::test PASSED                                                     [100%]

============================== 1 passed in 0.13s ===============================