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.