Using pytest-httpx to run intercepted requests through an in-memory Datasette instance#

1
import asyncio
2
from datasette.app import Datasette
3
import httpx
4
import pytest
5

6

7
@pytest.fixture
8
def non_mocked_hosts():
9
    # This ensures that httpx-mock will not affect things once a request
10
    # starts being processed within Datasette itself via datasette.client
11
    return ["localhost"]
12

13

14
# Here's an example function we will be testing. This uses httpx.get() to
15
# make an outbound HTTP request to a Datasette instance - we want to intercept
16
# that call and handle it ourselves using httpx_mock
17
def get_versions():
18
    response = httpx.get("https://datasette.example.com/-/versions.json")
19
    response.raise_for_status()
20
    return response.json()
21

22

23
# The httpx_mock fixtures comes from pytest-httpx
24
def test_get_version(httpx_mock):
25
    ds = Datasette()
26

27
    # This test is a regular function, but we need to be able to make some
28
    # await... calls later on - so we need the event loop to run them against
29
    loop = asyncio.get_event_loop()
30

31
    # This function will be called every time pytest-httpx intercepts an HTTP request
32
    def custom_response(request: httpx.Request):
33
        # Need to run this in async loop, because get_versions uses
34
        # sync HTTPX and not async HTTPX
35
        async def run():
36
            # Here we use ds.client.request() - an internal method within Datasette
37
            # which can be used to simulate passing an HTTP request through the ASGI app
38
            # This is why we needed to include localhost in non_mocked_hosts earlier
39
            response = await ds.client.request(
40
                request.method,
41
                request.url.path,
42
                content=request.read(),
43
                headers=request.headers,
44
            )
45
            # Create a fresh response to avoid an error where stream has been consumed
46
            response = httpx.Response(
47
                status_code=response.status_code,
48
                headers=response.headers,
49
                content=response.content,
50
            )
51
            return response
52

53
        return loop.run_until_complete(run())
54

55
    # We add custom_response as a callback function for any intercepted HTTP requests:
56
    httpx_mock.add_callback(custom_response)
57

58
    # And here's our actual test
59
    versions = get_versions()
60
    assert "asgi" in versions
61
    assert "datasette" in versions
62
    assert "python" in versions