Skip to content

Mock Transport API

For high-level usage see the testing guide.

from hyperhttp import MockTransport, MockResponse, Router

MockResponse

A lightweight response spec used by MockTransport handlers.

You can pick one of content, text, json and stream. Passing more than one raises ValueError — there's no good way to reconcile multiple bodies.

Parameters

status_code: HTTP status code, e.g. 200, 404. headers: Any headers-like input (dict, list of pairs, Headers). content: Raw bytes body. Content-Length is set automatically. text: Unicode body; encoded as UTF-8. Content-Type defaults to text/plain; charset=utf-8. json: Any JSON-serialisable object. Content-Type defaults to application/json. Use the sentinel MockResponse.NO_BODY (the default) to distinguish "no JSON" from json=None. stream: Async iterable yielding bytes. Use this to simulate streamed or chunked-encoded bodies. Content-Length is not auto-computed. http_version: e.g. "HTTP/1.1" or "HTTP/2".

Source code in hyperhttp/mock.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class MockResponse:
    """A lightweight response spec used by ``MockTransport`` handlers.

    You can pick *one* of ``content``, ``text``, ``json`` and ``stream``.
    Passing more than one raises ``ValueError`` — there's no good way to
    reconcile multiple bodies.

    Parameters
    ----------
    status_code:
        HTTP status code, e.g. ``200``, ``404``.
    headers:
        Any headers-like input (dict, list of pairs, ``Headers``).
    content:
        Raw bytes body. Content-Length is set automatically.
    text:
        Unicode body; encoded as UTF-8. Content-Type defaults to
        ``text/plain; charset=utf-8``.
    json:
        Any JSON-serialisable object. Content-Type defaults to
        ``application/json``. Use the sentinel ``MockResponse.NO_BODY``
        (the default) to distinguish "no JSON" from ``json=None``.
    stream:
        Async iterable yielding bytes. Use this to simulate streamed or
        chunked-encoded bodies. Content-Length is *not* auto-computed.
    http_version:
        e.g. ``"HTTP/1.1"`` or ``"HTTP/2"``.
    """

    __slots__ = (
        "status_code",
        "http_version",
        "_headers",
        "_body",
        "_stream",
    )

    def __init__(
        self,
        status_code: int = 200,
        *,
        headers: HeadersInput = None,
        content: Optional[Union[bytes, bytearray, memoryview]] = None,
        text: Optional[str] = None,
        json: Any = _UNSET,
        stream: Optional[AsyncIterator[bytes]] = None,
        http_version: str = "HTTP/1.1",
    ) -> None:
        self.status_code = int(status_code)
        self.http_version = http_version

        body: Optional[bytes] = None
        default_content_type: Optional[str] = None

        body_sources = sum(
            x is not None and x is not _UNSET
            for x in (content, text, stream)
        ) + (1 if json is not _UNSET else 0)
        if body_sources > 1:
            raise ValueError(
                "MockResponse accepts at most one of content=/text=/json=/stream="
            )

        if content is not None:
            body = bytes(content)
        elif text is not None:
            body = text.encode("utf-8")
            default_content_type = "text/plain; charset=utf-8"
        elif json is not _UNSET:
            body = json_dumps(json)
            default_content_type = "application/json"

        self._body = body
        self._stream = stream

        hdrs = Headers(headers) if headers is not None else Headers()
        if body is not None and "content-length" not in hdrs:
            hdrs["content-length"] = str(len(body))
        if default_content_type and "content-type" not in hdrs:
            hdrs["content-type"] = default_content_type
        self._headers = hdrs

    def __repr__(self) -> str:
        return f"MockResponse({self.status_code})"

    def _build_raw(self) -> RawResponse:
        if self._stream is not None:
            stream: AsyncIterator[bytes] = self._stream
        else:
            body = self._body if self._body is not None else b""
            stream = _single_chunk_stream(body)
        return RawResponse(
            status_code=self.status_code,
            http_version=self.http_version,
            headers=self._headers,
            stream=stream,
        )

Attributes

__slots__ = ('status_code', 'http_version', '_headers', '_body', '_stream') class-attribute instance-attribute

