Add httpx.MockTransport() (#1401)

* Add httpx.MockTransport

* Add docs on MockTransport

* Add pointer to RESPX

* Add note on pytest-httpx

* Tweak existing docs example to use 'httpx.MockTransport'

Co-authored-by: Florimond Manca <[email protected]>

docs/advanced.md

@@ -1062,6 +1062,31 @@ Which we can use in the same way:
 {"text": "Hello, world!"}
 ```
 
+### Mock transports
+
+During testing it can often be useful to be able to mock out a transport,
+and return pre-determined responses, rather than making actual network requests.
+
+The `httpx.MockTransport` class accepts a handler function, which can be used
+to map requests onto pre-determined responses:
+
+```python
+def handler(request):
+    return httpx.Response(200, json={"text": "Hello, world!"})
+
+
+# Switch to a mock transport, if the TESTING environment variable is set.
+if os.environ['TESTING'].upper() == "TRUE":
+    transport = httpx.MockTransport(handler)
+else:
+    transport = httpx.HTTPTransport()
+
+client = httpx.Client(transport=transport)
+```
+
+For more advanced use-cases you might want to take a look at either [the third-party
+mocking library, RESPX](https://lundberg.github.io/respx/), or the [pytest-httpx library](https://github.com/Colin-b/pytest_httpx).
+
 ### Mounting transports
 
 You can also mount transports against given schemes or domains, to control
@@ -1101,7 +1126,10 @@ Mocking requests to a given domain:
 ```python
 # All requests to "example.org" should be mocked out.
 # Other requests occur as usual.
-mounts = {"all://example.org": MockTransport()}
+def handler(request):
+    return httpx.Response(200, json={"text": "Hello, World!"})
+
+mounts = {"all://example.org": httpx.MockTransport(handler)}
 client = httpx.Client(mounts=mounts)
 ```
 

httpx/__init__.py

@@ -36,6 +36,7 @@
 from ._models import URL, Cookies, Headers, QueryParams, Request, Response
 from ._status_codes import StatusCode, codes
 from ._transports.asgi import ASGITransport
+from ._transports.mock import MockTransport
 from ._transports.wsgi import WSGITransport
 
 __all__ = [
@@ -65,6 +66,7 @@
     "InvalidURL",
     "Limits",
     "LocalProtocolError",
+    "MockTransport",
     "NetworkError",
     "options",
     "patch",

httpx/_transports/mock.py

@@ -0,0 +1,56 @@
+from typing import Callable, List, Optional, Tuple
+
+import httpcore
+
+from .._models import Request
+
+
+class MockTransport(httpcore.SyncHTTPTransport, httpcore.AsyncHTTPTransport):
+    def __init__(self, handler: Callable) -> None:
+        self.handler = handler
+
+    def request(
+        self,
+        method: bytes,
+        url: Tuple[bytes, bytes, Optional[int], bytes],
+        headers: List[Tuple[bytes, bytes]] = None,
+        stream: httpcore.SyncByteStream = None,
+        ext: dict = None,
+    ) -> Tuple[int, List[Tuple[bytes, bytes]], httpcore.SyncByteStream, dict]:
+        request = Request(
+            method=method,
+            url=url,
+            headers=headers,
+            stream=stream,
+        )
+        request.read()
+        response = self.handler(request)
+        return (
+            response.status_code,
+            response.headers.raw,
+            response.stream,
+            response.ext,
+        )
+
+    async def arequest(
+        self,
+        method: bytes,
+        url: Tuple[bytes, bytes, Optional[int], bytes],
+        headers: List[Tuple[bytes, bytes]] = None,
+        stream: httpcore.AsyncByteStream = None,
+        ext: dict = None,
+    ) -> Tuple[int, List[Tuple[bytes, bytes]], httpcore.AsyncByteStream, dict]:
+        request = Request(
+            method=method,
+            url=url,
+            headers=headers,
+            stream=stream,
+        )
+        await request.aread()
+        response = self.handler(request)
+        return (
+            response.status_code,
+            response.headers.raw,
+            response.stream,
+            response.ext,
+        )

tests/client/test_async_client.py

@@ -5,7 +5,6 @@
 import pytest
 
 import httpx
-from tests.utils import MockTransport
 
 
 @pytest.mark.usefixtures("async_environment")
@@ -247,7 +246,7 @@ def hello_world(request):
 
 @pytest.mark.usefixtures("async_environment")
 async def test_client_closed_state_using_implicit_open():
-    client = httpx.AsyncClient(transport=MockTransport(hello_world))
+    client = httpx.AsyncClient(transport=httpx.MockTransport(hello_world))
 
     assert not client.is_closed
     await client.get("http://example.com")
@@ -262,7 +261,7 @@ async def test_client_closed_state_using_implicit_open():
 
 @pytest.mark.usefixtures("async_environment")
 async def test_client_closed_state_using_with_block():
-    async with httpx.AsyncClient(transport=MockTransport(hello_world)) as client:
+    async with httpx.AsyncClient(transport=httpx.MockTransport(hello_world)) as client:
         assert not client.is_closed
         await client.get("http://example.com")
 
@@ -273,7 +272,7 @@ async def test_client_closed_state_using_with_block():
 
 @pytest.mark.usefixtures("async_environment")
 async def test_deleting_unclosed_async_client_causes_warning():
-    client = httpx.AsyncClient(transport=MockTransport(hello_world))
+    client = httpx.AsyncClient(transport=httpx.MockTransport(hello_world))
     await client.get("http://example.com")
     with pytest.warns(UserWarning):
         del client
@@ -291,8 +290,8 @@ def mounted(request: httpx.Request) -> httpx.Response:
 
 @pytest.mark.usefixtures("async_environment")
 async def test_mounted_transport():
-    transport = MockTransport(unmounted)
-    mounts = {"custom://": MockTransport(mounted)}
+    transport = httpx.MockTransport(unmounted)
+    mounts = {"custom://": httpx.MockTransport(mounted)}
 
     async with httpx.AsyncClient(transport=transport, mounts=mounts) as client:
         response = await client.get("https://www.example.com")