Testing with TestClient

stario.testing.TestClient runs HTTP handlers in-process — the same HTTP stack as serve(), without binding a port.

Always open the client with async with TestClient(...) so shutdown runs: open exchanges disconnect, background work drains, and bootstrap teardown runs. The Testing reference is canonical for full argument lists.

Basic usage#

Pass the same bootstrap callable you pass to serve(), so routes and middleware match production. Use TestClient(bootstrap, app_factory=build_app) when each test needs a fresh App — register routes in build_app() and keep bootstrap for shared setup and teardown.

from stario.testing import TestClient

from myapp.main import bootstrap

async def test_home():

    async with TestClient(bootstrap) as client:

        r = await client.get("/")

        assert r.status_code == 200

You can also pass a ready-made App for small, self-contained tests. With a bootstrap factory, client.app is the live app only after you enter the context manager.

The client fixture#

Most test files start with a fixture that opens one client for the module.

import pytest

from stario.testing import TestClient

from myapp.main import bootstrap

@pytest.fixture

async def client():

    async with TestClient(bootstrap) as c:

        yield c

async def test_page(client):

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

Integration test#

Use the real bootstrap when you want the same composition root as serve() — routes, middleware, and static assets all match production.

from stario.testing import TestClient

from myapp.main import bootstrap

async def test_items_list():

    async with TestClient(bootstrap) as client:

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

        assert r.status_code == 200

        data = r.json()

        assert "items" in data

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"

Small `App` in the test file#

When you only need one route or a narrow behavior, build an App, register handlers, and pass app to TestClient.

import json

from stario import App, Context, Route, Writer

from stario.testing import TestClient

ECHO = Route("POST /echo")

async def test_echo_json():

    app = App()

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

        payload = json.loads(await c.req.body())

        w.json({"you_sent": payload})

    app.add(ECHO, echo)

    async with TestClient(app) as client:

        r = await client.post(ECHO.href(), json={"x": 1})

        assert r.status_code == 200

        assert r.json()["you_sent"]["x"] == 1

Streaming#

Use async with client.stream(...) when the handler keeps the connection open — chunked body, SSE, live logs. Read with await r.body(), async for over r.iter_bytes(), or r.iter_events() for text/event-stream. stream does not follow redirects; inspect Location or use buffered get with follow_redirects.

Chunked bytes:

from stario import App, Context, Route, Writer

from stario.testing import TestClient

CHUNKED = Route("GET /chunked")

async def test_stream_reassembles_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.add(CHUNKED, chunked)

    async with TestClient(app) as client:

        async with client.stream(

            "GET", CHUNKED.href(), headers={"Accept-Encoding": "identity"}

        ) as r:

            assert r.status_code == 200

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

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

SSE-shaped responses:

from stario import App, Context, Route, Writer

from stario.testing import TestClient

EVENTS = Route("GET /events")

async def test_stream_reads_sse_events():

    app = App()

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

        w.headers.set("content-type", "text/event-stream")

        w.write_headers(200)

        w.write(b"event: ping\ndata: hello\n\n")

        w.end()

    app.add(EVENTS, sse)

    async with TestClient(app) as client:

        async with client.stream(

            "GET", EVENTS.href(), headers={"Accept-Encoding": "identity"}

        ) as r:

            events = [e async for e in r.iter_events()]

    assert events == [{"event": "ping", "data": "hello"}]

A live page#

The most useful test for a live page opens the stream, reads the first render, sends a command, and reads the repaint. In the chat-room shape it looks like this:

async def test_room_send_repaints_the_stream(client):

    await client.post("/rooms", json={"room_title": "General"})

    signals = json.dumps({"user_id": "u1", "username": "HappyFox", "color": "#e74c3c"})

    async with client.stream(

        "GET",

        "/rooms/general/subscribe",

        params={"datastar": signals},

        headers={"accept-encoding": "identity"},

    ) as live:

        events = live.iter_events()

        first = await anext(events)

        assert "General" in first["data"]

        await anext(events)  # the repaint for this tab's own join

        r = await client.post(

            "/rooms/general/send",

            json={"user_id": "u1", "username": "HappyFox", "color": "#e74c3c", "message": "hello room"},

        assert r.status_code == 204

        repaint = await anext(events)

        assert "hello room" in repaint["data"]

`Route` in tests#

Build paths with the same Route constants your app uses — no url_for or route name=:

from stario.testing import TestClient

from myapp.main import bootstrap

from myapp.urls import HOME

async def test_home_path():

    async with TestClient(bootstrap) as client:

        r = await client.get(HOME.href())

        assert r.status_code == 200

For static assets, use ASSETS.href("css/app.css") from your manifest module — the same strings views use in production.

Tracer snapshot#

To assert on finished spans for a request, use r.span_id with client.tracer:

from stario.testing import TestClient

from myapp.main import bootstrap

from myapp.urls import REPORT

async def test_request_traced():

    async with TestClient(bootstrap) as client:

        r = await client.get(REPORT.href(report_id="7"))

        t = client.tracer

        assert t.has_attribute(r.span_id, "request.path", REPORT.href(report_id="7"))

What to assert#

r.text, r.json(), r.status_code, r.headers, r.cookies.
r.span_id with client.tracer for finished spans and events.

client.exchanges lists buffered request/response pairs, not streaming calls. Call await client.drain_tasks() when you need background work finished before leaving the client context; exiting async with drains too. Do not call drain_tasks from inside work scheduled with app.create_task — that deadlocks.

Cookies, redirects, multipart#

Buffered calls update an automatic cookie jar. follow_redirects=True is the default; pass False to assert on a 303 or 307 without following. Multipart uses files= with optional form data=. See Testing for full argument lists.

Compression and large bodies#

Buffered get and post reassemble chunked bodies and decode Content-Encoding. TestClient(..., compression=CompressionConfig(...)) can mirror production codec defaults. For stream, prefer identity encoding when using iter_bytes.

By the way#

Use pytest with pytest-asyncio. Set asyncio_mode = auto under [tool.pytest.ini_options] in pyproject.toml so async def tests run without extra marks. In strict mode, use @pytest_asyncio.fixture for async fixtures, not plain @pytest.fixture.

The complete example is examples/chat-room/tests.