_body = body instance-attribute

_headers = hdrs instance-attribute

_stream = stream instance-attribute

http_version = http_version instance-attribute

status_code = int(status_code) instance-attribute

Functions

__init__(status_code=200, *, headers=None, content=None, text=None, json=_UNSET, stream=None, http_version='HTTP/1.1')

Source code in hyperhttp/mock.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def __init__(
    self,
    status_code: int = 200,
    *,
    headers: HeadersInput = None,
    content: Optional[Union[bytes, bytearray, memoryview]] = None,
    text: Optional[str] = None,
    json: Any = _UNSET,
    stream: Optional[AsyncIterator[bytes]] = None,
    http_version: str = "HTTP/1.1",
) -> None:
    self.status_code = int(status_code)
    self.http_version = http_version

    body: Optional[bytes] = None
    default_content_type: Optional[str] = None

    body_sources = sum(
        x is not None and x is not _UNSET
        for x in (content, text, stream)
    ) + (1 if json is not _UNSET else 0)
    if body_sources > 1:
        raise ValueError(
            "MockResponse accepts at most one of content=/text=/json=/stream="
        )

    if content is not None:
        body = bytes(content)
    elif text is not None:
        body = text.encode("utf-8")
        default_content_type = "text/plain; charset=utf-8"
    elif json is not _UNSET:
        body = json_dumps(json)
        default_content_type = "application/json"

    self._body = body
    self._stream = stream

    hdrs = Headers(headers) if headers is not None else Headers()
    if body is not None and "content-length" not in hdrs:
        hdrs["content-length"] = str(len(body))
    if default_content_type and "content-type" not in hdrs:
        hdrs["content-type"] = default_content_type
    self._headers = hdrs

__repr__()

Source code in hyperhttp/mock.py 
152
153
def __repr__(self) -> str:
    return f"MockResponse({self.status_code})"

_build_raw()

Source code in hyperhttp/mock.py 
155
156
157
158
159
160
161
162
163
164
165
166
def _build_raw(self) -> RawResponse:
    if self._stream is not None:
        stream: AsyncIterator[bytes] = self._stream
    else:
        body = self._body if self._body is not None else b""
        stream = _single_chunk_stream(body)
    return RawResponse(
        status_code=self.status_code,
        http_version=self.http_version,
        headers=self._headers,
        stream=stream,
    )

MockTransport

Bases: Transport

In-memory transport for tests.

Pass any of:

  • a callable handler(request) -> MockResponse (sync or async);
  • a single MockResponse — every request gets the same reply;
  • a sequence [MockResponse, ...] — popped left-to-right per call. Runs out → IndexError (loud failure beats silent reuse);
  • a sequence that cycles — wrap it with itertools.cycle first if you want repeats.

A handler may also raise any hyperhttp exception (or OSError) to simulate transport-level failures, which flows through the client's retry / circuit-breaker machinery exactly as the real thing would.

