responder/docs/source/testing.rst

Testing
=======

Responder includes a built-in test client powered by Starlette's
``TestClient``. You don't need to start a server — tests run in-process,
making them fast and reliable. There's no separate test server to manage,
no ports to allocate, and no race conditions to worry about. Just import
your app and start making requests.


Getting Started
---------------

Given a simple application in ``api.py``::

    import responder

    api = responder.API()

    @api.route("/")
    def hello(req, resp):
        resp.text = "hello, world!"

    if __name__ == "__main__":
        api.run()

You can test it with pytest. Every Responder ``API`` instance has a
``requests`` property that gives you a test client — use it exactly like
you'd use ``requests`` or ``httpx``::

    # test_api.py
    import api as service

    def test_hello():
        r = service.api.requests.get("/")
        assert r.text == "hello, world!"

Run your tests::

    $ pytest

That's really all there is to it. No configuration, no test server setup.


Using Fixtures
--------------

For larger test suites, pytest fixtures keep things organized. Create a
fixture that returns your API instance, and every test gets a fresh
reference to it::

    import pytest
    import api as service

    @pytest.fixture
    def api():
        return service.api

    def test_hello(api):
        r = api.requests.get("/")
        assert r.text == "hello, world!"

    def test_json(api):
        @api.route("/data")
        def data(req, resp):
            resp.media = {"key": "value"}

        r = api.requests.get(api.url_for(data))
        assert r.json() == {"key": "value"}

The ``api.url_for()`` method generates a URL for a given route endpoint,
so you don't have to hard-code paths in your tests. If you rename a route
later, your tests won't break. You can also give a route an explicit name and
reverse it by string — handy when the endpoint is a lambda or shares a name::

    @api.route("/users/{id}", name="user_detail")
    def user_detail(req, resp, *, id):
        ...

    api.url_for("user_detail", id=42)  # -> "/users/42"


Overriding Dependencies
-----------------------

Swap a dependency for a fake inside a ``with`` block — the override is
restored on exit, and it replaces (and bypasses the cache of) even an
``app``-scoped dependency::

    @api.dependency()
    def db():
        return RealDatabase()

    def test_users(api):
        with api.dependency_overrides(db=FakeDatabase()):
            r = api.requests.get("/users")
        assert r.status_code == 200

Each value may be a bare object (wrapped automatically) or a provider callable
— a callable receives sub-dependencies and the request just like a real one.


Testing JSON APIs
-----------------

Most APIs send and receive JSON. The test client makes this natural — pass
``json=`` to send a JSON body, and call ``.json()`` on the response to
parse it::

    def test_create_item(api):
        @api.route("/items")
        async def create(req, resp):
            data = await req.media()
            resp.media = {"created": data}
            resp.status_code = 201

        r = api.requests.post(api.url_for(create), json={"name": "widget"})
        assert r.status_code == 201
        assert r.json() == {"created": {"name": "widget"}}

You can also test content negotiation by setting the ``Accept`` header::

    r = api.requests.get("/data", headers={"Accept": "application/yaml"})
    assert "key: value" in r.text


Testing Request Validation
--------------------------

Responder validates typed handler inputs for you, returning ``422`` with a
``{"errors": [...]}`` body when something doesn't fit. That makes validation
tests delightfully boring: send bad data, assert the status code.

The simplest form is a Pydantic-annotated body parameter. On ``POST``,
``PUT``, ``PATCH``, and ``DELETE`` it receives the parsed, validated body —
no ``await req.media()`` required::

    from pydantic import BaseModel

    class Item(BaseModel):
        name: str
        price: float

    def test_body_validation(api):
        @api.route("/items", methods=["POST"])
        async def create(req, resp, *, item: Item):
            resp.media = item.model_dump()

        # Valid request
        r = api.requests.post("/items", json={"name": "thing", "price": 9.99})
        assert r.status_code == 200
        assert r.json() == {"name": "thing", "price": 9.99}

        # Missing required field
        r = api.requests.post("/items", json={"name": "thing"})
        assert r.status_code == 422
        assert "errors" in r.json()

