如何mock asyncio.run调用的异步函数get_responses返回结果？

核心思路 #

因为get_contacts是同步函数通过asyncio.run调用异步函数get_responses，所以可以直接mockget_responses本身，或者mock其内部依赖的异步方法，分两种方案实现：

方案1：直接mock get_responses函数 #

这种方式最简单，直接替换目标异步函数的返回值，无需关心其内部逻辑：

import pytest
from your_module import get_contacts  # 替换为实际模块路径

def test_get_contacts(mocker):
    # 定义模拟返回结果
    mock_result = [{"contact_id": 101}, {"contact_id": 102}]
    
    # 定义mock的异步函数
    async def mock_get_responses(**kwargs):
        return mock_result
    
    # 替换目标函数，注意路径要和get_contacts中导入get_responses的路径一致
    mocker.patch("your_module.get_responses", side_effect=mock_get_responses)
    
    # 执行测试
    result = get_contacts(contact_ids=[101, 102])
    
    # 断言结果符合预期
    assert result == mock_result

方案2：mock内部依赖的x.get_async_contract #

如果需要验证get_responses的内部逻辑（比如是否正确创建任务、调用依赖方法），可以mock其内部调用的异步方法：

首先安装依赖：

pip install pytest-asyncio

然后编写测试用例：

import pytest
import asyncio
from your_module import get_responses, get_contacts  # 替换为实际模块路径

@pytest.mark.asyncio
async def test_get_responses_internal_logic(mocker):
    # 模拟单个请求的返回结果
    mock_contract = {"contact_id": 101}
    
    # mock内部调用的x.get_async_contract方法
    mocker.patch("your_module.x.get_async_contract", return_value=mock_contract)
    
    # 注意：原代码中get_responses存在bug，需将await asyncio.gather(*tasks)赋值给responses
    # 修正后代码：responses = await asyncio.gather(*tasks)
    result = await get_responses(contact_ids=[101, 102])
    
    # 断言两个任务都返回了模拟结果
    assert result == [mock_contract, mock_contract]
    # 验证依赖方法被调用了两次
    your_module.x.get_async_contract.assert_has_calls([
        mocker.call(session=..., contact_id=101),
        mocker.call(session=..., contact_id=102)
    ], any_order=True)

# 同步函数的测试仍可复用方案1的方式
def test_get_contacts_with_internal_mock(mocker):
    mock_result = [{"contact_id": 101}, {"contact_id": 102}]
    
    async def mock_get_responses(**kwargs):
        return mock_result
    
    mocker.patch("your_module.get_responses", side_effect=mock_get_responses)
    
    result = get_contacts(contact_ids=[101, 102])
    assert result == mock_result

关键注意点 #

• 路径正确性：mock时必须使用被测试函数（get_contacts）中导入get_responses的路径，而非get_responses定义的路径，否则mock不会生效。
• 原代码bug修复：原get_responses中await asyncio.gather(*tasks)的结果未赋值给responses，导致永远返回空列表，需修正为responses = await asyncio.gather(*tasks)才能得到正确结果。

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