如何在 Pytest 中模拟 httpx.AsyncClient()

Question

我需要为一个用于从 API 获取数据的函数编写测试用例。在那里我使用 httpx.AsyncClient() 作为上下文管理器。但我不明白如何为该功能编写测试用例。

async def make_dropbox_request(url, payload, dropbox_token):
async with httpx.AsyncClient(timeout=None, follow_redirects=True) as client:
    headers = {
        'Content-Type': 'application/json',
        'authorization': 'Bearer '+ dropbox_token
    }
    # make the api call
    response = await client.post(url, headers=headers, json=payload)
    
    if response.status_code not in [200]:
        print('Dropbox Status Code: ' + str(response.status_code))

    if response.status_code in [200, 202, 303]:
        return json.loads(response.text)

    elif response.status_code == 401:
        raise DropboxAuthenticationError()

    elif response.status_code == 429:
        sleep_time = int(response.headers['Retry-After'])
        if sleep_time < 1*60:
            await asyncio.sleep(sleep_time)
            raise DropboxMaxRateLimitError()
        raise DropboxMaxDailyRateLimitError()

    raise DropboxHTTPError()

我需要编写测试用例而不调用API。所以我相信在这种情况下我需要模拟 client.post() 但我不明白该怎么做。如果有人能帮助我解决这个问题，那对我来说真的很有帮助。

这张图片还包括我的代码块

Answer 1

我也面临同样的问题并使用

patch

装饰器处理它。我分享我的代码，这样可能对其他人有帮助。

from unittest.mock import patch
import pytest
import httpx
from app.services import your_service


@pytest.mark.anyio
@patch(
    'app.services.your_service.httpx.AsyncClient.post',
    return_value = httpx.Response(200, json={'id': '9ed7dasdasd-08ff-4ae1-8952-37e3a323eb08'})
)
async def test_get_id(mocker):        
    result = await your_service.get_id()
    assert result == '9ed7dasdasd-08ff-4ae1-8952-37e3a323eb08'

Answer 2

TL;DR：使用

return_value.__aenter__.return_value

来模拟异步上下文。

假设您正在使用 Pytest 和 pytest-mock，您可以使用

mocker

夹具来模拟

httpx.AsyncClient

。

由于

post

函数是异步的，因此您需要使用

AsyncMock

。 最后，由于您使用异步上下文，因此您还需要使用

return_value.__aenter__.return_value

来正确模拟返回的上下文。请注意，对于同步上下文，只需使用

__enter__

而不是

__aenter__

。

@pytest.fixture
def mock_AsyncClient(mocker: MockerFixture) -> Mock:
      mocked_AsyncClient = mocker.patch(f"{TESTED_MODULE}.AsyncClient")

      mocked_async_client = Mock()
      response = Response(status_code=200)
      mocked_async_client.post = AsyncMock(return_value=response)
      mocked_AsyncClient.return_value.__aenter__.return_value = mocked_async_client

      return mocked_async_client

Answer 3

您可以尝试使用 RESPX 模拟库来测试和模拟您的

HTTPX

客户端。 在你的情况下，类似这样的事情应该可以做到：

async def make_dropbox_request(url, payload, dropbox_token):
    ...
    response = await client.post(url, headers=headers, json=payload)
    ...
    return response.json()
​
​
@respx.mock
async def test_dropbox_endpoint():
    url = "https://dropbox-api/some-endpoint/"
    endpoint = respx.post(url).respond(json={"some": "data"})
    result = await make_dropbox_request(url, ..., ...)
    assert endpoint.called
    assert result == {"some": "data"}

为了保持干燥并且不在每个测试中重复模拟，您可以在全局范围内设置自己的

pytest

固定装置或 respx 实例，预先模拟所有 dropbox api 端点，然后在每个测试中只需根据测试场景，以获得

make_dropbox_request

的完整测试覆盖。

@pytest.fixture()
async def dropbox_mock():
    async with respx.mock() as dropbox:
        # default endpoints and their responses
        dropbox.post("some-endpoint", name="foo").respond(404)
        dropbox.post("some-other-endpoint", name="bar").respond(404)
        #                                     ^ name routes for access in tests
        yield dropbox
​
​
async def test_some_case(dropbox_mock):
    dropbox_mock["foo"].respond(json={})
    ....

Answer 4

如果您使用 pytest，则可以依靠 pytest-httpx 来模拟来自 httpx 的客户端和/或 AsyncClient 并阻止实际发出任何请求。 请注意，如文档中所述，如果需要，您仍然可以为专用主机名添加一些例外。

由于您没有指定您实际想要测试的内容，我会让您弄清楚您想要检查的内容（url，标头，正文，...），但是以下示例将确保请求没有真正到达相关的主持人：

import pytest

@pytest.mark.asyncio
async def test_something_async(httpx_mock):
    await make_dropbox_request(...)

请注意，您不需要使用异步测试用例来模拟 AsyncClient（以防您的实际异步调用可以以同步方式触发）。

Answer 5

在我们的例子中，我们想使用

httpx

来模拟

requests_mock

，因此我们只需将所有 HTTP 请求定向到

requests

@pytest.fixture(autouse=True)
def _mock_httpx_request(mocker: MockerFixture) -> None:
    """Direct all httpx.AsyncClient.request calls to requests.request, so we can use requests_mock to mock."""

    async def _mock_httpx_request(self: AsyncClient, method: str, url: str, **kwargs: Any) -> httpx.Response:
        response = requests.request(method, url, **kwargs)

        request = self.build_request(method, url)  # need this to raise httpx.HTTPStatusError
        return httpx.Response(
            status_code=response.status_code, content=response.content, headers=response.headers, request=request
        )

    mocker.patch("httpx.AsyncClient.request", new=_mock_httpx_request)