Query strings, headers, and cookies validate the same way, through the typed
parameter markers ``Query``, ``Header``, ``Cookie``, and ``Path`` (see the
:doc:`tour <tour>`). A missing required value or a failed coercion is a
``422`` too::

    from responder import Query

    def test_query_validation(api):
        @api.route("/search")
        def search(req, resp, *, q: str = Query(...), limit: int = Query(10)):
            resp.media = {"q": q, "limit": limit}

        # limit is coerced from the query string to an int
        r = api.requests.get("/search?q=hi&limit=5")
        assert r.json() == {"q": "hi", "limit": 5}

        # the required q is missing -> 422
        r = api.requests.get("/search")
        assert r.status_code == 422
        assert "errors" in r.json()


Testing File Uploads
--------------------

File uploads use the ``files`` parameter, just like the ``requests``
library. Each file is a tuple of ``(filename, content, content_type)``::

    def test_upload(api):
        @api.route("/upload")
        async def upload(req, resp):
            files = await req.media("files")
            resp.media = {"received": list(files.keys())}

        files = {"doc": ("report.pdf", b"content", "application/pdf")}
        r = api.requests.post(api.url_for(upload), files=files)
        assert r.json() == {"received": ["doc"]}


Testing Headers and Cookies
---------------------------

Check response headers and cookies just like you would with any HTTP
client::

    def test_headers(api):
        @api.route("/")
        def view(req, resp):
            resp.headers["X-Custom"] = "hello"
            resp.cookies["session"] = "abc123"

        r = api.requests.get("/")
        assert r.headers["X-Custom"] == "hello"
        assert "session" in r.cookies


Testing Sessions
----------------

Sessions need a signing key. The old public default now raises, and an
instance built without one mints a random per-process key (fine for a quick
script, useless across a restart). So pass a real ``secret_key``. Because the
test client speaks ``http://``, also pass ``session_https_only=False`` —
otherwise the (Secure by default) session cookie won't round-trip::

    def test_session():
        api = responder.API(
            secret_key="test-secret-key-1234",
            session_https_only=False,
        )

        @api.route("/login", methods=["POST"])
        def login(req, resp):
            resp.session["user"] = "kenneth"

        @api.route("/me")
        def me(req, resp):
            resp.media = {"user": req.session.get("user")}

        # The test client persists cookies, so the session carries over.
        api.requests.post("/login")
        r = api.requests.get("/me")
        assert r.json() == {"user": "kenneth"}

``resp.session`` is a write-through view of ``req.session``, so assigning a
whole dict (``resp.session = {"user": "kenneth"}``) persists as well. See the
:doc:`configuration guide <guide-config>` for how the signing key is resolved
in production.


Testing Dependencies
--------------------

Dependencies resolve per request, so a route that uses one needs no special
wiring — register the provider, call the route. And because registering a
name again replaces the previous provider, you can swap in a fake (a stub
database, a fixed clock) right inside the test::

    def test_dependency(api):
        @api.dependency()
        def db():
            return {"alice": {"name": "Alice"}}

        @api.route("/users/{user_id}")
        def get_user(req, resp, *, user_id, db):
            resp.media = db.get(user_id, {})

        r = api.requests.get("/users/alice")
        assert r.json() == {"name": "Alice"}

Generator providers run their teardown after the response is sent, so any
assertion about cleanup belongs *after* the request returns. App-scoped
providers (``scope="app"``) only tear down on lifespan shutdown — enter the
client as a context manager (``with api.requests as session:``) so the
shutdown actually fires. A Pydantic body parameter always wins over a
same-named dependency, so an injected body is never shadowed by a provider.


Testing WebSockets
------------------

WebSocket tests use Starlette's ``TestClient`` directly, since WebSocket
connections require a different protocol. The ``websocket_connect`` context
manager gives you a connection you can send and receive on::

    from starlette.testclient import TestClient

    def test_websocket(api):
        @api.route("/ws", websocket=True)
        async def ws(ws):
            await ws.accept()
            name = await ws.receive_text()
            await ws.send_text(f"hello, {name}!")
            await ws.close()

        client = TestClient(api)
        with client.websocket_connect("/ws") as ws:
            ws.send_text("world")
            assert ws.receive_text() == "hello, world!"


Async Tests
-----------

