测试Asyncio ensure_future异步回调方法遇断言错误，求指导

问题说明 #

我有一个发起回调请求的异步方法，编写单元测试时遇到以下错误：

FAILED tests/test_async_mapper_controller.py::test_make_callback - AssertionError: expected call not found.

原异步方法代码 #

async def make_callback(
    async_call_back_request: AsyncCallbackRequest, url=None, url_suffix=None
):
    if not (url or _config.default_callback_url):
        return
    elif not url:
        url = _config.default_callback_url

    asyncio.ensure_future(
        _callback(async_call_back_request, url=url, url_suffix=url_suffix)
    )

async def _callback(
    async_call_back_request: AsyncCallbackRequest, url, url_suffix=None
):
    try:
        res = httpx.post(
            f"{url.rstrip('/')}{url_suffix}",
            headers={"content-type": "application/json"},
            content=async_call_back_request.model_dump_json(),
            timeout=_config.default_callback_timeout,
        )

        res.raise_for_status()
    except Exception as e:
        _logger.error(f"Error during callback: {e}")

当前测试代码 #

@pytest.mark.asyncio
async def test_make_callback():
    async_call_back_request = AsyncCallbackRequest(
        header=AsyncCallbackRequestHeader(
            message_id="123",
            message_ts="2021-05-01T12:00:00Z",
            action="test_action",
            status=StatusEnum.succ,
        ),
        message={"key": "value"},
    )

    url = "http://test.com/callback"
    url_suffix = "/suffix"

    with patch(
        "openg2p_spar_mapper_api.controllers.async_mapper_controller.httpx.post"
    ) as mock_post:
        mock_response = MagicMock()
        mock_response.raise_for_status.return_value = None
        mock_post.return_value = mock_response

        # Execute the make_callback method
        task = asyncio.ensure_future(
            AsyncMapperController.make_callback(
                async_call_back_request, url, url_suffix
            )
        )

        # Wait for the make_callback task to complete
        await task

        # This line waits for all other spawned tasks to complete
        await asyncio.gather(
            *[t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
        )

        # Check if httpx.post was called as expected
        mock_post.assert_called_once_with(
            f"{url.rstrip('/')}{url_suffix}",
            headers={"content-type": "application/json"},
            content=async_call_back_request.model_dump_json(),
            timeout=30,
        )

问题根源与修复方案 #

1. 异步任务等待逻辑不可靠 #

make_callback仅通过asyncio.ensure_future启动_callback任务就直接返回，你用await asyncio.gather(...)收集所有任务的方式容易误包含测试框架的其他任务，导致_callback未执行完成就进入断言环节。

修复方式：直接捕获asyncio.ensure_future创建的_callback协程，手动执行确保完成：

@pytest.mark.asyncio
async def test_make_callback():
    async_call_back_request = AsyncCallbackRequest(
        header=AsyncCallbackRequestHeader(
            message_id="123",
            message_ts="2021-05-01T12:00:00Z",
            action="test_action",
            status=StatusEnum.succ,
        ),
        message={"key": "value"},
    )

    url = "http://test.com/callback"
    url_suffix = "/suffix"

    with patch("asyncio.ensure_future") as mock_ensure_future, \
         patch("openg2p_spar_mapper_api.controllers.async_mapper_controller.httpx.post") as mock_post:
        mock_response = MagicMock()
        mock_response.raise_for_status.return_value = None
        mock_post.return_value = mock_response

        # 执行make_callback
        await AsyncMapperController.make_callback(async_call_back_request, url, url_suffix)

        # 确认ensure_future被调用，并获取_callback协程执行
        assert mock_ensure_future.called
        callback_coroutine = mock_ensure_future.call_args.args[0]
        await callback_coroutine

        # 断言调用是否符合预期
        mock_post.assert_called_once_with(
            f"{url.rstrip('/')}{url_suffix}",
            headers={"content-type": "application/json"},
            content=async_call_back_request.model_dump_json(),
            timeout=_config.default_callback_timeout,
        )

2. Timeout参数硬编码不匹配 #

测试中断言用了timeout=30，但实际代码使用的是_config.default_callback_timeout，如果配置值不是30就会导致断言失败。

修复方式：二选一即可：

• 补丁配置值：

  with patch("openg2p_spar_mapper_api.controllers.async_mapper_controller._config.default_callback_timeout", 30):
      # 测试代码...
      mock_post.assert_called_once_with(..., timeout=30)
  

• 直接引用配置变量：

  mock_post.assert_called_once_with(..., timeout=_config.default_callback_timeout)
  

3. （可选）修复异步方法中的同步IO问题 #

当前_callback里用了同步的httpx.post，在异步函数中会阻塞事件循环，建议改用异步客户端：

async def _callback(
    async_call_back_request: AsyncCallbackRequest, url, url_suffix=None
):
    try:
        async with httpx.AsyncClient() as client:
            res = await client.post(
                f"{url.rstrip('/')}{url_suffix}",
                headers={"content-type": "application/json"},
                content=async_call_back_request.model_dump_json(),
                timeout=_config.default_callback_timeout,
            )
        res.raise_for_status()
    except Exception as e:
        _logger.error(f"Error during callback: {e}")

对应的测试需要调整补丁路径为openg2p_spar_mapper_api.controllers.async_mapper_controller.httpx.AsyncClient.post。

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