3 Ways To Easily Test Your Python Async Functions With Pytest Asyncio

Posted on Edited on In Word count in article: 1.8k Reading time ≈ 7 mins.

We think that computers can do tasks “in parallel”. As much as we’d like, that’s not actually the case.

The modern computer is actually sequential, although at a very high clock speed unperceivable to humans.

Perhaps supercomputers or quantum computers are capable of true synchronous execution.

Almost all code is executed sequentially. However, there are optimisations we can do to efficiently use computing resources (CPU/RAM).

Async stands for Asynchronous and refers to the efficient allocation of resources.

For example, while one task is pending the system executes another task and then comes back to the pending one.

While it’s easy to write async functions, how do you Unit Test them?

Pytest Asyncio is a convenient framework to test your async Python functions using Pytest.

In this article we’ll dive deeper into the use cases of async functions and how to Unit Test them (using fixtures, markers and mocking) with a simple example.

By the end, you’ll come out with a better understanding of how to write async Python programs and unit tests.

  • What Are Async Functions — Quick Summary
  • Async — Use Cases
    • Non-Blocking I/O (Network)
    • Calling External Servers and APIs
    • Audio / Image Processing and Computer Vision
    • Machine Learning / Model Training
    • Server Responses And Chatbots
  • How To Test Async Functions Using Pytest-Asyncio — Example
    • Quick Repo/Code Explanation
      • Basic Async Functions
      • Cat Fact Class
      • Unit Tests
    • Testing Async Functions
      • 1. Using Async Markers
      • 2. Using Async Fixtures
      • 3. Using Async Mocking
  • Async Libraries
  • Conclusion
  • Additional Reading

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 using 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 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 Servers and APIs

Like the previous use case, calling external servers 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 else in the meanwhile when possible.

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, ChatGPT-3 was trained on 175 billion 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.

How To Test Async Functions Using Pytest-Asyncio — Example

Now that you have a solid base understanding of async, 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.

Quick Repo/Code Explanation

The async_application folder contains 2 files — core.py and cat_fact.py .

Basic Async Functions

The core.py file contains 2 functions

  1. Upload a file to a random server
  2. Save a file to local disk

These 2 operations are a simulation of 2 mutually exclusive tasks that can be handled asynchronously.

core.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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
# Spec 1. Upload a large file via HTTP Request 2. Write to Disk
import json
import logging
import uuid
from datetime import datetime
import requests
import asyncio

# Set Logging
logging.basicConfig(level=logging.INFO)


async def upload_file(file_name: str = None) -> dict:
    """
    Function to call the API via the Requests Library
    :param file_name: Filename, Str
    :return: Response. Type - JSON Formatted String
    """
    try:
        files = {
            "file": open(file_name, "rb"),
        }
        endpoint = "https://api.anonfiles.com/upload"
        logging.info(f"Uploading File `{file_name}` to Server...")
        response = requests.post(endpoint, files=files)
        await asyncio.sleep(3)  # TODO REMOVE
        response_json = response.json()
        logging.debug(f"Response from API: {response_json}")
        if response.status_code in (200, 201) and response_json["status"] is True:
            logging.info(f"File `{file_name}` successfully uploaded to Server!!!")
            return response_json
        response.raise_for_status()
    except requests.exceptions.HTTPError as errh:
        logging.error(errh)
    except requests.exceptions.ConnectionError as errc:
        logging.error(errc)
    except requests.exceptions.Timeout as errt:
        logging.error(errt)
    except requests.exceptions.RequestException as err:
        logging.error(err)


async def save_to_disk() -> bool:
    """
    Function to save a dict to disk as a JSON file
    :return: Bool
    """
    sample = {"Value": str(uuid.uuid4()), "Time": str(datetime.utcnow())}
    with open(
        f"./tests/unit/tmp_data/{str(uuid.uuid4())}.json", "w", encoding="utf-8"
    ) as f:
        json.dump(sample, f, ensure_ascii=False, indent=4)
        logging.info("Data Saved to Disk!!!")
        status = True
    return status
Cat Fact Class

The cat_fact.py file contains a CatFact class with 1 method get_cat_fact() which performs a REST API GET request to get a random Cat Fact.

We used this example in our article on Pytest Monkeypatch and Mocking.

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
import json
import logging
import requests

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

    async def get_cat_fact(self) -> tuple[int, dict] | str:
        """
        Function to get a Cat Fact from Rest API
        :return: JSON Encoded Response String
        """
        try:
            response = requests.get(self.base_url)
            if response.status_code in (200, 201):
                return response.status_code, response.json()
            else:
                return json.dumps({"ERROR": "Cat Fact Not Available"})
        except requests.exceptions.HTTPError as errh:
            logging.error(errh)
        except requests.exceptions.ConnectionError as errc:
            logging.error(errc)
        except requests.exceptions.Timeout as errt:
            logging.error(errt)
        except requests.exceptions.RequestException as err:
            logging.error(err)
