Support mocking responses using asgi/wsgi apps

[lundberg]

This PR adds support for mocking HTTPX responses using an ASGI or WSGI app.

Example:

import httpx
import respx
import pytest
from fastapi import FastAPI

app = FastAPI()


@app.get("/baz/")
async def baz():
    return {"ham": "spam"}


async def call_some_remote_api():
    async with httpx.AsyncClient() as client:
        return await client.get("https://foo.bar/baz/")


@pytest.mark.asyncio
@respx.mock
async def test():
    app_route = respx.route(host="foo.bar").mock(side_effect=respx.ASGIHandler(app))
    
    response = await call_some_remote_api()
    
    assert response.json() == {"ham": "spam"}
    assert app_route.called

[adriangb]

This looks like a great start!

I agree with #144 (comment) thought. Maybe respx can accept an httpx Transport implementation? So like:

from httpx import ASGITransport

@pytest.mark.asyncio
@respx.mock
async def test():
    app_route = respx.route(host="foo.bar").mock(transport=ASGITransport(app=app))
 
    response = await call_some_remote_api()
 
    assert response.json() == {"ham": "spam"}
    assert app_route.called

[lundberg]

Maybe respx can accept an httpx Transport implementation?

Exposing transport mocking might be a little bit too low level for the api IMO, also I'd like the .mock(...) to align with python builtin mock spec.

But, I see no reason to not allow mocking a specific transport @adriangb, so I guess we can make the ASGIHandler and WSGIHandler extend some generic transport forwarder, e.g. TransportSideEffect.

Suggested usage:

respx.route(host="example.org").mock(side_effect=TransportSideEffect(my_transport))
# - alternative usage -
route = respx.route(host="example.org")
route.side_effect = TransportSideEffect(my_transport)

[lundberg]

BTW, I've split this branch into #147 to add the support for async side effect separately.

[codecov-commenter]

Codecov Report

Merging #146 (0ddd7c0) into master (717fc1f) will not change coverage.
The diff coverage is 100.00%.

@@            Coverage Diff            @@
##            master      #146   +/-   ##
=========================================
  Coverage   100.00%   100.00%           
=========================================
  Files           16        17    +1     
  Lines         2548      2603   +55     
  Branches       145       145           
=========================================
+ Hits          2548      2603   +55     

Impacted Files	Coverage Δ
respx/handlers.py	100.00% <100.00%> (ø)
tests/test_mock.py	100.00% <100.00%> (ø)

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 717fc1f...0ddd7c0. Read the comment docs.

[lundberg]

@adriangb I've now updated this PR and implemented a generic base transport handler that both ASGI and WSGI handlers makes use of and extend.

Meaning, it allows simple mocking using apps, but also enables lower level use of the transport handlers if one really need to.

Mock using app:

from respx import ASGIHandler
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route


async def baz(request):
    return JSONResponse({"ham": "spam"})


app = Starlette(routes=[Route("/baz/", baz)])
app_mock = respx.mock()
app_route = app_mock.route(host="foo.bar").mock(side_effect=ASGIHandler(app))


@app_mock
async def test():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://foo.bar/baz/")
        assert response.json() == {"ham": "spam"}
        assert app_route.called

Mock using any transport works the same, except passing the TransportHandler or AsyncTransportHandler instead of ASGIHandler in above example.

from respx.handlers import AsyncTransportHandler
...
transport_handler = AsyncTransportHandler(AnyHTTPXAsyncTransport())
app_route = app_mock.route(host="foo.bar").mock(side_effect=transport_handler)
...

[adriangb]

That looks great!

[lundberg]

Fixed #144

[ghandic]

Does this require a async client -> async server?

I would like to make a mock server using fastapi and then the user decides if they want to call it sync or async

[ghandic]

At the moment I see

AssertionError: assert False
where False = isinstance(<coroutine object AsyncTransportHandler.__call__ at 0x1673069c0>, <class 'httpx.Response'>)
where <coroutine object AsyncTransportHandler.__call__ at 0x1673069c0> = <respx.models.ResolvedRoute object at 0x1673cf070>.response
and   <class 'httpx.Response'> = httpx.Response

[ghandic]

In my case I have both synchronous requests using httpx.Client and some using async httpx.AsyncClient - ideally I could just have one mock server to provide both

[lundberg]

Does your sync and async requests hit the same host/endpoint? If not you could add separate respx routes for the sync and async endpoints and mock them with either ASGIHandler or WSGIHandler.

[ghandic]

Yeah that's that I've done in the interim, think it's possible to have a mapping as a new feature?

I need to have flask to support the locking of the sync requests and then fastapi to support the async

Could just have fastapi if we check if the response is awaitable

[lundberg]

So the flask and fastapi apps mimics different services? I guess they're on different hosts or paths then when requesting. I'd say two routes is ideal for that.

There's no current natural place in respx to detect in a side effect if the request is async or not.

[ghandic]

In my case I'm using msal library for my Active Directory validations but they don't support async clients yet, and then I have my own ms graph client that I am using async client for.

It's ok at the moment, but what would be real nice is if those mocks could be portable packages for other projects and sync/async agnostic