test-setup-async

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:

# 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:

[tool.pytest.ini_options]
asyncio_mode = "auto"  # Auto-detect async tests

Step 2: Choose Fixture Type

Async Fixture (for setup/teardown):

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):

@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:

mock_service = AsyncMock()
mock_service.process.return_value = "result"

Side effect (exception):

mock_service.process.side_effect = Exception("Connection failed")

Side effect (sequence):

mock_service.fetch.side_effect = [
    {"batch": 1},
    {"batch": 2},
    {"batch": 3}
]

Side effect (custom function):

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:

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:

@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:

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:

assert mock_service.fetch_data.await_count == 3

Assert not called:

mock_service.error_handler.assert_not_awaited()

Usage Examples

Example 1: Async Fixture with Cleanup

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

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

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_repository.store_chunks.assert_awaited_once()

    assert result.is_success()

Example 4: Error Handling in Async Tests

import pytest
from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_error_handling():
    """Test service handles async errors correctly."""
    mock_db = AsyncMock()
    mock_db.execute_query.side_effect = Exception("Connection lost")

    service = MyService(database=mock_db)

    result = await service.fetch_data()

    # Verify ServiceResult.failure() was returned
    assert not result.is_success()
    assert "Connection lost" in result.error

    # Verify retry logic was triggered
    assert mock_db.execute_query.await_count == 3  # 3 retries

Example 5: Async Generator Fixture (Resource Pool)

import pytest_asyncio
from collections.abc import AsyncGenerator

@pytest_asyncio.fixture
async def resource_pool() -> AsyncGenerator[ResourcePool, None]:
    """Create resource pool with async initialization and cleanup."""
    pool = ResourcePool()

    # Async initialization
    await pool.initialize()
    await pool.create_connections(count=5)

    yield pool

    # Async cleanup
    await pool.drain()
    await pool.shutdown()

@pytest.mark.asyncio
async def test_resource_acquisition(resource_pool):
    """Test acquiring resource from pool."""
    async with resource_pool.acquire() as resource:
        result = await resource.execute("SELECT 1")

    assert result is not None

Expected Outcomes

Successful Async Test Setup

✅ Async Test Configured

Configuration:
  ✅ pytest-asyncio installed
  ✅ pyproject.toml configured (asyncio_mode = "auto")
  ✅ AsyncMock imported from unittest.mock

Test Results:
  ✅ All async tests discovered (@pytest.mark.asyncio)
  ✅ Async fixtures working (setup/teardown with yield)
  ✅ AsyncMock assertions passing (assert_awaited_once)
  ✅ No "coroutine was never awaited" warnings

Example test output:
  tests/unit/test_service.py::test_async_function PASSED
  tests/unit/test_service.py::test_with_async_fixture PASSED

Common Failure - Mock Configuration Error

❌ Test Failed: RuntimeWarning

Error: RuntimeWarning: coroutine 'AsyncMock' was never awaited
  File "tests/unit/test_service.py", line 42

Root Cause:
  - Used MagicMock instead of AsyncMock for async method
  - Mock configured incorrectly

Fix:
  # ❌ WRONG
  mock_service.fetch = MagicMock(return_value="result")

  # ✅ CORRECT
  from unittest.mock import AsyncMock
  mock_service.fetch = AsyncMock(return_value="result")

---

Requirements

Dependencies:

• pytest-asyncio>=0.21.0 - Async test support
• pytest>=7.0.0 - Test framework
• Python 3.11+ with async/await support

Configuration:

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
markers = [
    "asyncio: mark test as async",
]

Knowledge:

• Python async/await syntax
• pytest fixture patterns
• Mock object configuration
• AsyncMock vs MagicMock differences

Optional:

• FastAPI or async web framework experience
• Database async drivers (asyncpg, motor, neo4j async driver)

---

Common Patterns

Pattern 1: Mock with Both Sync and Async Methods

from unittest.mock import AsyncMock, MagicMock

mock = MagicMock()
mock.sync_method = MagicMock(return_value="sync_result")
mock.async_method = AsyncMock(return_value="async_result")

# Usage
result1 = mock.sync_method()          # Sync call
result2 = await mock.async_method()   # Async call

Pattern 2: Async Fixture Scope

# Function scope (default) - new instance per test
@pytest_asyncio.fixture(scope="function")
async def function_scoped_resource():
    resource = await create_resource()
    yield resource
    await resource.cleanup()

# Session scope - shared across all tests
@pytest_asyncio.fixture(scope="session")
async def session_scoped_resource():
    resource = await create_expensive_resource()
    yield resource
    await resource.cleanup()

Pattern 3: Parameterized Async Tests

@pytest.mark.asyncio
@pytest.mark.parametrize("input_value,expected", [
    ("test1", "result1"),
    ("test2", "result2"),
    ("test3", "result3"),
])
async def test_with_parameters(input_value, expected):
    result = await async_function(input_value)
    assert result == expected

Pattern 4: Testing Concurrent Operations

import asyncio
import pytest

@pytest.mark.asyncio
async def test_concurrent_execution():
    """Test multiple async operations running concurrently."""
    mock_service = AsyncMock()
    mock_service.process.side_effect = [
        asyncio.sleep(0.1, result="result1"),
        asyncio.sleep(0.1, result="result2"),
        asyncio.sleep(0.1, result="result3"),
    ]

    # Execute concurrently
    results = await asyncio.gather(
        mock_service.process("item1"),
        mock_service.process("item2"),
        mock_service.process("item3"),
    )

    assert len(results) == 3
    assert mock_service.process.await_count == 3

