A Practical Guide To Async Testing With Pytest-Asyncio

Posted on Edited on In Word count in article: 2.2k Reading time ≈ 8 mins.

In modern computing, it’s a common belief that computers execute tasks in parallel, effortlessly juggling multiple operations at once.

True parallelism is largely the domain of supercomputers and quantum computing.

But what about the everyday applications that form the backbone of our digital lives?

Asynchronous, or async, allows a program to efficiently execute its tasks (executing other tasks one while is waiting). This is also called concurrency.

When it comes to testing async code, things can get a bit tricky.

How can you ensure your async code behaves as expected? How do you test the interaction of different async tasks and functions without getting tangled in the complexities of event loops and await statements?

This is where Pytest Asyncio steps in.

pytest-asyncio simplifies handling event loops, managing async fixtures, and bridges the gap between async programming and thorough testing.

In this article we’ll dive deep into how to use the pytest-asyncio plugin to test Pytest Async functions, write async fixtures and use async mocking to test external services with a simple example.

By the end, you’ll come out with a better understanding of how to test your async Python programs.

Let’s get started then?

Link To GitHub Repo

What Are Async Functions — Quick Summary

The internet is flooded with information on async, so we won’t go into too much detail here.

This article from the FastAPI documentation gives a clear overview of concurrency and async with cute graphics.

The important thing to know is that Async is NOT parallel execution, rather the efficient allocation of compute resources.

We’ll look at real cases below but just know that while a task is pending or sleeping, the system executes another task.

For example tasks like file upload, writing to a database, I/O, or calling external APIs take time due to network and or server overhead.

The system makes an intelligent decision to execute another task in the meanwhile, thus saving time and resources.

Async — Use Cases

Now that you’ve got a basic understanding of async, perhaps you’re wondering where is this used in practice?

Good question, a variety of use cases.

Non-Blocking I/O (Network)

I/O operations, for example

  • File uploads or downloads
  • R/W to local disk or cloud storage
  • R/W to a database

All these tasks have network or server overhead (i.e. their speed of execution depends on the speed of the network or server).

So while the program is waiting for a response it can initiate or fully perform another task (instead of lying idle).

Calling External APIs or Services

Like the previous use case, calling external services may have an overhead due to server response speed and network.

External API calls take longer and waste valuable execution time so it’s good design practice that your program does something useful rather than just wait around.

Audio / Image Processing and Computer Vision

If your program requires image, audio processing or computer vision to detect entities or transcribe words, this can often take time.

It’s good to write optimised async code that can perform other tasks in the meanwhile.

Machine Learning / Model Training

Machine learning and model training takes time.

Highly performant models are often trained on millions if not billions of parameters.

For example, GPT-4 was trained on more than 1.7 trillion parameters.

During model training, performing other critical or non-critical tasks like file uploads, status changes, polling, R/W to db etc. may be possible.

Server Responses And Chatbots

Server responses and chatbots may need to respond to several requests at the same time.

Most of the time these are mutually exclusive.

Async code can be incredibly useful in these cases and reduce response latency.

This example in the documentation explains how to use it when building APIs in FastAPI.

Data ETL or ELT Operations

Data syncing or ELT (Extract, Load, Transform) operations can be time-consuming and can often be executed asynchronously.

Practical Example

Now that you have a solid base understanding of async and it’s use cass, let’s look at some examples of how to implement and test async code in Python using Pytest.

Here’s a link to the repo containing the examples.

Local Set Up

Some basics of Python and Pytest would be helpful:

  • Python (3.12)
  • Pytest

Create a virtual environment and install the required packages using the following command:

1
pip install -r requirements.txt

Feel free to use any package manager you wish - pipenv, poetry, conda etc.

Example Code

Let’s look at a simple example how to write and test a Python Async function.

async_examples/basic.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import asyncio


async def async_add(a, b):
    print("Starting async_add")
    await asyncio.sleep(5)  # Simulate an async operation
    print("Result From async_add", a + b)
    return a + b


async def async_sub(a, b):
    print("Starting async_sub")
    print("Result From async_sub", a - b)
    return a - b


async def main():
    # Schedule both async_add and async_sub to run concurrently.
    res_add, res_b = await asyncio.gather(
        async_add(1, 2),  # Pass the first function call
        async_sub(1, 2),  # Pass the second function call
    )


