使用pytest-asyncio和moto测试异步DynamoDB函数遇协程未等待警告

问题场景 #

我编写了一个异步函数用于列出并返回DynamoDB表：

async def func(connectionId: str | None) -> list[str]:
    async with aioboto3.Session().client("dynamodb") as db:
        response = await db.list_tables()
        table_names = response.get('TableNames', [])

    return table_names

使用moto编写单元测试时：

@pytest.mark.asyncio
@mock_aws
async def test_func():
    async with aioboto3.Session().client("dynamodb") as db:
        exp_result = ['test']
        await db.create_table(
                    TableName='test',
                    KeySchema=[{'AttributeName': 'connectionId', 'KeyType': 'HASH'}],
                    AttributeDefinitions=[{'AttributeName': 'connectionId', 'AttributeType': 'S'}],
                    ProvisionedThroughput={'ReadCapacityUnits': 1, 'WriteCapacityUnits': 1}
                )

        result = await func("2")
        assert result == exp_result

运行测试抛出警告：

RuntimeWarning: coroutine 'test_func' was never awaited

该问题仅在同时使用moto和pytest-asyncio时出现，同步版本的函数与测试可正常通过，已确认安装pytest-asyncio 0.23.6，接受非moto的解决方案。

解决方法 #

方法1：调整装饰器顺序 #

moto的@mock_aws装饰器需要放在@pytest.mark.asyncio下方，装饰器执行顺序是从下到上，确保pytest-asyncio先处理异步函数逻辑，再应用moto的mock：

@mock_aws
@pytest.mark.asyncio
async def test_func():
    # 原有测试代码保持不变

方法2：使用moto异步上下文管理器 #

改用mock_aws异步上下文管理器包裹测试逻辑，避免装饰器顺序冲突：

@pytest.mark.asyncio
async def test_func():
    async with mock_aws():
        async with aioboto3.Session().client("dynamodb") as db:
            exp_result = ['test']
            await db.create_table(
                        TableName='test',
                        KeySchema=[{'AttributeName': 'connectionId', 'KeyType': 'HASH'}],
                        AttributeDefinitions=[{'AttributeName': 'connectionId', 'AttributeType': 'S'}],
                        ProvisionedThroughput={'ReadCapacityUnits': 1, 'WriteCapacityUnits': 1}
                    )

            result = await func("2")
            assert result == exp_result

方法3：用pytest-mock直接模拟客户端（非moto方案） #

如果不想依赖moto，可通过pytest-mock直接模拟DynamoDB客户端的行为：

@pytest.mark.asyncio
async def test_func(mocker):
    # 模拟异步DB客户端的list_tables返回值
    mock_db = mocker.Mock()
    mock_db.list_tables.return_value = {'TableNames': ['test']}
    
    # 模拟aioboto3的client异步上下文
    mock_session_client = mocker.patch('aioboto3.Session.client')
    mock_session_client.return_value.__aenter__.return_value = mock_db

    result = await func("2")
    assert result == ['test']

内容的提问来源于stack exchange，提问作者Vishal Balaji