Back to Details

test-setup-async

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

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.
提供使用pytest-asyncio、AsyncMock和异步fixture在Python中测试异步函数的正确模式，确保测试能正确处理异步上下文管理器、副作用和资源清理。

Quick Start

快速开始

Most common use case:
python
undefined
最常见使用场景：
python
undefined

Testing an async function with mocked dependencies

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()
undefined
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()
undefined

Table of Contents

目录

  1. When to Use This Skill
  2. What This Skill Does
  3. Instructions
  4. Usage Examples
  5. Expected Outcomes
  6. Requirements
  7. Troubleshooting
  8. Red Flags to Avoid
  9. Automation Scripts
  1. 何时使用该技能
  2. 该技能能实现什么
  3. 操作步骤
  4. 使用示例
  5. 预期结果
  6. 依赖要求
  7. 问题排查
  8. 需要避免的常见错误
  9. 自动化脚本

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"
在以下场景使用本技能：
  • 测试异步函数 - 使用
    async def
    和
    await
    定义的函数
  • 创建异步fixture - 包含异步操作的初始化/清理逻辑
  • 模拟异步服务 - 数据库连接、API客户端、外部服务
  • 处理异步上下文管理器 - 
    async with
    语句
  • 测试异步生成器 - 
    async for
    模式
  • 调试异步测试失败 - 出现"RuntimeWarning: coroutine was never awaited"警告时
触发场景：
  • "测试异步函数"
  • "模拟异步方法"
  • "创建异步fixture"
  • "AsyncMock配置"
  • "pytest-asyncio环境搭建"

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.
本技能提供以下场景的实现模式：
  1. 异步测试配置 - 在pyproject.toml中配置pytest-asyncio
  2. 异步fixture创建 - 使用@pytest_asyncio.fixture结合AsyncGenerator
  3. AsyncMock使用 - 为异步方法设置return_value和side_effect
  4. 异步上下文管理器 - 模拟
    async with
    语句的__aenter__和__aexit__方法
  5. 异步断言 - 使用assert_awaited_once()、assert_awaited_once_with()等断言方法
  6. 错误处理 - 测试异步代码中的异常
详细的分步模式请查看下方的操作步骤章节。

Instructions

操作步骤

Step 1: Install Dependencies

步骤1：安装依赖

bash
uv pip install pytest-asyncio
Add to 
pyproject.toml
:
toml
[tool.pytest.ini_options]
asyncio_mode = "auto"  # Auto-detect async tests
bash
uv pip install pytest-asyncio
在
pyproject.toml
中添加配置：
toml
[tool.pytest.ini_options]
asyncio_mode = "auto"  # Auto-detect async tests

Step 2: Choose Fixture Type

步骤2：选择Fixture类型

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
异步Fixture（用于初始化/清理）：
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()
同步Fixture（用于模拟对象）：
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

步骤3：模拟异步方法

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
设置简单返回值：
python
mock_service = AsyncMock()
mock_service.process.return_value = "result"
设置副作用（异常）：
python
mock_service.process.side_effect = Exception("Connection failed")
设置副作用（序列返回）：
python
mock_service.fetch.side_effect = [
    {"batch": 1},
    {"batch": 2},
    {"batch": 3}
]
设置副作用（自定义函数）：
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

步骤4：测试异步上下文管理器

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()
模拟异步上下文管理器：
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
测试使用示例：
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

步骤5：验证异步调用

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()
断言已被等待：
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
断言调用次数：
python
assert mock_service.fetch_data.await_count == 3
断言未被调用：
python
mock_service.error_handler.assert_not_awaited()

Usage Examples

使用示例

Example 1: Async Fixture with Cleanup

示例1：带清理逻辑的异步Fixture

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

示例2：带序列副作用的AsyncMock

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

示例3：测试带有异步依赖的服务

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

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

    assert result.is_success()

Example 4: Error Handling in Async Tests

示例4：异步测试中的错误处理

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

示例5：异步生成器Fixture（资源池）

python
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
python
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
✅ 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")
❌ 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:
toml
undefined
依赖包：
  • pytest-asyncio>=0.21.0
    - 异步测试支持
  • pytest>=7.0.0
    - 测试框架
  • Python 3.11+ 支持async/await
配置要求：
toml
undefined

pyproject.toml

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)

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

**知识储备：**
- Python async/await语法
- pytest fixture模式
- 模拟对象配置
- AsyncMock与MagicMock的区别

**可选经验：**
- FastAPI或异步Web框架使用经验
- 数据库异步驱动（asyncpg, motor, neo4j async driver）使用经验

---

Common Patterns

常见模式

Pattern 1: Mock with Both Sync and Async Methods

模式1：同时包含同步和异步方法的模拟

python
from unittest.mock import AsyncMock, MagicMock