if __name__ == "__main__":
    asyncio.run(main())

This example is a simple async function that adds and subtracts two numbers.

Let’s run it

1
python async_examples/basic.py

pytest-asyncio-0

You can see how the system executes the async functions concurrently. It starts with the async_add function, then while it’s sleeping it executes the async_sub function and finally finishes the async_add function.

Note you need to use the async keyword before defining the function and await it when calling it.

Now how can you write unit tests for these async functions?

Testing Async Functions (Pytest-Asyncio)

Unit testing async functions is similar to regular functions and this is made easy using the pytest-asyncio plugin.

It takes care of the event loop while providing support for coroutines as test functions. It also allows you to await code inside tests.

You can write your tests as you would normally and use the @pytest.mark.asyncio marker to tell Pytest that the function is async along with the async keyword before defining your test functions.

tests/test_basic.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import pytest
import asyncio
import pytest_asyncio
from async_examples.basic import async_add, async_sub


# Basic Async test function
@pytest.mark.asyncio
async def test_async_add():
    result = await async_add(1, 2)
    assert result == 3


@pytest.mark.asyncio
async def test_async_sub():
    result = await async_sub(1, 2)
    assert result == -1

Running it

1
pytest tests/unit/test_basic.py

pytest-asyncio-1

This simple example shows how to write unit tests for async functions using Pytest-Asyncio.

Now let’s move up a notch to see how we can use Fixtures and Mocking to test async functions.

Async Fixtures

If you’ve been reading this website for a while, you know how important fixtures are in Pytest.

Fixtures help you to set up and tear down resources before and after tests, offering you immense flexibility via the scope parameter.

Fixtures are normally defined as below and can easily be used in test functions.

1
2
3
4
5
6
7
8
9
10
11
import pytest

@pytest.fixture
def my_fixture():
    return "Hello World"

def test_my_fixture1(my_fixture):
    assert my_fixture == "Hello World"

def test_my_fixture2(my_fixture):
    assert my_fixture == "Hello World"

So how do you write async fixtures?

Very similar. You use the @pytest_asyncio.fixture() marker and define the fixture as you would normally.

You can then access the fixture in your test functions as you would normally. This helps you avoid having to await the fixture manually.

tests/test_basic.py

1
2
3
4
5
6
7
8
9
10
11
# Async fixtures
@pytest_asyncio.fixture
async def loaded_data():
    await asyncio.sleep(1)  # Simulate loading data asynchronously
    return {"key": "value"}


# Async test functions with fixtures
@pytest.mark.asyncio
async def test_fetch_data(loaded_data):
    assert loaded_data["key"] == "value"

Now let’s use all what we’ve learnt about Pytest async to test a real-world example using async fixtures and mocking.

Real-World Example (Cat Fact API)

The example we’ll be using is a Cat Fact API that returns a random cat fact.

async_examples/cat_fact.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import logging
import aiohttp
import asyncio


class CatFact:
    def __init__(self):
        self.base_url = "https://meowfacts.herokuapp.com/"

    async def get_cat_fact(self):
        """
        Asynchronously get a Cat Fact from Rest API and return a dict with status and fact.
        """
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(self.base_url) as response:
                    if response.status in (200, 201):
                        json_response = await response.json()
                        return {"status": response.status, "result": json_response}
                    else:
                        return {
                            "status": response.status,
                            "error": "Cat Fact Not Available",
                        }
        except aiohttp.ClientError as err:
            logging.error(f"Client error: {err}")
            return {"status": 500, "error": "Failed to fetch Cat Fact"}
        except Exception as e:
            logging.error(f"Unexpected error: {e}")
            return {"status": 500, "error": "Failed to fetch Cat Fact"}


async def main():
    cat_fact = CatFact()
    result = await cat_fact.get_cat_fact()
    print(result)


if __name__ == "__main__":
    asyncio.run(main())

The above code uses the aiohttp library to make an async GET request to the Cat Fact API and return a random cat fact. It then returns a Dict object with the status and fact.

Let’s run this to make sure it works.

1
python async_examples/cat_fact.py

pytest-asyncio-2

Let’s go ahead and write basic tests.

