A Practical Guide To Async Testing With Pytest-Asyncio

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

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

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.

Async — Use Cases

Non-Blocking I/O (Network)

I/O operations, for example

• File images 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).

Calling External APIs or Services

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

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

Machine Learning / Model Training

During model training, performing other critical or non-critical tasks like file images, 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.

Data ETL or ELT Operations

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

Testing Async Functions (Pytest-Asyncio)

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.

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

Async Fixtures

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

# Async fixtures  
@pytest_asyncio.fixture  
async def loaded_data():  
    await asyncio.sleep(1)  # Simulate loading data asynchronously  
    return {"key": "value"}

Async Mocking Using AsyncMock

# 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!"