Testing Helpers¶
Host-only fakes for testing code that depends on chumicro-requests.
Excluded from every device bundle by name, so they never land on a
microcontroller.
FakeHttpClient¶
In-memory HttpClient stand-in. Same external surface (get / check
/ handle / busy / on_oversized); tests script responses via
enqueue_response() (success) or enqueue_error() (failure). Each
get() pops one scripted entry; the next handle() tick completes
the RequestHandle.
from chumicro_requests.testing import FakeHttpClient
fake = FakeHttpClient()
fake.enqueue_response(status=200, body=b'{"temp_f": 72}')
weather = WeatherFetcher(http_client=fake)
weather.tick(now_ms=0) # internally calls fake.get(...)
weather.tick(now_ms=10) # one handle() tick completes the request
assert weather.last_temperature == 72
assert fake.calls[0].url == "http://api.example.test/weather"
Streamed requests¶
A stream=True request against the fake compresses the real client's
contract into one tick: handle() completes the request, and the
scripted body drains through read_body_into — bytes first, then 0
for end of body.
fake = FakeHttpClient()
fake.enqueue_response(status=200, body=b"blob-bytes")
handle = fake.get("http://example.test/blob", stream=True)
fake.handle(now_ms=0)
buffer = bytearray(4)
assert handle.read_body_into(buffer) == 4 # b"blob"
assert handle.result.streamed is True
assert fake.calls[0].stream is True
fake.cancel() mirrors HttpClient.cancel(): the in-flight handle
finishes with an HttpError and its on_done callback fires.
enqueue_error example¶
from chumicro_requests import HttpTimeoutError
from chumicro_requests.testing import FakeHttpClient
fake = FakeHttpClient()
fake.enqueue_error(HttpTimeoutError("simulated timeout"))
weather = WeatherFetcher(http_client=fake)
weather.tick(now_ms=0)
weather.tick(now_ms=10)
assert weather.last_error_message == "simulated timeout"
Usage from other libraries¶
Libraries that depend on chumicro-requests can import the fake directly in their own test suites:
Libraries that expose injectable services ship their own test fakes alongside the production code, so every consumer uses the same shared fake.
API Reference¶
chumicro_requests.testing
¶
FakeHttpClient plus low-level fixtures for the client test suite.
Public entry points: :class:FakeHttpClient, :func:make_factory,
:func:canned_response, :func:make_client, and :func:drive_until_done.
FakeHttpClient
¶
In-memory :class:HttpClient stand-in for tests.
busy
property
¶
True while a request is in flight (between request method and handle).
enqueue_response(*, status=200, reason='OK', http_version='HTTP/1.1', headers=None, body=b'', oversized_dropped=False)
¶
Script the next request to succeed with this response.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
status
|
int
|
HTTP status code (default 200). |
200
|
reason
|
str
|
Reason phrase (default |
'OK'
|
http_version
|
str
|
Protocol version string (default |
'HTTP/1.1'
|
headers
|
CaseInsensitiveDict | dict | list | tuple | None
|
|
None
|
body
|
bytes
|
Response body bytes (default empty). |
b''
|
oversized_dropped
|
bool
|
Set the matching flag on the response. |
False
|
enqueue_error(error)
¶
Script the next request to fail with error.
request(method, url, *, body=None, json=None, headers=None, timeout_ms=None, max_redirects=None, on_done=None, stream=False)
¶
Mirror :meth:HttpClient.request against the scripted queue.
get(url, *, headers=None, timeout_ms=None, max_redirects=None, on_done=None, stream=False)
¶
Mirror :meth:HttpClient.get against the scripted queue.
post(url, *, body=None, json=None, headers=None, timeout_ms=None, max_redirects=None, on_done=None, stream=False)
¶
Mirror :meth:HttpClient.post against the scripted queue.
put(url, *, body=None, json=None, headers=None, timeout_ms=None, max_redirects=None, on_done=None, stream=False)
¶
Mirror :meth:HttpClient.put against the scripted queue.
patch(url, *, body=None, json=None, headers=None, timeout_ms=None, max_redirects=None, on_done=None, stream=False)
¶
Mirror :meth:HttpClient.patch against the scripted queue.
delete(url, *, headers=None, timeout_ms=None, max_redirects=None, on_done=None, stream=False)
¶
Mirror :meth:HttpClient.delete against the scripted queue.
check(now_ms)
¶
Return True while a request is in flight.
handle(now_ms)
¶
Complete the in-flight request from the scripted outcome.
cancel()
¶
Mirror :meth:HttpClient.cancel: fail the in-flight request.
canned_response(*, status=200, reason='OK', body=b'', extra_headers=())
¶
Build an HTTP/1.1 response byte-string with Content-Length.
make_factory(socket_or_factory)
¶
Return a transport_factory wrapping socket_or_factory in a :class:FakeSocketConnector.
make_client(*, socket_or_factory=None, **kwargs)
¶
Build an :class:HttpClient against FakeTicks + a FakeSocket.
drive_until_done(client, handle, ticks, *, max_ticks=200, advance_ms=1)
¶
Tick client until handle .done flips True.
Raises:
| Type | Description |
|---|---|
AssertionError
|
The handle never completed within max_ticks. |