rogulski.it

My name is Piotr, a passionate pythonista and this is my blog!

Pytest with respx and vcr! The best way to test remote service requests

Posted at — Oct 24, 2021

Stop mocking remote service httpx requests by yourself! There is a better, automatic way to do it.

Thanks to my lovely friend prmtl I have learned the best way for testing remote service requests in Python.

Requirements

Testing httpx with respx (pytest)

First, I would like to present a simple case to use respx, yet powerful, a utility for mocking out the HTTPX, and HTTP Core, libraries according to the documentation.

For example and explanation let’s implement a fake remote service which will call a random API generator and return it:

from typing import Dict, Optional, Any

import httpx


class FakerService:
    base_url: str = "https://fakerapi.it"

    @classmethod
    async def request(
        cls,
        method: str,
        endpoint: str,
        headers: Optional[Dict[str, str]] = None,
        timeout: int = 30,
        **kwargs: Any,
    ) -> Dict[str, Any]:

        url = f"{cls.base_url}{endpoint}"
        async with httpx.AsyncClient(headers=headers) as client:
            response = await client.request(
                method=method, url=url, timeout=timeout, **kwargs
            )

        response.raise_for_status()
        return response.json()

    @classmethod
    async def get(
        cls, endpoint: str, headers: Optional[Dict[str, str]] = None, **kwargs: Any
    ) -> Dict[str, Any]:
        return await cls.request("get", endpoint, headers, **kwargs)

This class includes two methods (easily extensible for more HTTP methods like posts, puts, etc…). request method creates httpx.AsyncClient client instance and performs a request to the remote service using async/await. Then response status code is validated, and the JSON is returned. Simple but very usable.

import pytest
import respx
from httpx import Response

from remotes import FakerService

pytestmark = pytest.mark.asyncio


@pytest.mark.respx(base_url=FakerService.url)
async def test_faker_service_get_using_respx(respx_mock: respx.MockRouter) -> None:

    respx_mock.get("/api/v1/books?_quantity=1").mock(
        return_value=Response(
            200,
            json={
                "title": "First, because I'm.",
                "author": "Enola Lang",
                "genre": "Iste",
            },
        )
    )

    response = await FakerService.get("/api/v1/books?_quantity=1")
    assert response == {
        "title": "First, because I'm.",
        "author": "Enola Lang",
        "genre": "Iste",
    }

Using pytest, we can use mark with respx to mock a base URL for our remote service, it improves the readability and we do not repeat ourselves in the code every time we are performing a request in the code. Helpful when more than one call is made.

For a better pytest experience, it is recommended to use respx_mock fixture that is a respx.MockRouter object which is used to mock our response.

When the FakerService.get method is executed, respx checks which URLs were mocked, and if it finds a proper path it is mocking it. Otherwise, response not mocked error is raised:

respx.models.AllMockedAssertionError: RESPX: <Request('GET', 'https://fakerapi.it/api/v1/not-mocked')> not mocked!

Testing httpx with vcr

VCR is a tool used to simplify testing HTTP requests by recording requests and responses made to remote services. It can be configured to run it once or more. It automates the whole process by performing a recording and saving it to the yaml or json file. When the test is started recorded response is automatically mocked by its URL and test case name and injected into the results.

For our case we will use vcrpy with a pytest plugin: pytest-vcr.

Configuration and usage

A plugin can be configured due to the project’s needs. Output directory, filter headers and more can be modified. In the example we will use most of the default configs, saving the recorded file to the default directory cassettes/ in yaml format.

from typing import Any, Dict

import pytest


pytestmark = pytest.mark.asyncio

@pytest.fixture(scope="module")
def vcr_config() -> Dict[str, Any]:
    return {
        "filter_headers": ["authorization", "host"],
        "ignore_localhost": True,
        "record_mode": "once",
    }


@pytest.mark.vcr()
async def test_faker_service_get() -> None:
    response = await FakerService.get(
        "/api/v1/books?_quantity=1", params={"param1": "qwerty"}
    )

    assert response == {
        "status": "OK",
        "code": 200,
        "total": 1,
        "data": [
            {
                "status": "OK",
                "code": 200,
                "total": 1,
                "data": [
                    {
                        "title": "Cat. 'Do you play.",
                        "author": "Mervin Zulauf",
                        "genre": "Ad",
                        "description": (
                            "I've got to the porpoise, \"Keep back, please: we don't want to be?'"
                            "it asked. 'Oh, I'm not myself, you see.' 'I don't quite understand you,'"
                            " she said, without opening its eyes, 'Of course, of."
                        ),
                        "isbn": "9782943949905",
                        "image": "http://placeimg.com/480/640/any",
                        "published": "1998-09-03",
                        "publisher": "Ut Sapiente",
                    }
                ],
            }
        ],
    }

All which needs to be done is to decorate a test function with @pytest.mark.vcr() decorator, it will automatically detect all requests done in a test case and record it

When we will run the test for the first time pytest -k test_faker_service_get request to the external service will be performed, new directory cassettes will be created with a new file named the same as test function: test_faker_service_get.yaml.

The file will include all important information retrieved from the remote service:

interactions:
- request:
    body: ''
    headers:
      accept:
      - '*/*'
      accept-encoding:
      - gzip, deflate
      connection:
      - keep-alive
      user-agent:
      - python-httpx/0.20.0
    method: GET
    uri: https://fakerapi.it/api/v1/books?_quantity=1&param1=qwerty
  response:
    content: '{"status":"OK","code":200,"total":1,"data":[{"title":"Cat. ''Do you
      play.","author":"Mervin Zulauf","genre":"Ad","description":"I''ve got to the
      porpoise, \"Keep back, please: we don''t want to be?'' it asked. ''Oh, I''m
      not myself, you see.'' ''I don''t quite understand you,'' she said, without
      opening its eyes, ''Of course, of.","isbn":"9782943949905","image":"http:\/\/placeimg.com\/480\/640\/any","published":"1998-09-03","publisher":"Ut
      Sapiente"}]}'
    headers:
      Access-Control-Allow-Credentials:
      - 'true'
      Access-Control-Allow-Headers:
      - Content-Type, Authorization, X-Requested-With
      Access-Control-Allow-Methods:
      - GET
      Access-Control-Allow-Origin:
      - '*'
      Access-Control-Max-Age:
      - '86400'
      Cache-Control:
      - no-cache, private
      Connection:
      - keep-alive
      Content-Encoding:
      - gzip
      Content-Type:
      - application/json
      Date:
      - Sat, 23 Oct 2021 11:55:56 GMT
      Server:
      - nginx
      Transfer-Encoding:
      - chunked
      Vary:
      - Accept-Encoding
      X-Powered-By:
      - PHP/7.3.16
      X-RateLimit-Limit:
      - '30'
      X-RateLimit-Remaining:
      - '29'
      X-UA-Compatible:
      - IE=Edge,chrome=1
    http_version: HTTP/1.1
    status_code: 200
version: 1

It holds all request information, including status code, headers, body and more. Please check vcr docs for more details.

It is also worth mentioning that the same config works for multiple requests created in the same test case. It will automatically append more requests details to the same file per test function name. Feel free to check this example on my github.

Summary

I’ve observed that in most cases devs are mocking the whole response in the code, using custom fixtures and json responses (including me before I found this approach). A shared case may speed up some processes and basic configurations with improvement on the readability and human error.