Python Async Testing Patterns — Core Concepts

Setting Up Async Tests

The standard approach uses pytest-asyncio:

pip install pytest-asyncio

import pytest
import asyncio

@pytest.mark.asyncio
async def test_basic_async():
    result = await asyncio.sleep(0, result=42)
    assert result == 42

Configure it in pyproject.toml to auto-detect async tests:

[tool.pytest.ini_options]
asyncio_mode = "auto"

With auto mode, you don’t need @pytest.mark.asyncio on every test.

Async Fixtures

Fixtures that set up and tear down async resources:

@pytest.fixture
async def db_connection():
    conn = await asyncpg.connect("postgresql://localhost/testdb")
    yield conn
    await conn.close()

@pytest.fixture
async def populated_db(db_connection):
    await db_connection.execute(
        "INSERT INTO users (name) VALUES ($1)", "Alice"
    )
    yield db_connection
    await db_connection.execute("DELETE FROM users")

The yield pattern works the same as sync fixtures — setup before yield, teardown after.

Mocking Async Functions

Use AsyncMock from unittest.mock:

from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_fetch_user():
    mock_db = AsyncMock()
    mock_db.fetch_one.return_value = {"name": "Alice", "id": 42}

    user = await get_user(42, db=mock_db)
    assert user.name == "Alice"
    mock_db.fetch_one.assert_awaited_once_with(
        "SELECT * FROM users WHERE id = $1", 42
    )

AsyncMock automatically returns a coroutine, so await mock_fn() works without special setup.

Patching Async Methods

@pytest.mark.asyncio
async def test_with_patch():
    with patch("myapp.client.fetch", new_callable=AsyncMock) as mock:
        mock.return_value = {"status": "ok"}
        result = await my_function()
        assert result["status"] == "ok"

Testing Exception Handling

@pytest.mark.asyncio
async def test_timeout_handling():
    with pytest.raises(TimeoutError):
        async with asyncio.timeout(0.01):
            await asyncio.sleep(10)

@pytest.mark.asyncio
async def test_cancelled_cleanup():
    resource = MockResource()
    task = asyncio.create_task(use_resource(resource))
    await asyncio.sleep(0.01)
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    assert resource.closed  # Verify cleanup happened

Testing TaskGroups

@pytest.mark.asyncio
async def test_taskgroup_error_propagation():
    results = []

    async def good_task():
        results.append("ok")

    async def bad_task():
        raise ValueError("boom")

    with pytest.raises(ExceptionGroup) as exc_info:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(good_task())
            tg.create_task(bad_task())

    assert len(exc_info.value.exceptions) == 1
    assert isinstance(exc_info.value.exceptions[0], ValueError)

Avoiding Timing-Dependent Tests

Flaky async tests often depend on timing. Instead of sleep(), use synchronization primitives:

# BAD — timing dependent
@pytest.mark.asyncio
async def test_producer_consumer():
    queue = asyncio.Queue()
    asyncio.create_task(producer(queue))
    await asyncio.sleep(0.5)  # Hope producer is done?
    assert not queue.empty()

# GOOD — event-driven
@pytest.mark.asyncio
async def test_producer_consumer():
    queue = asyncio.Queue()
    done = asyncio.Event()

    async def tracked_producer(q):
        await producer(q)
        done.set()

    asyncio.create_task(tracked_producer(queue))
    await asyncio.wait_for(done.wait(), timeout=5.0)
    assert not queue.empty()

Common Misconception: “Async Tests Run in Parallel”

By default, pytest-asyncio runs each test in its own event loop, sequentially. Tests don’t run concurrently with each other. If you need parallel test execution, use pytest-xdist — but each worker still runs tests sequentially within its event loop.

Testing Patterns Summary

What You’re Testing	Pattern
Simple async function	await + assert
Function with I/O dependency	AsyncMock
Cancellation handling	task.cancel() + verify cleanup
Timeout behavior	asyncio.timeout() in test
Multiple concurrent tasks	TaskGroup + event synchronization
Background task side effects	Event or Queue to signal completion

One thing to remember: Replace await asyncio.sleep() in tests with asyncio.Event and asyncio.wait_for() — event-driven synchronization eliminates the timing-dependent flakiness that plagues async test suites.