The sync client covers most tests, but some behavior is genuinely async —
SSE streams, concurrent handlers, app-scoped dependency teardown. For those,
``responder.testing.AsyncTestClient`` is the async mirror of
``api.requests``: an ``httpx.AsyncClient`` dispatching requests in-process
over ``responder.testing.ASGIStreamingTransport``. No server, no sockets.
Unlike ``httpx.ASGITransport`` — which runs the handler to completion and
buffers the whole body before returning — this transport returns as soon as
the response starts and yields body chunks live, so it can read from endless
SSE streams (heartbeats included) and close them when done. It works with
plain ``asyncio.run`` (no pytest plugin required) or with ``pytest-asyncio``::

    import asyncio
    from responder.testing import AsyncTestClient

    def test_async_hello(api):
        @api.route("/")
        def hello(req, resp):
            resp.text = "hello, world!"

        async def main():
            async with AsyncTestClient(api) as client:
                r = await client.get("/")
                assert r.text == "hello, world!"

        asyncio.run(main())

Entering the client with ``async with`` also runs the application's
lifespan — startup fires on enter, shutdown on exit — the async counterpart
of ``with api.requests as session:``. Requests made without ``async with``
work too, but skip lifespan events, just like ``api.requests``.

Because requests run on the test's own event loop, you can drive several
in flight at once::

    async def main():
        async with AsyncTestClient(api) as client:
            responses = await asyncio.gather(
                client.get("/a"), client.get("/b"), client.get("/c"),
            )


Testing Server-Sent Events
--------------------------

Responder ships SSE support via ``resp.sse`` — and ``responder.testing``
ships the matching parser, so you never hand-roll ``text/event-stream``
parsing in a test. Each parsed event is an ``SSEEvent`` with ``data``,
``event`` (``"message"`` when the server sent no ``event:`` field), ``id``,
and ``retry`` attributes; ``event.json()`` decodes the ``data`` payload.

For a finite stream, the sync client buffers the whole body — parse it with
``parse_sse``::

    from responder.testing import parse_sse

    def test_events(api):
        @api.route("/events")
        async def events(req, resp):
            @resp.sse
            async def stream():
                for n in range(3):
                    yield {"data": {"n": n}, "id": n}

        r = api.requests.get("/events")
        assert r.headers["Content-Type"].startswith("text/event-stream")
        events = parse_sse(r.text)
        assert len(events) == 3
        assert events[0].json() == {"n": 0}
        assert events[0].id == "0"

For live or infinite streams, pair ``AsyncTestClient`` with ``iter_sse``,
which yields events as the server produces them — stop iterating whenever
you've seen enough::

    from responder.testing import AsyncTestClient, iter_sse

    async def main():
        async with AsyncTestClient(api) as client:
            async with client.stream("GET", "/events") as r:
                async for event in iter_sse(r):
                    assert event.event == "message"
                    break  # done after the first event

    asyncio.run(main())

``collect_sse(response)`` is the one-shot form, gathering every event from a
finite stream into a list. Heartbeat comment lines (from
``resp.sse(heartbeat=...)``) are skipped by the parser, exactly as a
browser ``EventSource`` would skip them.


Testing Error Handling
----------------------

By default, the test client raises exceptions from your route handlers,
which is usually what you want — it makes bugs obvious. But when you're
testing error handling specifically, you want to see the error response
instead. Disable exception propagation with ``raise_server_exceptions``::

    from starlette.testclient import TestClient

    def test_500(api):
        @api.route("/fail")
        def fail(req, resp):
            raise ValueError("something broke")

        client = TestClient(api, raise_server_exceptions=False)
        r = client.get(api.url_for(fail))
        assert r.status_code == 500

If you've registered a custom exception handler, you can test that too::

    def test_custom_error(api):
        @api.exception_handler(ValueError)
        async def handle(req, resp, exc):
            resp.status_code = 400
            resp.media = {"error": str(exc)}

        @api.route("/fail")
        def fail(req, resp):
            raise ValueError("bad input")

        client = TestClient(api, raise_server_exceptions=False)
        r = client.get(api.url_for(fail))
        assert r.status_code == 400
        assert r.json() == {"error": "bad input"}

Two conveniences make error tests shorter. ``responder.abort()`` raises a
rendered HTTP error from anywhere in a handler. By default, framework-generated
errors use ``application/problem+json``; because ``abort()`` raises a regular
``HTTPException``, the test client returns it as a response (no
``raise_server_exceptions=False`` needed)::

    from responder import abort

    def test_abort(api):
        @api.route("/admin")
        def admin(req, resp):
            abort(403, detail="Forbidden")

        r = api.requests.get("/admin", headers={"Accept": "application/json"})
        assert r.status_code == 403
        assert r.json()["title"] == "Forbidden"
        assert r.json()["detail"] == "Forbidden"