Red Flags to Avoid

Mock Configuration Mistakes

1. Using MagicMock for async methods - Always use AsyncMock for async def methods
2. Missing return_value - AsyncMock() without return_value returns another AsyncMock
3. Forgetting await - Calling async mock without await causes warnings
4. Wrong assertion method - Use assert_awaited_once() not assert_called_once() for async
5. Sync fixture for async resource - Use @pytest_asyncio.fixture for async setup/teardown

Test Design Mistakes

1. No @pytest.mark.asyncio decorator - Async tests won't run correctly
2. Mixing sync and async fixtures incorrectly - Driver is sync, session is async
3. Not configuring asyncio_mode - Tests may fail inconsistently
4. Using run_in_executor for simple async - Unnecessary complexity

Cleanup Mistakes

1. No cleanup in async fixtures - Resources leak (connections, files)
2. Missing yield in fixture - Cleanup code never runs
3. Exception in setup prevents cleanup - Use try/finally or pytest handles it

Testing Mistakes

1. Not testing async context managers - Missing aenter/aexit mocks
2. No side_effect for sequences - Can't test retry logic or multi-call scenarios
3. Ignoring await_count - Not verifying async method called correct number of times

---

Troubleshooting

Issue: "RuntimeWarning: coroutine was never awaited"

Cause: Mock returned coroutine instead of value Fix: Use AsyncMock instead of MagicMock for async methods

Issue: "TypeError: object AsyncMock can't be used in 'await' expression"

Cause: Used AsyncMock() for driver/client, should use MagicMock() Fix: Outer object is sync, only context manager methods are async

Issue: "assert_awaited_once() raises AttributeError"

Cause: Used MagicMock instead of AsyncMock Fix: Async methods must use AsyncMock

Issue: Fixture cleanup not running

Cause: Exception during test prevents cleanup Fix: Use try/finally in fixture or @pytest_asyncio.fixture handles it

Integration Points

With Other Skills

setup-async-testing integrates with:

• implement-feature-complete - Provides async test patterns for Stage 2 (TDD)
• debug-test-failures - Works with async test debugging
• setup-pytest-fixtures - Combines sync and async fixture patterns
• implement-async-context-manager - Testing async context managers

With Agent Workflows

Agents should use this skill:

• @unit-tester - When testing async functions
• @integration-tester - For async database/API tests
• @implementer - During TDD workflow with async code

With Testing Tools

Compatible with:

• pytest-asyncio (required)
• AsyncMock from unittest.mock
• pytest fixtures and markers
• Neo4j async driver testing
• FastAPI async endpoint testing

Expected Benefits

Metric	Without Skill	With Skill	Improvement
Test Setup Time	30 min (research)	5 min (template)	83% faster
Mock Configuration Errors	40% (wrong mock type)	5% (correct patterns)	88% reduction
"Coroutine never awaited" Warnings	20% of tests	0% (proper AsyncMock)	100% elimination
Async Fixture Cleanup Issues	30% (leaks)	0% (proper yield)	100% fix
Test Reliability	70% (flaky async)	98% (stable)	40% improvement
Knowledge Transfer Time	2 hours (learning)	15 min (examples)	88% faster

Success Metrics

Metric	Target	Measurement
Zero Async Warnings	100%	pytest output
Mock Configuration Accuracy	100%	AsyncMock validation
Fixture Cleanup Success	100%	Resource leak detection
Test Reliability	98%+ pass rate	CI/CD metrics
Pattern Adoption	90% of async tests	Code review

Validation Process

Step 1: Dependency Validation

# Ensure pytest-asyncio installed
uv pip list | grep pytest-asyncio

# Verify pyproject.toml configuration
cat pyproject.toml | grep asyncio_mode

Step 2: Test Configuration Validation

# Check test file has correct decorator
@pytest.mark.asyncio
async def test_name():
    ...

Step 3: Mock Configuration Validation

# Verify AsyncMock used for async methods
from unittest.mock import AsyncMock
mock_service.async_method = AsyncMock(return_value=...)

Step 4: Fixture Validation

# Async fixtures use AsyncGenerator
from collections.abc import AsyncGenerator

@pytest_asyncio.fixture
async def resource() -> AsyncGenerator[Resource, None]:
    resource = await create()
    yield resource
    await resource.cleanup()

Step 5: Execution Validation

# Run tests with verbose output
uv run pytest tests/test_async.py -v

# Verify no warnings
# ✓ No "RuntimeWarning: coroutine was never awaited"
# ✓ All async assertions pass

Automation Scripts

The scripts/ directory contains production-ready automation utilities:

1. validate_async_tests.py - Validate async test patterns, detect anti-patterns

   python scripts/validate_async_tests.py tests/unit/ --severity warning
   

2. convert_to_async.py - Convert sync tests to async automatically

   python scripts/convert_to_async.py test_file.py --dry-run
   

3. generate_async_fixture.py - Generate fixture boilerplate

   python scripts/generate_async_fixture.py neo4j_driver database
   

📖 See references/SCRIPTS-REFERENCE.md for complete documentation and examples.

Python Scripts

• convert_to_async.py - Convert synchronous tests to async test patterns
• generate_async_fixture.py - Generate async fixture boilerplate code
• validate_async_tests.py - Validate async test patterns and find anti-patterns

See Also

• pytest-asyncio documentation
• templates/async-test-template.py - Copy-paste template
• references/SCRIPTS-REFERENCE.md - Complete automation scripts documentation
• references/QUICK-REFERENCE.md - One-page quick reference
• references/IMPLEMENTATION-SUMMARY.md - Technical implementation details