mock = MagicMock()
mock.sync_method = MagicMock(return_value="sync_result")
mock.async_method = AsyncMock(return_value="async_result")
python
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

Usage

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

Pattern 2: Async Fixture Scope

模式2：异步Fixture作用域

python
undefined
python
undefined

Function scope (default) - new instance per test

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()
@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

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()
undefined
@pytest_asyncio.fixture(scope="session") async def session_scoped_resource(): resource = await create_expensive_resource() yield resource await resource.cleanup()
undefined

Pattern 3: Parameterized Async Tests

模式3：参数化异步测试

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

模式4：测试并发操作

python
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
python
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
  1. 使用MagicMock模拟异步方法 - 始终为
    async def
    方法使用AsyncMock
  2. 未设置return_value - 未设置return_value的AsyncMock会返回另一个AsyncMock
  3. 忘记使用await - 调用异步模拟方法时不使用
    await
    会导致警告
  4. 使用错误的断言方法 - 异步方法需使用
    assert_awaited_once()
    而非
    assert_called_once()
  5. 使用同步Fixture处理异步资源 - 异步初始化/清理需使用
    @pytest_asyncio.fixture

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
  1. 缺少@pytest.mark.asyncio装饰器 - 异步测试无法正确运行
  2. 错误地混合同步和异步Fixture - 驱动是同步的，会话是异步的
  3. 未配置asyncio_mode - 测试可能会出现不一致的失败
  4. 为简单异步操作使用run_in_executor - 增加不必要的复杂度

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
  1. 异步Fixture中无清理逻辑 - 会导致资源泄漏（连接、文件等）
  2. Fixture中缺少yield - 清理代码永远不会执行
  3. 初始化时的异常导致清理逻辑不执行 - 使用try/finally或依赖pytest的自动处理

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
  1. 未测试异步上下文管理器 - 缺少对__aenter__/__aexit__的模拟
  2. 未设置序列型side_effect - 无法测试重试逻辑或多调用场景
  3. 忽略await_count - 未验证异步方法的调用次数是否正确

Troubleshooting

问题排查

Issue: "RuntimeWarning: coroutine was never awaited"

问题："RuntimeWarning: coroutine was never awaited"

Cause: Mock returned coroutine instead of value Fix: Use 
AsyncMock
instead of 
MagicMock
for async methods
原因： 模拟对象返回了协程而非具体值 解决： 为异步方法使用
AsyncMock
而非
MagicMock

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

问题："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
原因： 为驱动/客户端使用了
AsyncMock()
，应使用
MagicMock()
解决： 外层对象是同步的，仅上下文管理器方法是异步的

Issue: "assert_awaited_once() raises AttributeError"

问题："assert_awaited_once() raises AttributeError"

Cause: Used 
MagicMock
instead of 
AsyncMock
Fix: Async methods must use 
AsyncMock
原因： 使用了
MagicMock
而非
AsyncMock
解决： 异步方法必须使用
AsyncMock

Issue: Fixture cleanup not running

问题：Fixture清理逻辑未执行

Cause: Exception during test prevents cleanup Fix: Use 
try/finally
in fixture or 
@pytest_asyncio.fixture
handles it
原因： 测试过程中的异常导致清理逻辑无法执行 解决： 在Fixture中使用try/finally，或依赖@pytest_asyncio.fixture的自动处理

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
setup-async-testing可与以下技能集成：
  • implement-feature-complete - 为Stage 2（TDD）提供异步测试模式
  • debug-test-failures - 配合进行异步测试调试
  • setup-pytest-fixtures - 结合同步和异步Fixture模式
  • implement-async-context-manager - 测试异步上下文管理器

With Agent Workflows

与Agent工作流集成

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
以下Agent应使用本技能：
  • @unit-tester - 测试异步函数时
  • @integration-tester - 进行异步数据库/API测试时
  • @implementer - 在TDD工作流中处理异步代码时

With Testing Tools

与测试工具集成

Compatible with:
  • pytest-asyncio (required)
  • AsyncMock from unittest.mock
  • pytest fixtures and markers
  • Neo4j async driver testing
  • FastAPI async endpoint testing
兼容工具：
  • pytest-asyncio（必需）
  • unittest.mock.AsyncMock
  • pytest fixtures和markers
  • Neo4j异步驱动测试
  • FastAPI异步端点测试

Expected Benefits

预期收益

