如何在Pytest中测试含AioHttpClient与BasicAuth的异步函数？

核心思路 #

这类异步HTTP请求函数的测试核心是完全mock掉真实网络调用，重点验证两个维度：一是请求参数（包括BasicAuth凭证）是否符合预期，二是函数对不同响应结果的处理逻辑是否正确。

具体实现方案 #

1. 依赖准备 #

先安装必要的测试工具包：

pip install pytest pytest-asyncio aioresponses

2. 用aioresponses快速模拟HTTP交互 #

aioresponses是专门适配aiohttp的mock工具，能简化请求模拟和参数验证，是这类场景的首选方案。

示例测试代码：

import os
import aiohttp
import pytest
from your_module import example_function

@pytest.mark.asyncio
async def test_example_function_success(aioresponses):
    # 临时设置环境变量，避免依赖系统配置
    os.environ["SYNTH_URL"] = "https://mock-synth-service.com"
    os.environ["MY_PASSWORD"] = "test_secret_pass"

    mock_url = os.getenv("SYNTH_URL")
    expected_auth = aiohttp.BasicAuth('username', os.getenv("MY_PASSWORD"))

    # 注册mock响应，同时验证请求头里的Authorization是否匹配
    aioresponses.post(
        mock_url,
        status=200,
        body=b"valid_response_data",
        match_headers={"Authorization": expected_auth.encode()}
    )

    # 执行被测试函数
    result = await example_function(your_test_params)

    # 断言结果和请求次数
    assert result == b"valid_response_data"
    assert len(aioresponses.requests) == 1

@pytest.mark.asyncio
async def test_example_function_failure(aioresponses):
    os.environ["SYNTH_URL"] = "https://mock-synth-service.com"

    # 模拟400错误响应
    aioresponses.post(mock_url, status=400, body=b"bad_request_error")

    result = await example_function(your_test_params)

    # 根据函数错误分支的逻辑断言结果
    assert result == ...  # 替换为函数错误场景的预期返回值

3. 手动mock的替代方案（无第三方库） #

如果不想引入额外依赖，可以用unittest.mock手动mockaiohttp.ClientSession：

from unittest.mock import AsyncMock, patch
import aiohttp
import pytest
from your_module import example_function

@pytest.mark.asyncio
async def test_example_function_manual_mock():
    # 设置测试用环境变量
    os.environ["SYNTH_URL"] = "https://mock-url.com"
    os.environ["MY_PASSWORD"] = "test_pass"

    # mock响应对象
    mock_resp = AsyncMock()
    mock_resp.ok = True
    mock_resp.read.return_value = b"mock_data"

    # mock会话对象，适配async with上下文
    mock_session = AsyncMock()
    mock_session.post.return_value.__aenter__.return_value = mock_resp

    # 替换被测试函数里的AioHttpClient.session()
    with patch("your_module.AioHttpClient.session", return_value=mock_session):
        result = await example_function(your_test_params)

        # 验证post方法的调用参数是否正确
        mock_session.post.assert_called_once_with(
            "https://mock-url.com",
            data=your_expected_data,  # 替换为测试用的data_in值
            auth=aiohttp.BasicAuth('username', "test_pass")
        )
        assert result == b"mock_data"

关键注意事项 #

• 环境变量隔离：用monkeypatch.setenv()临时设置环境变量更安全，测试结束后会自动恢复，避免污染全局环境。
• 上下文管理器适配：因为代码用了async with ...，mock时必须模拟__aenter__方法的返回值，否则会报错。
• Auth参数验证：一定要验证auth参数的正确性，这是这类带认证请求测试的核心点之一，确保凭证传递符合预期。

内容的提问来源于stack exchange，提问作者Mr.Slow