test-setup-async

What this skill does

# Setup Async Testing

## Purpose

Provide correct patterns for testing async functions in Python using pytest-asyncio, AsyncMock, and async fixtures. Ensures tests properly handle async context managers, side effects, and cleanup.

## Quick Start

**Most common use case:**
```python
# Testing an async function with mocked dependencies
import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_my_async_function():
    mock_service = AsyncMock()
    mock_service.fetch_data.return_value = {"result": "success"}

    result = await my_async_function(mock_service)

    assert result == {"result": "success"}
    mock_service.fetch_data.assert_awaited_once()
```

## Table of Contents

1. [When to Use This Skill](#when-to-use-this-skill)
2. [What This Skill Does](#what-this-skill-does)
3. [Instructions](#instructions)
4. [Usage Examples](#usage-examples)
5. [Expected Outcomes](#expected-outcomes)
6. [Requirements](#requirements)
7. [Troubleshooting](#troubleshooting)
8. [Red Flags to Avoid](#red-flags-to-avoid)
9. [Automation Scripts](#automation-scripts)

---

## When to Use This Skill

Use this skill when:
- **Testing async functions** - Functions using `async def` and `await`
- **Creating async fixtures** - Setup/teardown with async operations
- **Mocking async services** - Database connections, API clients, external services
- **Handling async context managers** - `async with` statements
- **Testing async generators** - `async for` patterns
- **Debugging async test failures** - "RuntimeWarning: coroutine was never awaited"

**Trigger phrases:**
- "Test async function"
- "Mock async method"
- "Create async fixture"
- "AsyncMock configuration"
- "pytest-asyncio setup"

---

## What This Skill Does

This skill provides patterns for:

1. **Async test configuration** - pytest-asyncio setup in pyproject.toml
2. **Async fixture creation** - Using @pytest_asyncio.fixture with AsyncGenerator
3. **AsyncMock usage** - Mocking async methods with return_value and side_effect
4. **Async context managers** - Mocking `async with` statements (__aenter__, __aexit__)
5. **Async assertions** - assert_awaited_once(), assert_awaited_once_with()
6. **Error handling** - Testing exceptions in async code

See Instructions section below for detailed step-by-step patterns.

---

## Instructions

### Step 1: Install Dependencies
```bash
uv pip install pytest-asyncio
```

Add to `pyproject.toml`:
```toml
[tool.pytest.ini_options]
asyncio_mode = "auto"  # Auto-detect async tests
```

### Step 2: Choose Fixture Type

**Async Fixture (for setup/teardown):**
```python
import pytest_asyncio
from collections.abc import AsyncGenerator

@pytest_asyncio.fixture
async def database_connection() -> AsyncGenerator[Connection, None]:
    """Create database connection with cleanup."""
    conn = await create_connection()
    yield conn
    await conn.close()
```

**Sync Fixture (for mocks):**
```python
@pytest.fixture
def mock_service():
    """Create mock service (non-async fixture for mocks)."""
    from unittest.mock import AsyncMock
    mock = AsyncMock()
    mock.fetch_data.return_value = {"data": "test"}
    return mock
```

### Step 3: Mock Async Methods

**Simple return value:**
```python
mock_service = AsyncMock()
mock_service.process.return_value = "result"
```

**Side effect (exception):**
```python
mock_service.process.side_effect = Exception("Connection failed")
```

**Side effect (sequence):**
```python
mock_service.fetch.side_effect = [
    {"batch": 1},
    {"batch": 2},
    {"batch": 3}
]
```

**Side effect (custom function):**
```python
async def custom_behavior(arg):
    if arg == "fail":
        raise ValueError("Invalid")
    return f"processed_{arg}"

mock_service.process.side_effect = custom_behavior
```

### Step 4: Test Async Context Managers

**Mock async context manager:**
```python
from unittest.mock import AsyncMock

@pytest.fixture
def mock_driver():
    """Mock Neo4j driver with async context manager."""
    driver = MagicMock()  # Driver itself is sync

    # Create async session mock
    mock_session = AsyncMock()
    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
    mock_session.__aexit__ = AsyncMock(return_value=None)

    # Configure session behavior
    mock_result = AsyncMock()
    mock_result.data = AsyncMock(return_value=[{"node": "data"}])
    mock_session.run = AsyncMock(return_value=mock_result)

    # Driver returns session
    driver.session = MagicMock(return_value=mock_session)

    return driver
```

**Test usage:**
```python
@pytest.mark.asyncio
async def test_with_session(mock_driver):
    async with mock_driver.session() as session:
        result = await session.run("MATCH (n) RETURN n")
        data = await result.data()

    assert data == [{"node": "data"}]
    mock_driver.session.assert_called_once()
```

### Step 5: Verify Async Calls

**Assert awaited:**
```python
mock_service.fetch_data.assert_awaited_once()
mock_service.fetch_data.assert_awaited_once_with(arg="value")
mock_service.fetch_data.assert_awaited()  # At least once
```

**Assert call count:**
```python
assert mock_service.fetch_data.await_count == 3
```

**Assert not called:**
```python
mock_service.error_handler.assert_not_awaited()
```

## Usage Examples

### Example 1: Async Fixture with Cleanup
```python
import pytest_asyncio
from collections.abc import AsyncGenerator
from neo4j import AsyncDriver, AsyncGraphDatabase

@pytest_asyncio.fixture(scope="function")
async def neo4j_driver() -> AsyncGenerator[AsyncDriver, None]:
    """Create Neo4j driver with cleanup."""
    driver = AsyncGraphDatabase.driver(
        "bolt://localhost:7687",
        auth=("neo4j", "password")
    )

    # Setup: Clear test data
    async with driver.session() as session:
        await session.run("MATCH (n:TestNode) DETACH DELETE n")

    yield driver

    # Cleanup: Close driver
    await driver.close()
```

### Example 2: AsyncMock with Side Effect Sequence
```python
import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_batch_processing():
    mock_api = AsyncMock()

    # First call succeeds, second fails, third succeeds
    mock_api.fetch_batch.side_effect = [
        {"batch": 1, "items": [1, 2, 3]},
        Exception("Rate limit exceeded"),
        {"batch": 2, "items": [4, 5, 6]}
    ]

    # Process first batch
    result1 = await mock_api.fetch_batch()
    assert result1["batch"] == 1

    # Second call raises exception
    with pytest.raises(Exception, match="Rate limit"):
        await mock_api.fetch_batch()

    # Third call succeeds
    result3 = await mock_api.fetch_batch()
    assert result3["batch"] == 2

    assert mock_api.fetch_batch.await_count == 3
```

### Example 3: Testing Service with Async Dependencies
```python
import pytest
import pytest_asyncio
from unittest.mock import AsyncMock
from project_watch_mcp.domain.values.service_result import ServiceResult

@pytest.fixture
def mock_embedding_service():
    """Create mock embedding service."""
    service = AsyncMock()
    service.get_embeddings.return_value = ServiceResult.success(
        [[0.1, 0.2, 0.3], [0.4, 0.5, 0.6]]
    )
    return service

@pytest.fixture
def mock_repository():
    """Create mock code repository."""
    repo = AsyncMock()
    repo.store_chunks.return_value = ServiceResult.success(True)
    return repo

@pytest.mark.asyncio
async def test_chunking_service(mock_embedding_service, mock_repository):
    """Test chunking service with async dependencies."""
    from project_watch_mcp.application.services.chunking_service import ChunkingService

    # Create service with mocked dependencies
    service = ChunkingService(
        embedding_service=mock_embedding_service,
        repository=mock_repository,
        settings=mock_settings
    )

    # Execute async operation
    result = await service.process_file("test.py")

    # Verify async calls
    mock_embedding_service.get_embeddings.assert_awaited_once()
    mock_reposi