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

# Async Testing

## Table of Contents
- [Setup](#setup)
- [Basic Async Tests](#basic-async-tests)
- [Async Fixtures](#async-fixtures)
- [Testing Async Generators](#testing-async-generators)
- [Timeouts](#timeouts)

## Setup

### Installation

```bash
pip install pytest-asyncio
```

### Configuration

```toml
# 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

```ini
[pytest]
asyncio_mode = auto
```

## Basic Async Tests

### Simple Async Test

```python
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

```python
@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

```python
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

```python
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

```python
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

```python
@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

```python
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

```python
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

```python
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

```python
@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

```python
@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

```python
@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

```python
@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

```python
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()
```