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

I've been working on a tool called dclient which is a CLI client tool for talking to Datasette instances.

I wanted to write some tests for that tool which would simulate an entire Datasette instance for it to talk to, without starting a localhost server running Datasette itself.

I figured out a pattern for doing that using pytest-httpx to intercept outbound HTTP requests and send them through a Datasette ASGI application instead.

Here's a simplified example of this pattern, with inline comments explaining how it works.

Dependencies are:

pip install pytest pytest-httpx httpx datasette

I saved this as and ran it with pytest

import asyncio
from import Datasette
import httpx
import pytest

def non_mocked_hosts():
    # This ensures that httpx-mock will not affect things once a request
    # starts being processed within Datasette itself via datasette.client
    return ["localhost"]

# Here's an example function we will be testing. This uses httpx.get() to
# make an outbound HTTP request to a Datasette instance - we want to intercept
# that call and handle it ourselves using httpx_mock
def get_versions():
    response = httpx.get("")
    return response.json()

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

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

    # This function will be called every time pytest-httpx intercepts an HTTP request
    def custom_response(request: httpx.Request):
        # Need to run this in async loop, because get_versions uses
        # sync HTTPX and not async HTTPX
        async def run():
            # Here we use ds.client.request() - an internal method within Datasette
            # which can be used to simulate passing an HTTP request through the ASGI app
            # This is why we needed to include localhost in non_mocked_hosts earlier
            response = await ds.client.request(
            # Create a fresh response to avoid an error where stream has been consumed
            response = httpx.Response(
            return response

        return loop.run_until_complete(run())

    # We add custom_response as a callback function for any intercepted HTTP requests:

    # And here's our actual test
    versions = get_versions()
    assert "asgi" in versions
    assert "datasette" in versions
    assert "python" in versions

You can see a much more complex example of this pattern in action in this file:

Created 2023-07-24T21:31:50-07:00 · Edit