Testing with TestClient

stario.testing.TestClient runs HTTP handlers in-process with no listening socket—the same HTTP stack as stario serve, without binding a port. Always open the client with async with TestClient(...) so shutdown runs (disconnect open exchanges, drain background work, then bootstrap teardown—or signal_shutdown for a bare App).

This page is copy-paste recipes; the Testing reference is canonical for full argument lists and pytest-asyncio setup.

Basic usage

Pass the same bootstrap callable you use with stario serve (see Testing — pattern 1) 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(), keep bootstrap for shared setup/teardown only (see Testing — API).

python
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.

Async test setup (pytest-asyncio)

Use pytest with pytest-asyncio. Set asyncio_mode = auto under [tool.pytest.ini_options] in your app’s pyproject.toml so ordinary async def tests run without extra marks. In strict mode, use @pytest_asyncio.fixture for async fixtures (not plain @pytest.fixture) and follow pytest-asyncio—see also Testing.

Examples

Integration test: real bootstrap, query params, JSON body

Use this when you want the same composition root as stario serve—routes, middleware, and static assets all match production.

python
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.

python
import json
 
import stario.responses as responses
from stario import App, Context, Writer
from stario.testing import TestClient
from stario.routing import UrlPath
 
ECHO = UrlPath("/echo")
 
 
async def test_echo_json():
    app = App()
 
    async def echo(c: Context, w: Writer) -> None:
        payload = json.loads(await c.req.body())
        responses.json(w, {"you_sent": payload})
 
    app.post(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 (chunks and SSE)

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:

python
import stario.responses as responses
from stario import App, Context, Writer
from stario.testing import TestClient
from stario.routing import UrlPath
 
CHUNKED = UrlPath("/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.get(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:

python
from stario import App, Context, Writer
from stario.testing import TestClient
from stario.routing import UrlPath
 
EVENTS = UrlPath("/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.get(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"}]

UrlPath in tests

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

python
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 (optional)

To assert on finished spans for a request, use r.span_id with client.tracer (see Testing — pattern 1):

python
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"))

Fixture pattern

python
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

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 (see Testing reference)

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 TestClient(...) also drains on exit. See Testing — pattern 3. Do not call drain_tasks from inside work scheduled with app.create_task (deadlock risk).

Cookies, redirects, multipart

Buffered calls update an automatic cookie jar. Use follow_redirects=True (default) or False when you need to assert on a 303/307 without following. Multipart: files= with optional form data=. See Testing reference for full argument lists.

Compression and large bodies

Buffered get / 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.


Related: Testing · Reading and writing Datastar signals · Request