Python单元测试中如何等待async异步函数返回响应结果？

原有代码核心错误 #

• 函数名不匹配：待测试的异步函数名为fetch_data，但测试代码中调用的是不存在的fetch_embedding方法
• 入参类型错误：fetch_data要求第二个参数为session实例，你第一个测试直接传入了mock的ClientSession类本身，未完成实例化
• 异步mock构造错误：response.json()是异步方法，直接赋值普通dict会导致await调用失败；mock的post返回的response对象需要符合异步调用规范，不能直接静态赋值属性
• 测试逻辑错误：第二个测试完全未调用你自己写的fetch_data方法，仅执行了mock的post方法，没有测试业务逻辑
• 事件循环管理错误：单次测试结束就关闭事件循环，会导致后续测试无法获取可用循环，且手动管理loop冗余度高

最简测试方案 #

推荐使用pytest-asyncio插件简化异步测试流程，无需手动管理事件循环：

1. 先安装依赖：

pip install pytest-asyncio

2. 修正后的测试用例：

import asyncio
import mock
import pytest
from aiohttp import ClientResponse

# 成功返回场景测试
@pytest.mark.asyncio
@mock.patch('aiohttp.ClientSession')
async def test_fetch_data_success(mock_session_cls):
    # 构造mock返回的response
    mock_response = mock.MagicMock(spec=ClientResponse)
    mock_response.status = 200
    mock_response.raise_for_status.return_value = None
    # 异步json方法需要返回可await的对象
    mock_response.json = mock.AsyncMock(return_value={"test": "value"})
    
    # 构造session实例的post方法
    mock_session = mock.MagicMock()
    mock_session.post = mock.AsyncMock(return_value=mock_response)
    mock_session_cls.return_value = mock_session

    # 实例化业务类，替换为你自己的embedder_endpoint实际值
    embedder_endpoint = "http://test-endpoint"
    bulk_embedder = BulkEmbedder(embedder_endpoint=embedder_endpoint)
    
    # 直接await调用异步方法即可，pytest-asyncio会自动管理事件循环
    result = await bulk_embedder.fetch_data("test_text", mock_session)
    
    # 校验返回结果
    assert result == {"test": "value"}
    mock_session.post.assert_called_once_with(
        url=embedder_endpoint,
        json={"text": "test_text"}
    )

# 异常场景测试
@pytest.mark.asyncio
@mock.patch('aiohttp.ClientSession')
async def test_fetch_data_failure(mock_session_cls):
    # 构造post方法抛异常的场景
    mock_session = mock.MagicMock()
    mock_session.post = mock.AsyncMock(side_effect=Exception("请求失败"))
    mock_session_cls.return_value = mock_session

    bulk_embedder = BulkEmbedder(embedder_endpoint="http://test-endpoint")
    
    # 原代码中post抛异常后response为None，后续assert response.status会触发AttributeError，可直接校验该异常
    with pytest.raises(AttributeError):
        await bulk_embedder.fetch_data("test_text", mock_session)

原有业务函数优化提示 #

你当前写的fetch_data存在隐患：如果post请求触发异常，response会保持为None，后续执行assert response.status == 200时会直接抛出AttributeError，建议调整异常分支逻辑，要么直接抛出业务异常，要么提前return避免后续空指针调用。

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