lab_ai/.agents/skills/pytest/references/async-testing.md

Async Testing

Table of Contents

• Setup
• Basic Async Tests
• Async Fixtures
• Testing Async Generators
• Timeouts

Setup

Installation

pip install pytest-asyncio

Configuration

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"  # Automatically handle async tests
# OR
asyncio_mode = "strict"  # Require explicit @pytest.mark.asyncio

With pytest.ini

[pytest]
asyncio_mode = auto

Basic Async Tests

Simple Async Test

import pytest

# With asyncio_mode = "auto", decorator is optional
async def test_async_function():
    result = await async_operation()
    assert result == "success"

# With asyncio_mode = "strict", decorator is required
@pytest.mark.asyncio
async def test_async_with_marker():
    result = await async_operation()
    assert result == "success"

Async Test Class

@pytest.mark.asyncio
class TestAsyncOperations:
    async def test_fetch_data(self):
        data = await fetch_data()
        assert data is not None

    async def test_process_data(self):
        result = await process_data({"key": "value"})
        assert result["processed"] is True

Testing Async Exceptions

import pytest

@pytest.mark.asyncio
async def test_async_exception():
    with pytest.raises(ValueError):
        await async_function_that_raises()

@pytest.mark.asyncio
async def test_async_exception_message():
    with pytest.raises(ValueError, match="Invalid input"):
        await validate_input("")

Async Fixtures

Basic Async Fixture

import pytest

@pytest.fixture
async def async_client():
    """Async fixture for HTTP client."""
    async with httpx.AsyncClient() as client:
        yield client

@pytest.mark.asyncio
async def test_api_call(async_client):
    response = await async_client.get("https://api.example.com/data")
    assert response.status_code == 200

Async Database Fixture

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

@pytest.fixture(scope="session")
def event_loop():
    """Create event loop for session-scoped async fixtures."""
    import asyncio
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session")
async def engine():
    """Session-scoped async engine."""
    engine = create_async_engine("postgresql+asyncpg://...")
    yield engine
    await engine.dispose()

@pytest.fixture
async def db_session(engine):
    """Function-scoped database session with rollback."""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with AsyncSession(engine) as session:
        yield session
        await session.rollback()

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

Fixture with Async Setup/Teardown

@pytest.fixture
async def resource():
    # Async setup
    resource = await create_resource()
    await resource.initialize()

    yield resource

    # Async teardown
    await resource.cleanup()
    await resource.close()

Testing Async Generators

Async Generator Function

async def async_data_generator():
    for i in range(5):
        await asyncio.sleep(0.1)
        yield i

@pytest.mark.asyncio
async def test_async_generator():
    results = []
    async for item in async_data_generator():
        results.append(item)

    assert results == [0, 1, 2, 3, 4]

Async Context Manager

from contextlib import asynccontextmanager

@asynccontextmanager
async def async_resource():
    resource = await create_resource()
    try:
        yield resource
    finally:
        await resource.close()

@pytest.mark.asyncio
async def test_async_context_manager():
    async with async_resource() as resource:
        result = await resource.operation()
        assert result is not None

Timeouts

Test Timeout

import asyncio
import pytest

@pytest.mark.asyncio
@pytest.mark.timeout(5)  # 5 second timeout
async def test_slow_operation():
    result = await potentially_slow_operation()
    assert result is not None

Using asyncio.wait_for

@pytest.mark.asyncio
async def test_with_timeout():
    try:
        result = await asyncio.wait_for(
            slow_operation(),
            timeout=2.0
        )
        assert result is not None
    except asyncio.TimeoutError:
        pytest.fail("Operation timed out")

Testing Timeout Behavior

@pytest.mark.asyncio
async def test_timeout_is_raised():
    """Verify function properly times out."""
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(
            never_completes(),
            timeout=0.1
        )

Concurrent Tests

Testing Concurrent Operations

@pytest.mark.asyncio
async def test_concurrent_requests():
    async with httpx.AsyncClient() as client:
        # Run 10 requests concurrently
        tasks = [
            client.get(f"https://api.example.com/items/{i}")
            for i in range(10)
        ]
        responses = await asyncio.gather(*tasks)

        assert all(r.status_code == 200 for r in responses)

Testing Race Conditions

@pytest.mark.asyncio
async def test_counter_thread_safety():
    counter = AsyncCounter()

    async def increment():
        for _ in range(100):
            await counter.increment()

    # Run 10 concurrent incrementers
    await asyncio.gather(*[increment() for _ in range(10)])

    assert counter.value == 1000

Event Loop Configuration

Custom Event Loop

import pytest
import asyncio

@pytest.fixture(scope="session")
def event_loop_policy():
    """Use uvloop for faster async tests."""
    import uvloop
    return uvloop.EventLoopPolicy()

@pytest.fixture(scope="session")
def event_loop(event_loop_policy):
    asyncio.set_event_loop_policy(event_loop_policy)
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()