Mocking Async API Calls in FastAPI Tests

#fastapi #python #testing #webdev

When engaging in software development, there is often a need to send SMS, emails, or make API calls to other systems. When testing such programs, it is common to mock these interactions to ensure that the testing process does not fail due to an unavailable network connection.

In FastAPI, there are two ways to define functions: using def and async def. According to the official documentation, testing these functions requires the use of TestClient for def and httpx.AsyncClient for async def.

This article provides a practical method for mocking the API caller within async def functions.

code

Here the API endpoint makes an API call to 3rd party

# example_router.py

import httpx

@router.post('/examples')
async def add_example(payload) -> Any:
    try:
        async with httpx.AsyncClient() as client:
            r = await client.post('https://3rd-party-api.example.com',
                                  json={**payload.dict()})
            r.raise_for_status()
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    return JsonResponse(content={})

We should mock the httpx.AsyncClient inside add_example.

First, make a fixture

# conftest.py

@pytest.fixture
async def async_client(app: FastAPI) -> AsyncClient:
    return AsyncClient(app=app, base_url='http://localhost:1234')

Then write the test with mock, the key points are:

Uses the with patch to mock the httpx.AsyncClient.
Mock your calling method, here is post

You cannot patch the whole httpx.AsyncClient using @patch decorator because it will mock all httpx.AsyncClient including the one declared in your test.

# test_example_router.py

from unittest.mock import AsyncMock, patch

from httpx import AsyncClient
import pytest

@pytest.mark.asyncio
async def test_contact_us(async_client: AsyncClient) -> None:
    with patch('httpx.AsyncClient') as mock_client:
        mock_post = AsyncMock()
        mock_client.return_value.__aenter__.return_value.post = mock_post

        response = await async_client.post(url='/api/examples',
                                           json={
                                               'desc': 'desc',
                                               'name': 'Brid Pett',
                                               'phone': '123456789',
                                               'email': 'test@cc.cc',
                                           })
        mock_post.assert_called_once()
        assert response.status_code == status_code