Source code in hyperhttp/mock.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
class MockTransport(Transport):
    """In-memory transport for tests.

    Pass **any** of:

    * a **callable** ``handler(request) -> MockResponse`` (sync or async);
    * a **single** ``MockResponse`` — every request gets the same reply;
    * a **sequence** ``[MockResponse, ...]`` — popped left-to-right per
      call. Runs out → ``IndexError`` (loud failure beats silent reuse);
    * a **sequence that cycles** — wrap it with ``itertools.cycle`` first
      if you want repeats.

    A handler may also raise any ``hyperhttp`` exception (or ``OSError``)
    to simulate transport-level failures, which flows through the client's
    retry / circuit-breaker machinery exactly as the real thing would.
    """

    http_version = "HTTP/1.1"

    def __init__(
        self,
        handler: Union[
            Handler,
            MockResponse,
            Sequence[MockResponse],
            Mapping[str, MockResponse],
        ],
    ) -> None:
        self._handler = _normalise_handler(handler)
        self._calls: List[Any] = []
        self._closed = False

    # -- assertion API -----------------------------------------------------

    @property
    def calls(self) -> List[Any]:
        """List of every ``Request`` this transport has served (FIFO)."""
        return self._calls

    @property
    def call_count(self) -> int:
        return len(self._calls)

    @property
    def last_request(self) -> Optional[Any]:
        return self._calls[-1] if self._calls else None

    def reset(self) -> None:
        """Forget all recorded calls. The handler is unaffected."""
        self._calls.clear()

    # -- Transport protocol ------------------------------------------------

    @property
    def closed(self) -> bool:
        return self._closed

    @property
    def reusable(self) -> bool:
        return not self._closed

    @property
    def in_flight(self) -> int:
        return 0

    @property
    def max_concurrent(self) -> int:
        return 1 << 30  # unlimited for tests

    @property
    def host_port(self) -> str:
        return "mock"

    async def aclose(self) -> None:
        self._closed = True

    async def handle_request(
        self,
        *,
        method: str,
        url: URL,
        headers: Headers,
        body: Any,
        timeout: Optional[float],
    ) -> RawResponse:
        # Build a lightweight Request-shaped object so handlers get the
        # nice ``request.method`` / ``request.url`` / ``request.headers``
        # / ``request.content`` accessors they'd see in real code.
        from hyperhttp.client import Request  # local import — avoid cycle

        request = Request(method=method, url=url, headers=headers, content=body)
        self._calls.append(request)

        result = self._handler(request)
        if inspect.isawaitable(result):
            result = await result
        response = _coerce_response(result)
        return response._build_raw()

Attributes

_calls = [] instance-attribute

_closed = False instance-attribute

_handler = _normalise_handler(handler) instance-attribute

call_count property

calls property

List of every Request this transport has served (FIFO).

closed property

host_port property

http_version = 'HTTP/1.1' class-attribute instance-attribute

in_flight property

last_request property

max_concurrent property

reusable property

Functions

__init__(handler)

Source code in hyperhttp/mock.py 
198
199
200
201
202
203
204
205
206
207
208
209
def __init__(
    self,
    handler: Union[
        Handler,
        MockResponse,
        Sequence[MockResponse],
        Mapping[str, MockResponse],
    ],
) -> None:
    self._handler = _normalise_handler(handler)
    self._calls: List[Any] = []
    self._closed = False

aclose() async

Source code in hyperhttp/mock.py 
252
253
async def aclose(self) -> None:
    self._closed = True

handle_request(*, method, url, headers, body, timeout) async

Source code in hyperhttp/mock.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
async def handle_request(
    self,
    *,
    method: str,
    url: URL,
    headers: Headers,
    body: Any,
    timeout: Optional[float],
) -> RawResponse:
    # Build a lightweight Request-shaped object so handlers get the
    # nice ``request.method`` / ``request.url`` / ``request.headers``
    # / ``request.content`` accessors they'd see in real code.
    from hyperhttp.client import Request  # local import — avoid cycle

    request = Request(method=method, url=url, headers=headers, content=body)
    self._calls.append(request)

    result = self._handler(request)
    if inspect.isawaitable(result):
        result = await result
    response = _coerce_response(result)
    return response._build_raw()

reset()

Forget all recorded calls. The handler is unaffected.

Source code in hyperhttp/mock.py 
226
227
228
def reset(self) -> None:
    """Forget all recorded calls. The handler is unaffected."""
    self._calls.clear()

Router

Method + path-prefix dispatcher.

Tiny on purpose — the point of MockTransport is that you write your own handler. This helper just exists so the 90 % case ("GET /users returns 200, everything else 404") is a one-liner:

router = Router()
router.get("/users",        lambda req: MockResponse(200, json=[...]))
router.post("/users",       lambda req: MockResponse(201))
router.route("GET", "/health", lambda req: MockResponse(204))

mock = MockTransport(router)

Matching is exact on (method, url.path). For anything fancier (wildcards, path-parameters), dispatch inside your own handler. Unmatched requests return MockResponse(404).