Unit Tests

The Unit Tests can be found under the tests/unit/.. path and contain 2 files.

The first file test_async_application.py contains unit tests for the async application (upload photo and save to disk).

The second file test_async_cat_fact.py contains tests for the CatFact application.

Testing Async Functions

1. Using Async Markers

To test async functions it’s important to use the @pytest.mark.asyncio marker.

This tells Pytest the functions in the test are async and executes them accordingly. Without it, Pytest won’t await them and they’ll will pass regardless.

Note- To test the first unit test test_upload_file please save a random image _sample.jpeg_ on your home directory.

You can see we use the async keyword before defining the test function (same as the course code).

test_async_application.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
41
42
from pathlib import Path
import pytest
import asyncio
from async_application.core import upload_file, save_to_disk

home = str(Path.home())


@pytest.mark.asyncio
async def test_upload_file() -> None:
    """
    Function to test the Upload File Method
    :return: None
    """
    result_upload = await upload_file(file_name=f"{home}/sample.jpeg")
    assert result_upload["status"] is True


@pytest.mark.asyncio
async def test_save_to_disk() -> None:
    """
    Function to test the Save to Disk Method
    :return: None
    """
    result_save = await save_to_disk()
    assert result_save is True


@pytest.mark.asyncio
async def test_main() -> None:
    """
    Function to test the above 2 methods
    :return: None
    """
    task1 = asyncio.create_task(upload_file(file_name=f"{home}/sample.jpeg"))
    task2 = asyncio.create_task(save_to_disk())

    value_task1 = await task1
    value_task2 = await task2

    assert value_task1["status"] is True
    assert value_task2 is True

In this file we have 3 test functions — 1) Test upload file 2) Test save to disk and 3) main function that performs both the above tasks asynchronously.

pytest-asyncio-1

If you pay attention to the logs of the 3rd Unit Test test_main we can see that the system began the file upload, then while the upload was in progress, saved a file to disk and then completed the file upload.

This is a classic example of Async.

2. Using Async Fixtures

How do we use Fixtures for asynchronous functions?

Very similar to how we write them otherwise, except with a @pytest_asyncio.fixture() marker.

conftest.py

1
2
3
4
5
6
7
8
import pytest_asyncio
from async_application.cat_fact import CatFact


@pytest_asyncio.fixture()
async def async_app_client():
    async_cat_fact = CatFact()
    yield async_cat_fact

Moving onto the 2nd unit test file test_async_cat_fact , we have a unit tests that uses this fixture.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import pytest
import asyncio
from asyncmock import AsyncMock


@pytest.mark.asyncio
async def test_get_cat_fact_fixture(async_app_client) -> None:
    """
    Test for get_cat_fact method using Async Fixture
    :param async_app_client:
    :return: None
    """
    task1 = asyncio.create_task(async_app_client.get_cat_fact())

    value_task1 = await task1
    print(value_task1)

3. Using Async Mocking

Given the CatFact API returns a different cat fact on every get request, unit testing it (assertions) would be difficult as we can’t predict the response.

This is where Mocking or Monkeypatching shines.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@pytest.fixture
def mock_thing() -> AsyncMock:
    """
    Async Mock Fixture
    :return:
    """
    mock_thing = AsyncMock()
    mock_thing.CatFact.get_cat_fact = AsyncMock(
        return_value="Mother cats " "teach their kittens " "to use the litter box."
    )
    return mock_thing


@pytest.mark.asyncio
async def test_get_cat_fact_mock(mock_thing) -> None:
    """
    Test for get_cat_fact method using Async Mocking
    :param mock_thing: Mock fixture
    :return: None
    """
    result = await mock_thing.CatFact.get_cat_fact()
    assert result == "Mother cats teach their kittens to use the litter box."

On lin 14, we define the mock fixture object mock_thing using the AsyncMoc class, specifying its’ return response.

We then await it as the method get_cat_fact() is called which then produces the above response. We can assert it as expected.

pytest-asyncio-2

Async Libraries

Although not related to the example, I wanted to include a small section on which packages are currently built on Asyncio and are worth looking at.

FastAPI, Django, AsyncSSH are some of the most popular.

This repo includes a fairly comprehensive list.

Conclusion

This has been a challenging article.

Wrapping your head around async is not easy and presents some challenges — especially with the syntax and boilerplate code.

You just need to remember that async is just an efficient implementation of sequential execution.

In this article, we looked at some interesting real use cases for asynchronous code.

Next, we looked at how to implement unit testing using the Pytest Asyncio library and async markers.

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

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

https://fastapi.tiangolo.com/async/

https://testdriven.io/blog/fastapi-crud/

https://github.com/timofurrer/awesome-asyncio

async test patterns for Pytest

https://superfastpython.com/python-asyncio/

https://www.youtube.com/watch?v=t5Bo1Je9EmE