tests/test_async_cat_fact.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import pytest
import pytest_asyncio
from asyncmock import AsyncMock
from async_examples.cat_fact import CatFact


@pytest_asyncio.fixture
async def cat_fact():
    return CatFact()


# Test with Real API call (Not Recommended)
@pytest.mark.asyncio
async def test_get_cat_fact(cat_fact):
    result = await cat_fact.get_cat_fact()
    print(result)
    assert result["status"] == 200
    assert "data" in result["result"]

In the above code we have

  • Defined an async fixture cat_fact that returns an instance of the CatFact class.
  • Written a test function test_get_cat_fact that calls the get_cat_fact method of the CatFact class and asserts the response.

Running the test

1
pytest tests/unit/test_async_cat_fact.py

pytest-asyncio-3

If you notice my subtle comment in the test function, it’s not recommended to make real API calls in your tests.

This is because it’s slow, unreliable and can cause your tests to fail if the API is down or the network is slow. Although you may have a contract with the API provider, it’s not uncommon for breaking changes to occur.

So how do you test this without making real API calls?

You can mock the get_cat_fact method of the CatFact class using the AsyncMock class from the asyncmock library.

Async Mocking Using AsyncMock

Mocking is a powerful technique in testing that allows you to replace real objects with a double (or fake) object that you can control.

In this case, we’ve decided to mock the get_cat_fact method of the CatFact class to abstract away it’s implementation detail, but you can also mock the requests.get method of the requests library.

tests/test_async_cat_fact.py

1
2
3
4
5
6
7
8
9
10
11
12
13
# Test with Mocked API call (Recommended when interacting with external services)
@pytest.mark.asyncio
async def test_get_cat_fact_mocked(mocker):
    # Mock the get_cat_fact method of CatFact
    mock_response = {"status": 200, "result": {"data": "Cats are awesome!"}}
    mocker.patch.object(CatFact, "get_cat_fact", AsyncMock(return_value=mock_response))

    cat_fact_instance = CatFact()
    result = await cat_fact_instance.get_cat_fact()

    assert result["status"] == 200
    assert "data" in result["result"]
    assert result["result"]["data"] == "Cats are awesome!"

Let’s review and break down what’s going on in here.

  • We defined a new test function test_get_cat_fact_mocked that takes a mocker fixture as an argument. This fixture is provided by the pytest-mock plugin.
  • We used the mocker.patch.object method to mock the get_cat_fact method of the CatFact class to return a pre-defined response. This uses the AsyncMock class from the asyncmock library.
  • We then created an instance of the CatFact class and called the get_cat_fact method to get the response.

Calling get_cat_fact will now return the mocked response instead of making a real API call.

Running the test

1
pytest tests/unit/test_async_cat_fact.py

pytest-asyncio-4

By mocking, you can test your code without making real API calls, making your tests faster, more reliable, and independent of external services.

Async Libraries

Although not related to the example code, I wanted to quickly touch on some of the Python libraries currently built on Asyncio and are worth looking at.

FastAPI, Django, AsyncSSH are some of the most popular and this list is growing.

This repo includes a fairly comprehensive list of popular libraries and frameworks built on Asyncio.

Conclusion

This has been an interesting article.

Wrapping your head around async is not easy and presents some challenges — especially with the syntax, boilerplate and understanding event loops and await.

You need to remember that async is an efficient implementation of sequential execution and is powerful in the sense that your code can execute operations while other parts are in a pending or sleeping state.

In this article, we looked at some interesting use cases for asynchronous code including practical examples of how to write and test Python async code.

Lastly, we looked at unit testing a class that calls the CatFact API to get random cat facts using async fixtures and mocking.

I hope you found this article useful and that you now have a better understanding of how to write and test async code in Python, as this is the next level when you want to write high-performance code.

If you have any ideas for improvement or like me to cover any topics please comment below or send me a message via Twitter, GitHub or Email.

Till the next time… Cheers!

Additional Reading

Example Code
pytest-asyncio Documentation
How Pytest Fixtures Can Help You Write More Readable And Efficient Tests
What is Setup and Teardown in Pytest? (Importance of a Clean Test Environment)
13 Proven Ways To Improve Test Runtime With Pytest
Python Testing 101 (How To Decide What To Test)
Concurrency and async
Awesome asyncio
async test patterns for Pytest