Source code in hyperhttp/mock.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
class Router:
    """Method + path-prefix dispatcher.

    Tiny on purpose — the point of ``MockTransport`` is that you write your
    own handler. This helper just exists so the 90 % case ("GET /users
    returns 200, everything else 404") is a one-liner:

        router = Router()
        router.get("/users",        lambda req: MockResponse(200, json=[...]))
        router.post("/users",       lambda req: MockResponse(201))
        router.route("GET", "/health", lambda req: MockResponse(204))

        mock = MockTransport(router)

    Matching is **exact** on ``(method, url.path)``. For anything fancier
    (wildcards, path-parameters), dispatch inside your own handler.
    Unmatched requests return ``MockResponse(404)``.
    """

    __slots__ = ("_routes", "_default")

    def __init__(
        self,
        *,
        default: Optional[Union[Handler, MockResponse, int]] = None,
    ) -> None:
        self._routes: dict = {}
        self._default: Handler = (
            _normalise_handler(default)
            if default is not None
            else (lambda _req: MockResponse(404))
        )

    def route(self, method: str, path: str, handler: Handler) -> "Router":
        self._routes[(method.upper(), path)] = _wrap_handler(handler)
        return self

    def get(self, path: str, handler: Handler) -> "Router":
        return self.route("GET", path, handler)

    def post(self, path: str, handler: Handler) -> "Router":
        return self.route("POST", path, handler)

    def put(self, path: str, handler: Handler) -> "Router":
        return self.route("PUT", path, handler)

    def patch(self, path: str, handler: Handler) -> "Router":
        return self.route("PATCH", path, handler)

    def delete(self, path: str, handler: Handler) -> "Router":
        return self.route("DELETE", path, handler)

    def head(self, path: str, handler: Handler) -> "Router":
        return self.route("HEAD", path, handler)

    def options(self, path: str, handler: Handler) -> "Router":
        return self.route("OPTIONS", path, handler)

    def __call__(self, request: Any) -> Any:
        handler = self._routes.get((request.method, request.url.path))
        if handler is None:
            return self._default(request)
        return handler(request)

Attributes

__slots__ = ('_routes', '_default') class-attribute instance-attribute

_default = _normalise_handler(default) if default is not None else (lambda _req: MockResponse(404)) instance-attribute

_routes = {} instance-attribute

Functions

__call__(request)

Source code in hyperhttp/mock.py 
342
343
344
345
346
def __call__(self, request: Any) -> Any:
    handler = self._routes.get((request.method, request.url.path))
    if handler is None:
        return self._default(request)
    return handler(request)

__init__(*, default=None)

Source code in hyperhttp/mock.py 
305
306
307
308
309
310
311
312
313
314
315
def __init__(
    self,
    *,
    default: Optional[Union[Handler, MockResponse, int]] = None,
) -> None:
    self._routes: dict = {}
    self._default: Handler = (
        _normalise_handler(default)
        if default is not None
        else (lambda _req: MockResponse(404))
    )

delete(path, handler)

Source code in hyperhttp/mock.py 
333
334
def delete(self, path: str, handler: Handler) -> "Router":
    return self.route("DELETE", path, handler)

get(path, handler)

Source code in hyperhttp/mock.py 
321
322
def get(self, path: str, handler: Handler) -> "Router":
    return self.route("GET", path, handler)

head(path, handler)

Source code in hyperhttp/mock.py 
336
337
def head(self, path: str, handler: Handler) -> "Router":
    return self.route("HEAD", path, handler)

options(path, handler)

Source code in hyperhttp/mock.py 
339
340
def options(self, path: str, handler: Handler) -> "Router":
    return self.route("OPTIONS", path, handler)

patch(path, handler)

Source code in hyperhttp/mock.py 
330
331
def patch(self, path: str, handler: Handler) -> "Router":
    return self.route("PATCH", path, handler)

post(path, handler)

Source code in hyperhttp/mock.py 
324
325
def post(self, path: str, handler: Handler) -> "Router":
    return self.route("POST", path, handler)

put(path, handler)

Source code in hyperhttp/mock.py 
327
328
def put(self, path: str, handler: Handler) -> "Router":
    return self.route("PUT", path, handler)

route(method, path, handler)

Source code in hyperhttp/mock.py 
317
318
319
def route(self, method: str, path: str, handler: Handler) -> "Router":
    self._routes[(method.upper(), path)] = _wrap_handler(handler)
    return self