Contract Testing with HTTPX

Today, we are going to talk about how to achieve contract testing with HTTPX.

What is contract testing?

Contract testing is a methodology for ensuring that two separate systems (such as two microservices) are compatible and are able to communicate with one other. It captures the interactions that are exchanged between each service, storing them in a contract, which can then be used to verify that both parties adhere to it. - Matt Fellows

What is HTTPX?

HTTPX is a fully featured HTTP client for Python 3, which provides sync and async APIs, and support for both HTTP/1.1 and HTTP/2.

How to do contract testing with HTTPX?

Well, to be completely transparent, I'm not sure if what you are about to read classifies as contract testing.

The problem we'll be trying to solve is the following:

Consider we have multiples services running, and they depend on each other. We want to make sure that a service is not able able to break another one.

To achieve this your first thought would be "let's write end to end tests", but that will slow things down, as each service needs to be up to run the tests, and given that, the setup needed is a bit more complex.

Check this blog post (which I didn't read, but looks good) for more information about E2E testing vs Contract Testing.

The solution

Let's assume we have two services. For obvious reasons, those services are FastAPI based.

Note

This can be achieved with any web framework. What matters here is that you should be using httpx.

Service AService BRun

import httpx
from fastapi import APIRouter, FastAPI

router = APIRouter(prefix="/a")


@router.get("/")
def get_a():
    return {"a": "a"}


@router.get("/call_b")
async def call_b():
    async with httpx.AsyncClient() as client:
        response = await client.get("http://localhost:8000/b/")
        return response.json()


app = FastAPI()
app.include_router(router)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, port=8001)  # (1)!

1. The port is 8001, not 8000, to avoid conflicts with the other service.

from fastapi import APIRouter, FastAPI

router = APIRouter(prefix="/b")


@router.get("/")
def get_b():
    return {"b": "b"}


app = FastAPI()
app.include_router(router)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, port=8000)

Install the dependencies:

python -m pip install uvicorn fastapi httpx

Then open the terminal and run:

python service_a.py

Then open another terminal, and run:

python service_b.py

---

Cool. Now, let's call the /a/call_b endpoint:

http :8001/a/call_b # (1)!

1. The HTTP client used is called HTTPie, but you can use curl, or just go to the browser, and access http://localhost:8001/a/call_b.

As we see, the response is:

{
    "b": "b"
}

---

Now, if we want to create a test, we can do something like:

TestClientAsyncClientRun

import pytest
from fastapi.testclient import TestClient

from service_a import app


@pytest.fixture()
def client():
    return TestClient(app)


def test_call_b(client: TestClient) -> None:
    response = client.get("/a/call_b")
    assert response.status_code == 200
    assert response.json() == {"b": "b"}

See more on the Starlette documentation.

import httpx
import pytest
import pytest_asyncio

from service_a import app


@pytest_asyncio.fixture(name="client")
async def testclient():
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        yield client


@pytest.mark.asyncio
async def test_call_b(client: httpx.AsyncClient) -> None:
    response = await client.get("/a/call_b")
    assert response.status_code == 200
    assert response.json() == {"b": "b"}

See more on the Starlette documentation.

Install the dependencies:

python -m pip install pytest pytest-asyncio httpx requests

Then open the terminal and run:

python -m pytest test_service_a.py

---

That works perfectly, right?

Well, yes. But remember what I said in the beginning?

To achieve this your first thought would be "let's write end to end tests", but that will slow things down, as each service needs to be up to run the tests, and given that, the setup needed is a bit more complex.

So, what if we want to run the tests without having to run the services?

Patch the HTTP client

We can patch the HTTPX client to make it call the service B, without actually running the service B.

To achieve that, we'll be using RESPX: a simple library, yet powerful, for mocking out the HTTPX and HTTPCore libraries.

It's easy, let me show you! We just need to add a single fixture on the test_service_a.py file:

TestClientAsyncClientRun

import pytest
import pytest_asyncio
import respx
from fastapi.testclient import TestClient

from service_a import app
from service_b import app as app_b


@pytest.fixture()
def client():
    return TestClient(app)


@pytest_asyncio.fixture()
async def create_contracts():
    async with respx.mock:  # (1)!
        respx.route(host="localhost", port=8000).mock(
            side_effect=respx.ASGIHandler(app_b)
        )
        yield


def test_call_b(client: TestClient) -> None:
    response = client.get("/a/call_b")
    assert response.status_code == 200
    assert response.json() == {"b": "b"}

1. The respx.mock context manager is used to mock the HTTPX client.

   Read more about it on the RESPX documentation.

import httpx
import pytest
import pytest_asyncio
import respx

from service_a import app
from service_b import app as app_b


@pytest_asyncio.fixture(name="client")
async def testclient():
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        yield client


@pytest_asyncio.fixture()
async def create_contracts():
    async with respx.mock:  # (1)!
        respx.route(host="localhost", port=8000).mock(
            side_effect=respx.ASGIHandler(app_b)
        )
        yield


@pytest.mark.asyncio
async def test_call_b(client: httpx.AsyncClient) -> None:
    response = await client.get("/a/call_b")
    assert response.status_code == 200
    assert response.json() == {"b": "b"}

1. The respx.mock context manager is used to mock the HTTPX client.

   Read more about it on the RESPX documentation.

Install the dependencies:

python -m pip install pytest pytest-asyncio httpx respx requests

Then open the terminal and run:

python -m pytest test_service_a.py

---

Nice! We did it!

Now, we can run the tests without having to run the services.

If you are a curious person, feel free to compare the tests with the time command:

time python -m pytest test_service_a.py

Be surprised.

Info

You can also read the continuation of this article here.