And ``api.add_exception_handler(exc_or_status, handler)`` is the imperative
twin of the ``@api.exception_handler`` decorator — handy for wiring handlers
inside a fixture. It accepts an exception class or an integer status code.


Testing Lifespan Events
-----------------------

If your app uses startup and shutdown events (for database connections,
caches, etc.), you need the test client to trigger them. Wrap the client
in a ``with`` block — startup runs on enter, shutdown runs on exit::

    def test_with_lifespan(api):
        started = {"value": False}

        @api.on_event("startup")
        async def on_startup():
            started["value"] = True

        @api.route("/")
        def check(req, resp):
            resp.media = {"started": started["value"]}

        with api.requests as session:
            r = session.get("http://localhost/")
            assert r.json() == {"started": True}

Without the ``with`` block, lifespan events won't fire, which can lead to
confusing test failures if your routes depend on startup initialization.


Testing Before and After Hooks
------------------------------

Before-request and after-request hooks run automatically during tests,
just like in production. You can verify their effects on the response::

    def test_hooks(api):
        @api.before_request
        def add_version(req, resp):
            resp.headers["X-Version"] = "5.0"

        @api.after_request
        def add_timing(req, resp):
            resp.headers["X-Served-By"] = "responder"

        @api.route("/")
        def view(req, resp):
            resp.text = "ok"

        r = api.requests.get("/")
        assert r.headers["X-Version"] == "5.0"
        assert r.headers["X-Served-By"] == "responder"


Testing Rate Limiting
---------------------

Rate limiters are just hooks — they run automatically during tests.
Verify the headers and the 429 response::

    from responder.ext.ratelimit import RateLimiter

    def test_rate_limiting():
        api = responder.API(allowed_hosts=["localhost"], sessions=False)
        limiter = RateLimiter(requests=2, period=60)
        limiter.install(api)

        @api.route("/")
        def view(req, resp):
            resp.text = "ok"

        # First two requests succeed
        for _ in range(2):
            r = api.requests.get("http://localhost/")
            assert r.status_code == 200
            assert "X-RateLimit-Remaining" in r.headers

        # Third request is rate limited
        r = api.requests.get("http://localhost/")
        assert r.status_code == 429


Testing Mounted Apps
--------------------

When testing WSGI apps mounted at a subroute, use ``localhost`` as the
host to avoid Werkzeug's trusted host validation::

    from flask import Flask

    def test_flask_mount():
        api = responder.API(allowed_hosts=["localhost"], sessions=False)

        flask_app = Flask(__name__)
        @flask_app.route("/")
        def hello():
            return "Hello from Flask!"

        api.mount("/flask", flask_app)

        r = api.requests.get("http://localhost/flask")
        assert r.status_code == 200
        assert "Hello from Flask" in r.text


Tips
----

- **Keep tests fast.** The in-process test client is already fast — no
  network overhead. Avoid ``time.sleep()`` in tests.

- **One API per test** when testing configuration. If you need a specific
  ``API()`` configuration (like ``cors=True``), create a new instance
  in the test rather than sharing the fixture.

- Use ``api.url_for()`` instead of hard-coded paths. It's a small
  thing, but it makes refactoring painless.

- **Test the contract, not the implementation.** Assert on status codes,
  response bodies, and headers — not on internal state.
  ``examples/atelier.py`` is the canonical contract example: it validates the
  generated OpenAPI document and drives a generated client against the same app.
  See :doc:`examples` for the rest of the runnable examples.

- **Use ``localhost`` for mounted WSGI apps.** Werkzeug 3.1.7+ validates
  the ``Host`` header, so avoid synthetic hosts like ``;`` in tests.

- **Quiet the session-key warning.** Any ``API()`` built without a
  ``secret_key`` mints a random per-process key and logs a warning. In tests
  that don't touch sessions, pass ``sessions=False``; in tests that do, pass a
  real ``secret_key=...`` (plus ``session_https_only=False`` so the cookie
  round-trips over ``http``).

- **``req.method`` is uppercase.** It returns a plain ``str`` such as
  ``"GET"`` / ``"POST"``. Comparisons are case-sensitive, so compare against
  the uppercase form (``== "GET"``) or call ``req.method.lower()`` yourself.