How to mock httpx.AsyncClient() in Pytest

user_16944274 0
Viewed 986

I need to write test case for a function which use to fetch data from API. In there i used httpx.AsyncClient() as context manager. But i don't understand how to write test case for that function.

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()

I need to write test cases without calling the API. So there for i believe in this case i need to mock client.post() but i do not understand how to do that. If anyone can help me to figure this out that would be really helpful for me.

This image also include my code block

httpx pytest asynchronous mocking python
2 Answers

TL;DR: use return_value.__aenter__.return_value to mock the async context.

Assuming you are using Pytest and pytest-mock, your can use the mocker fixture to mock httpx.AsyncClient.

Since the post function is async, you will need to use an AsyncMock. Finally, since you use an async context, you will also need to use return_value.__aenter__.return_value to properly mock the returned context. Note for a synchronous context, simply use __enter__ instead of __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
user_21137170

I also faced with same issue and handled it with patch decorator. I share my code, so that can maybe help for others.

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'
user_101785080
Related