Testing

Run the HTTP stack in-process with stario.testing.TestClient: same status codes, headers, cookies, and bodies as production, without a listening socket.

Always use async with TestClient(...) before calling get, post, request, or stream. Exit runs disconnect, optional app shutdown, and drain_tasks().

Practical recipes: Testing with TestClient.

pytest-asyncio: set asyncio_mode = auto under [tool.pytest.ini_options], or mark async tests with @pytest.mark.asyncio.

Bootstrap vs inline App: pass your production bootstrap for integration tests, or construct a small App in the test for narrow cases. 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 — put routes in build_app(), use bootstrap for wiring and teardown. 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.

URLs: Stario 4 has no url_for or route name=. Build paths with UrlPath(...).href() or AssetManifest.href() the same way application code does.

1. Request and response (buffered)

Use await client.get(...), post(...), or request(...) when the handler finishes the exchange in one go. Assert request shape (params, json, headers, cookies) and response (r.status_code, r.text, r.json(), r.headers, r.cookies).

from stario.routing import UrlPath

from myapp.app import bootstrap

from stario.testing import TestClient

items = UrlPath("/items")

async def test_list_items():

    async with TestClient(bootstrap) as client:

        r = await client.get(items.href(), params={"page": "2"})

        assert r.status_code == 200

        assert "items" in r.json()

async def test_create_item():

    async with TestClient(bootstrap) as client:

        r = await client.post("/items", json={"name": "cup"})

        assert r.status_code == 201

        assert r.json()["name"] == "cup"

For a tiny app without bootstrap:

from stario import App, Context, Writer

from stario.testing import TestClient

async def test_ping_inline():

    app = App()

    async def ping(c: Context, w: Writer) -> None:

        w.write_headers(200)

        w.end()

    app.get("/ping", ping)

    async with TestClient(app) as client:

        assert (await client.get("/ping")).status_code == 200

r.span_id is the root request span id for assertions on client.tracer. Request shape: Request.

2. Long-lived response (stream)

When the handler keeps the connection open (SSE, chunked bodies), use async with client.stream("GET", path, ...). Read with await r.body(), async for chunk in r.iter_bytes(), or for SSE async for ev in r.iter_events(). Leaving the block disconnects the request and awaits the handler.

stream does not follow redirects. Send Accept-Encoding: identity when using iter_bytes() on compressed responses.

from stario import App, Context, Writer

from stario.testing import TestClient

async def test_stream_joins_byte_chunks():

    app = App()

    async def chunked(c: Context, w: Writer) -> None:

        w.write_headers(200)

        w.write(b"hel")

        w.write(b"lo")

        w.end()

    app.get("/chunked", chunked)

    async with TestClient(app) as client:

        async with client.stream(

            "GET", "/chunked", headers={"Accept-Encoding": "identity"}

        ) as r:

            parts = [chunk async for chunk in r.iter_bytes()]

    assert b"".join(parts) == b"hello"

Assert telemetry with r.span_id and client.tracer.find_span(..., root_id=r.span_id). client.exchanges lists recent buffered round-trips.

3. Background work after the response

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 to wait until the app task queue is quiet and tracer spans have settled.

Do not call drain_tasks() from inside a coroutine scheduled with app.create_task — it can deadlock (same risk as awaiting drain from within handler-scheduled work).

For assertions on late work, use a child span: start() in the handler, end() when the background task finishes, then inspect client.tracer after drain_tasks().

import asyncio

import stario.responses as responses

from stario import App, Context, Writer

from stario.testing import TestClient

async def test_response_then_background():

    app = App()

    async def handler(c: Context, w: Writer) -> None:

        child = c.span.step("work.after_response")

        child.start()

        async def bg():

            try:

                await asyncio.sleep(0.02)

                child.attr("work.result", "done")

            finally:

                child.end()

        c.app.create_task(bg())

        responses.empty(w, 204)

    app.post("/action", handler)

    async with TestClient(app) as client:

        r = await client.post("/action")

        assert r.status_code == 204

        await client.drain_tasks()

    step = client.tracer.find_span("work.after_response", root_id=r.span_id)

    assert step is not None

    assert step.attributes.get("work.result") == "done"

API