stario.testing.TestClient runs the HTTP stack in-process — the same status codes, headers, cookies, and bodies as production, without a listening socket. Testing with TestClient has the recipes.
Always use async with TestClient(...) before get, post, request, or stream. Exit runs disconnect, optional app shutdown, and drain_tasks().
Set asyncio_mode = auto under [tool.pytest.ini_options], or mark async tests with @pytest.mark.asyncio.
Pass your production bootstrap for integration tests, or construct a small App for a narrow case. With a bootstrap, client.app is available only inside the context manager. Use TestClient(bootstrap, app_factory=build_app) when tests must not share mutable state on one App. If you load the app with async with aload_app(bootstrap) as app, pass TestClient(app, owns_shutdown=False) so the client does not signal shutdown while aload_app still owns teardown.
There is no url_for or route name=. Build paths with Route(...).href(), or Assets.href() / Files.href().
Buffered requests#
Use await client.get(...), post(...), query(...), or request(...) when the handler finishes in one go. Assert r.status_code, r.text, r.json(), r.headers, r.cookies. r.json() parses with stario.json.loads — see JSON. r.span_id is the root request span id for client.tracer.
from stario.testing import TestClientfrom myapp.app import bootstrap async def test_list_items(): async with TestClient(bootstrap) as client: r = await client.get("/items", params={"page": "2"}) assert r.status_code == 200Streaming responses#
When the handler keeps the connection open, use async with client.stream("GET", path, ...). Read with await r.body(), async for chunk in r.iter_bytes(), or async for ev in r.iter_events(). Leaving the block disconnects the request.
stream does not follow redirects. Send Accept-Encoding: identity when using iter_bytes() on compressed responses.
Background work#
Buffered requests return when the response body is finished. Work scheduled with app.create_task may still run. Call await client.drain_tasks() after the request. Do not call drain_tasks() from inside a coroutine scheduled with app.create_task.
API#
class TestClient(app_or_bootstrap, *, app_factory=None, owns_shutdown=True, base_url='http://testserver', headers=None, cookies=None, follow_redirects=True, max_redirects=20, compression=None, requests=None, request_timeout=30.0)#
Async HTTP client that drives an App through the real server protocol.
Pass a wired App or the bootstrap async generator your program uses in production. Enter async with TestClient(...) first; app is then the live application. Every request is serialized to HTTP/1.1 bytes and fed to the same HttpProtocol that serve() runs, on an in-memory connection. Routing, 308 / 400 target answers, framing, compression, and the request span all come from production code.
Buffered requests — get / post / … return a TestResponse once the handler has finished: the whole body is in content and the request span is closed, so client.tracer.get_span(response.span_id) works right away. Redirects are followed up to max_redirects unless overridden per call.
Streaming — stream yields a TestStreamResponse as soon as headers are in; read with iter_bytes / iter_events. Leaving the block hangs up (the handler sees c.disconnected) and waits for the handler to finish.
timeout / request_timeout cap one exchange. Past it the client hangs up, cancels that handler, and raises TimeoutError.
Exit — open connections close, drain_tasks runs, then bootstrap teardown (if any) matches normal app shutdown.
async TestClient.drain_tasks()#
Wait until App.drain_tasks is quiet and tracer has no open spans.
Called automatically after hanging up every connection when exiting the client context. Unsafe to call from work scheduled via app.create_task (can deadlock).
async TestClient.request(method, url, *, params=None, headers=None, cookies=None, json=None, data=None, files=None, content=None, follow_redirects=None, timeout=None)#
Issue an arbitrary HTTP method and return a fully buffered TestResponse.
TestClient.stream(method, url, *, params=None, headers=None, cookies=None, json=None, data=None, files=None, content=None, timeout=None)#
Start a request and yield TestStreamResponse once headers are available.
Does not follow redirects (inspect Location yourself). Leaving the block hangs up this connection and waits for the handler coroutine. timeout caps the whole exchange, including body iteration after headers. Send Accept-Encoding: identity when reading iter_bytes so the body is not compressed.
async TestClient.get(url, **kwargs)#
async TestClient.head(url, **kwargs)#
async TestClient.post(url, **kwargs)#
async TestClient.put(url, **kwargs)#
async TestClient.patch(url, **kwargs)#
async TestClient.delete(url, **kwargs)#
async TestClient.options(url, **kwargs)#
async TestClient.query(url, **kwargs)#
class TestResponse(status_code, url, headers, content, request, span_id, history=<factory>, cookies=<factory>)#
Buffered HTTP result from TestClient.get, TestClient.post, etc.
Fields
- status_code(int):—
- url(str):—
- headers(Headers):—
- content(bytes):—
- request(ClientRequest):—
- span_id(UUID):—
- history(list[TestResponse]):default:
factory - cookies(dict[str, str]):default:
factory
The body is fully buffered; Content-Encoding is decoded when present. Pair span_id with TestClient.tracer for assertions.
TestResponse.json()#
TestResponse.raise_for_status()#
class TestStreamResponse(status_code, url, headers, request, span_id, cookies, _exchange, _deadline=None)#
Streaming result from TestClient.stream once response headers exist.
Fields
- status_code(int):—
- url(str):—
- headers(Headers):—
- request(ClientRequest):—
- span_id(UUID):—
- cookies(dict[str, str]):—
- _exchange(Exchange):—
- _deadline(float | None):default:
None
Use body, iter_bytes, or iter_events to read the entity body as the handler writes it. Leaving the stream context hangs up the connection (the handler sees c.disconnected) and waits for the handler to finish.
async TestStreamResponse.body()#
Concatenate all body chunks after they arrive.
TestStreamResponse.disconnect()#
Hang up now instead of at the end of the stream block.
async TestStreamResponse.iter_bytes()#
Yield body chunks as the handler writes them (framing removed).
async TestStreamResponse.iter_events()#
Parse text/event-stream into dicts (keys such as event, id, data).
class TestTracer()#
Test-side view of telemetry for TestClient.
Implements the stario.telemetry.Tracer protocol for request dispatch. Assertions in tests should use the query helpers below; they only return finished span snapshots.
TestTracer.create(name, attributes=None, /, *, parent=None)#
TestTracer.find_span(name, *, root_id=None, parent_id=None)#
First finished span named name, in start-time order.
When several requests reuse the same span name, pass root_id=r.span_id (from the matching TestResponse) so you match the right subtree. parent_id requires an exact parent link.
TestTracer.finished_spans()#
Finished spans in start-time order.
TestTracer.get_event(span_id, event_name, *, index=0)#
The index-th TelemetryEvent named event_name, or None.
TestTracer.get_events(span_id, *, name=None)#
Events recorded on a finished span; filter with name when set.
TestTracer.get_span(span_id)#
Return the finished TelemetrySpan for span_id, or None.
Open spans and unknown ids yield None.
TestTracer.has_attribute(span_id, key, value=<object object at 0x738ad0f99160>)#
Whether the finished span's attributes include key.
Pass value to require equality; omit it to assert presence only.
TestTracer.has_event(span_id, event_name)#
Whether get_events(span_id) contains at least one event_name.
TestTracer.has_open_spans()#
True while any span created through this tracer has not been ended.
TestTracer.on_end(span)#
TestTracer.stats()#
aload_app(bootstrap, *, app_factory=None, tracer=None)#
Load bootstrap like production; emits server.startup / server.shutdown spans.
Pass tracer to share one TestTracer with TestClient (same instance as client.tracer when the client wires bootstrap).