MetricWithout SkillWith SkillImprovement
Test Setup Time30 min (research)5 min (template)83% faster
Mock Configuration Errors40% (wrong mock type)5% (correct patterns)88% reduction
"Coroutine never awaited" Warnings20% of tests0% (proper AsyncMock)100% elimination
Async Fixture Cleanup Issues30% (leaks)0% (proper yield)100% fix
Test Reliability70% (flaky async)98% (stable)40% improvement
Knowledge Transfer Time2 hours (learning)15 min (examples)88% faster
指标未使用本技能使用本技能提升效果
测试环境搭建时间30分钟（需调研）5分钟（使用模板）提升83%效率
模拟配置错误率40%（错误的模拟类型）5%（使用正确模式）降低88%错误率
"协程未被等待"警告20%的测试出现0%（正确使用AsyncMock）完全消除警告
异步Fixture资源泄漏30%（资源未释放）0%（正确使用yield）完全解决泄漏问题
测试稳定性70%（异步测试不稳定）98%（稳定运行）提升40%稳定性
知识传递时间2小时（学习成本）15分钟（参考示例）提升88%效率

Success Metrics

成功指标

MetricTargetMeasurement
Zero Async Warnings100%pytest output
Mock Configuration Accuracy100%AsyncMock validation
Fixture Cleanup Success100%Resource leak detection
Test Reliability98%+ pass rateCI/CD metrics
Pattern Adoption90% of async testsCode review
指标目标值测量方式
无异步警告100%pytest输出结果
模拟配置准确率100%AsyncMock验证
Fixture清理成功率100%资源泄漏检测
测试通过率98%+CI/CD指标
模式使用率90%的异步测试使用代码评审

Validation Process

验证流程

Step 1: Dependency Validation

步骤1：依赖验证

bash
undefined
bash
undefined

Ensure pytest-asyncio installed

Ensure pytest-asyncio installed

uv pip list | grep pytest-asyncio
uv pip list | grep pytest-asyncio

Verify pyproject.toml configuration

Verify pyproject.toml configuration

cat pyproject.toml | grep asyncio_mode
undefined
cat pyproject.toml | grep asyncio_mode
undefined

Step 2: Test Configuration Validation

步骤2：测试配置验证

python
undefined
python
undefined

Check test file has correct decorator

Check test file has correct decorator

@pytest.mark.asyncio async def test_name(): ...
undefined
@pytest.mark.asyncio async def test_name(): ...
undefined

Step 3: Mock Configuration Validation

步骤3：模拟配置验证

python
undefined
python
undefined

Verify AsyncMock used for async methods

Verify AsyncMock used for async methods

from unittest.mock import AsyncMock mock_service.async_method = AsyncMock(return_value=...)
undefined
from unittest.mock import AsyncMock mock_service.async_method = AsyncMock(return_value=...)
undefined

Step 4: Fixture Validation

步骤4：Fixture验证

python
undefined
python
undefined

Async fixtures use AsyncGenerator

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()
undefined
from collections.abc import AsyncGenerator
@pytest_asyncio.fixture async def resource() -> AsyncGenerator[Resource, None]: resource = await create() yield resource await resource.cleanup()
undefined

Step 5: Execution Validation

步骤5：执行验证

bash
undefined
bash
undefined

Run tests with verbose output

Run tests with verbose output

uv run pytest tests/test_async.py -v
uv run pytest tests/test_async.py -v

Verify no warnings

Verify no warnings

✓ No "RuntimeWarning: coroutine was never awaited"

✓ No "RuntimeWarning: coroutine was never awaited"

✓ All async assertions pass

✓ All async assertions pass

undefined
undefined

Automation Scripts

自动化脚本

The 
scripts/
directory contains production-ready automation utilities:
  1. validate_async_tests.py - Validate async test patterns, detect anti-patterns
    bash
    python scripts/validate_async_tests.py tests/unit/ --severity warning
  2. convert_to_async.py - Convert sync tests to async automatically
    bash
    python scripts/convert_to_async.py test_file.py --dry-run
  3. generate_async_fixture.py - Generate fixture boilerplate
    bash
    python scripts/generate_async_fixture.py neo4j_driver database
📖 See references/SCRIPTS-REFERENCE.md for complete documentation and examples.
scripts/
目录包含生产可用的自动化工具：
  1. validate_async_tests.py - 验证异步测试模式，检测反模式
    bash
    python scripts/validate_async_tests.py tests/unit/ --severity warning
  2. convert_to_async.py - 自动将同步测试转换为异步测试
    bash
    python scripts/convert_to_async.py test_file.py --dry-run
  3. generate_async_fixture.py - 生成Fixture模板代码
    bash
    python scripts/generate_async_fixture.py neo4j_driver database
📖 完整文档和示例请查看references/SCRIPTS-REFERENCE.md

Python Scripts

Python脚本列表

  • 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
  • convert_to_async.py - 将同步测试转换为异步测试模式
  • generate_async_fixture.py - 生成异步Fixture模板代码
  • validate_async_tests.py - 验证异步测试模式并查找反模式

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
  • pytest-asyncio官方文档
  • templates/async-test-template.py - 可直接复制使用的模板
  • references/SCRIPTS-REFERENCE.md - 完整自动化脚本文档
  • references/QUICK-REFERENCE.md - 单页速查手册
  • references/IMPLEMENTATION-SUMMARY.md - 技术实现细节