875 reads

Python: Checking API Application Requests to Third-Party Services

by Aleksei SharypovSeptember 16th, 2023

Too Long; Didn't Read

I will show you some examples of how you can verify requests to third-party services in tests using of async FastAPI + httpx (not only) + mocks. I'll show you 2 methods. The main task is to compare the request that our application sent to the mock. If it is different, then we need to return an error in the test. At the same time, a correct request may often have a different visual representation, so just comparing the request body with the template will not work. I compare dict or list as it's more flexible.

featured image - Python: Checking API Application Requests to Third-Party Services

For many, using a mock of a third-party service in tests is a common thing. The code's response to a specific response (HTTP code and body) is checked. But how often do you check what request your Python application makes to this very third-party service? After all, the mock will return the desired response for any request body.

I'm really keen on writing tests. Moreover, I often use TDD. For me, it is important to cover the code with tests as much as possible. I often study other people's projects and see gaps not only in the tests but also in the code itself. And these gaps go into production. It happens because sometimes, we forget about some details.

In addition to checking requests to third-party services, these include, for example, migration testing, transaction rollbacks, etc…

Now, I will show you some examples of how you can verify requests to third-party services in tests. I will do it using the example of async FastAPI + httpx. However, aiohttp, requests, and even Django have a similar set of methods for verification.

The main task is to compare the request that our application sent to the stub. If it is different, then we need to return an error in the test. At the same time, a correct request may often have a different visual representation, so just comparing the request body with the template will not work. I compare dict or list as it's more flexible.

As an example, let's take the following simple application:

import httpx
from fastapi import FastAPI

app = FastAPI()
client = httpx.AsyncClient()


@app.post("/")
async def root():
    result1 = await client.request(
        "POST",
        "http://example.com/user",
        json={"name": "John", "age": 21},
    )
    user_json = result1.json()
    result2 = await client.request(
        "POST",
        "http://example1.com/group/23",
        json={"user_id": user_json["id"]},
    )
    group_json = result2.json()
    return {"user": user_json, "group": group_json}

This application makes two consecutive requests. For the second request, it uses the result from the first one. In both cases, when using standard mocks, our application can send an absolutely invalid request and will still receive the desired mock response.

So, let's see what we can do about it.

Test without checking requests

The most popular package for sending requests to third-party services is httpx. I usually use pytest_httpx as mocks, and for aiohttp, I often use aiorequests.

A simple test to check this endpoint looks something like this:

import httpx
import pytest
import pytest_asyncio
from pytest_httpx import HTTPXMock
from app.main import app

TEST_SERVER = "test_server"


@pytest_asyncio.fixture
async def non_mocked_hosts() -> list:
    return [TEST_SERVER]


@pytest.mark.asyncio
async def test_success(httpx_mock: HTTPXMock):
    httpx_mock.add_response(
        method="POST",
        url="http://example.com/user",
        json={"id": 12, "name": "John", "age": 21},
    )
    httpx_mock.add_response(
        method="POST",
        url="http://example1.com/group/23",
        json={"id": 23, "user_id": 12},
    )
    async with httpx.AsyncClient(app=app, base_url=f"http://{TEST_SERVER}") as async_client:
        response = await async_client.post("/")
    assert response.status_code == 200
    assert response.json() == {
        'user': {'id': 12, 'name': 'John', 'age': 21},
        'group': {'id': 23, 'user_id': 12},
    }

For requests to the endpoint of the application, we also use httpx.AsyncClient. Not to accidentally mock it, we add the non_mocked_hosts fixture. And in the test, we add two mocks using httpx_mock.add_response. Method and URL are used as matchers here. If the request with the mocked URL and method is not called before the test ends, we will get an error.

Now, let's try to do something to ensure that the request body is the one that the external service expects.

The first option is built into httpx_mock. We can add content - this is a byte object of the full request text.

httpx_mock.add_response(
    method="POST",
    url="http://example.com/user",
    json={"id": 12, "name": "John", "age": 21},
    content=b'{"name": "John", "age": 21}',
)

In this case, during the search, not only the method and URL will be compared, but also this very content. But, as I mentioned above, we definitely need to know the request body. A discrepancy in any character will result in an error. For example, a space between a colon and a key, a line transition and any formatting details, or just a different order of keys in the dictionary. This way of testing JSON is very inefficient. However, the package does not have a built-in regular validator of requests in JSON format.

Simple method

Getting requests at the end of the test and comparing them with the templates seems to be the easiest way.

import json

import httpx
import pytest
import pytest_asyncio
from pytest_httpx import HTTPXMock
from app.main import app

TEST_SERVER = "test_server"


@pytest_asyncio.fixture
async def non_mocked_hosts() -> list:
    return [TEST_SERVER]


@pytest.mark.asyncio
async def test_with_all_requests_success(httpx_mock: HTTPXMock):
    httpx_mock.add_response(
        method="POST",
        url="http://example.com/user",
        json={"id": 12, "name": "John", "age": 21},
    )
    httpx_mock.add_response(
        method="POST",
        url="http://example1.com/group/23",
        json={"id": 23, "user_id": 12},
    )
    async with httpx.AsyncClient(app=app, base_url=f"http://{TEST_SERVER}") as async_client:
        response = await async_client.post("/")
    assert response.status_code == 200
    assert response.json() == {
        'user': {'id': 12, 'name': 'John', 'age': 21},
        'group': {'id': 23, 'user_id': 12},
    }
    expected_requests = [
        {"url": "http://example.com/user", "json": {"name": "John", "age": 21}},
        {"url": "http://example1.com/group/23", "json": {"user_id": 12}},
    ]
    for expecter_request, real_request in zip(expected_requests, httpx_mock.get_requests()):
        assert expecter_request["url"] == real_request.url
        assert expecter_request["json"] == json.loads(real_request.content)

This method is quite huge and requires repeating the code.

More complex method

I decided to fix the package a bit. I could have created a child class by overriding the necessary methods, but I created a slightly different layer. To do this, I added the APIMock class:

from typing import Any
import httpx
import json
from pytest_httpx import HTTPXMock


class APIMock:
    def __init__(self, httpx_mock: HTTPXMock):
        self.httpx_mock = httpx_mock

    async def request_mock(
        self,
        method: str,
        url: str,
        status_code: int = 200,
        request_for_check: Any = None,
        **kwargs,
    ):
        async def custom_response(request: httpx.Request) -> httpx.Response:
            response = httpx.Response(
                status_code,
                **kwargs,
            )
            if request_for_check:
                assert request_for_check == json.loads(request.content), \
                    f"{request_for_check} != {json.loads(request.content)}"
            return response
        self.httpx_mock.add_callback(custom_response, method=method, url=url)

There is only one public method in this class. It adds the mock by calling add_callback from httpx_mock.

Its structure is from the original add_response from httpx_mock method, but it is a bit shorter. If you need more parameters, you can always add them.

To use it, it is enough to import it through a fixture and call instead add_request.

Below is an example of a test using the APIMock class and request verification.

import httpx
import pytest
import pytest_asyncio
from pytest_httpx import HTTPXMock
from app.main import app
from app.api_mock import APIMock

TEST_SERVER = "test_server"


@pytest_asyncio.fixture
async def non_mocked_hosts() -> list:
    return [TEST_SERVER]


@pytest_asyncio.fixture
async def requests_mock(httpx_mock: HTTPXMock):
    return APIMock(httpx_mock)


@pytest.mark.asyncio
async def test_with_requests_success(requests_mock):
    await requests_mock.request_mock(
        method="POST",
        url="http://example.com/user",
        json={"id": 12, "name": "John", "age": 21},
        request_for_check={"name": "John", "age": 21},
    )
    await requests_mock.request_mock(
        method="POST",
        url="http://example1.com/group/23",
        json={"id": 23, "user_id": 12},
        request_for_check={"user_id": 12},
    )
    async with httpx.AsyncClient(app=app, base_url=f"http://{TEST_SERVER}") as async_client:
        response = await async_client.post("/")
    assert response.status_code == 200
    assert response.json() == {
        'user': {'id': 12, 'name': 'John', 'age': 21},
        'group': {'id': 23, 'user_id': 12},
    }

It looks quite simple and convenient. The structure allows you to specify the expected request while creating a mock as when using content. I found this method to be the most convenient one. Moreover, other classes can be inherited from this one, and they will mock each specific endpoint to a third-party service. This is very convenient.

The same can be done for the aioresponses package as well as for responses. They also support the callback function.

In theory, you will have to write a test for such a test since it’s quite logical. But this is not a problem since we have a separate class.