# Testing

Test HTTP, cookies, files, lifespan, and WebSockets with Flasgo's official client.

Source: https://flasgo.dev/guides/testing/

Flasgo's test client runs the application as ASGI without opening a network socket. The examples on this page build on the application below.

## HTTP tests

```python
from flasgo import Flasgo, Request

app = Flasgo(settings={"CSRF_ENABLED": False})


@app.post("/users")
async def create_user(request: Request):
    payload = await request.json()
    return {"name": payload["name"]}, 201


def test_create_user() -> None:
    client = app.test_client()
    response = client.post("/users", json={"name": "Ada"})

    assert response.status_code == 201
    assert response.json() == {"name": "Ada"}
```

Disabling CSRF is appropriate only for a focused non-browser test fixture. Keep separate tests with CSRF enabled for browser-facing routes.

The client provides `get`, `head`, `post`, `put`, `patch`, `delete`, and generic `request` methods. It supports:

- `json=` and raw `body=`;
- form `data=` and multipart `files=`;
- persistent cookies and `clear_cookies()`;
- custom headers and schemes;
- ordered header pairs for repeated wire headers;
- `follow_redirects=True`, with earlier responses in `history`;
- `status_code`, `headers`, `body`, `text`, `json()`, and `location`.

The client builds ASGI scopes the way a server does: `path` is percent-decoded, while `raw_path` and `query_string` stay percent-encoded, so non-ASCII paths and query strings work. Cookies set with `Max-Age=0` or a past `Expires` are removed from the jar. Followed redirects honour the scheme of an absolute `Location`, treat an explicit default port such as `http://localhost:80/` as the same origin, drop the body and its `Content-Type`/`Content-Length` headers when a `301`/`302`/`303` switches to GET, and stop at a redirect to another origin, returning that redirect response instead of replaying it against the application.

## Forms and uploads

```python
@app.post("/profiles")
async def create_profile(request: Request):
    form = await request.form()
    avatar = form.file("avatar")
    return {
        "display_name": form.get("display_name"),
        "avatar": avatar.filename if avatar else None,
    }


def test_profile_upload() -> None:
    client = app.test_client()
    response = client.post(
        "/profiles",
        data={"display_name": "Ada"},
        files={"avatar": ("avatar.txt", b"contents", "text/plain")},
    )

    assert response.status_code == 200
    assert response.json() == {"display_name": "Ada", "avatar": "avatar.txt"}
```

## Lifespan

Use a context manager when the application depends on startup and shutdown:

```python
@app.lifespan
async def lifespan(app: Flasgo):
    app.state.ready = True
    try:
        yield
    finally:
        app.state.ready = False


@app.get("/health", public=True)
def health():
    return {"ready": app.state.ready}


def test_ready_state() -> None:
    with app.test_client() as client:
        assert app.state.ready is True
        assert client.get("/health").status_code == 200

    assert app.state.ready is False
```

One synchronous context owns one lifecycle event loop, run with `asyncio.Runner` so leftover tasks, async generators, and the default executor are shut down when the context exits. A failed lifespan handler makes entering the context raise `RuntimeError`. If the ASGI application itself crashes during startup, for example in wrapping middleware, the client raises that exception immediately instead of waiting for its 5-second startup timeout. For async tests:

```python
async def test_async_request() -> None:
    async with app.test_client() as client:
        response = await client.arequest("GET", "/health")
        assert response.status_code == 200
```

## WebSockets

Synchronous WebSocket sessions require a synchronous client context:

```python
from flasgo import WebSocket


@app.websocket("/rooms/<int:room_id>", public=True)
async def room(websocket: WebSocket, room_id: int):
    await websocket.accept()
    async for message in websocket.iter_json():
        await websocket.send_json({"room_id": room_id, "echo": message})


def test_room_websocket() -> None:
    with app.test_client() as client:
        with client.websocket_connect("/rooms/7") as websocket:
            websocket.send_json({"message": "hello"})
            assert websocket.receive_json() == {
                "room_id": 7,
                "echo": {"message": "hello"},
            }
```

Use `headers=[("x-version", "1"), ("x-version", "2")]` with HTTP requests or `websocket_connect` when a test must preserve duplicate headers instead of collapsing them into a mapping. This is useful for verifying that typed scalar header bindings reject ambiguity while collection bindings accept repeated fields.

Async sessions use the async client context:

```python
async def test_room_websocket_async() -> None:
    async with app.test_client() as client:
        async with client.awebsocket_connect("/rooms/7") as websocket:
            await websocket.send_json({"message": "hello"})
            assert await websocket.receive_json() == {
                "room_id": 7,
                "echo": {"message": "hello"},
            }
```

Pass `origin=`, `headers=`, and `subprotocols=` to test handshake policy. Use `origin=None` specifically to verify that missing origins are denied. Closing a session after the server has already closed the socket still waits for the handler to finish, so handler exceptions surface in the test.

## Dependency overrides

```python
from typing import Annotated

from flasgo import Depends


def greeting() -> str:
    return "Hello"


def test_greeting() -> None:
    def replacement() -> str:
        return "Test greeting"

    with app.override_dependencies({greeting: replacement}):
        response = app.test_client().get("/greeting")
        assert response.json() == {"message": "Test greeting"}

    assert app.test_client().get("/greeting").json() == {"message": "Hello"}


@app.get("/greeting", public=True)
def greet(message: Annotated[str, Depends(greeting)]):
    return {"message": message}
```

Overrides nest and restore automatically, remain isolated between concurrent async test contexts, and preserve the original OpenAPI schema. Since 0.11.0, an override also applies when the request is made through a client opened earlier with `with app.test_client() as client:`, because each request runs in a copy of the caller's context. For yielding providers, assert cleanup order and cleanup on validation failure, handler errors, send failures, and cancellation. Use function scope when a cleanup/commit failure must prevent response delivery.

## Incremental streaming

`astream` exercises backpressure without buffering the entire response:

```python
from flasgo import StreamingResponse


@app.get("/chunks", public=True)
def chunks() -> StreamingResponse:
    async def content():
        yield b"first\n"
        yield b"second\n"

    return StreamingResponse(content())


async def test_stream() -> None:
    async with app.test_client() as client:
        async with client.astream("GET", "/chunks") as response:
            assert response.status_code == 200
            iterator = response.iter_bytes()
            assert await anext(iterator) == b"first\n"
            assert await anext(iterator) == b"second\n"
            assert [chunk async for chunk in iterator] == []
```

`astream(method, path, headers=..., body=..., json=..., scheme=...)` is an async context manager that yields an `AsyncTestStream`, exported from `flasgo` for type annotations. Leaving it disconnects the stream; its one-message queue retains backpressure. A stream that fails after sending headers raises an incomplete-response error instead of reporting successful completion. Do not use it inside a synchronous client lifespan context. The ordinary buffered client remains useful for finite streams.

Also test explicit response contracts with extra private fields, invalid output (`500`), and nested models. For server-side sessions, cover revocation, rotation, expiry, stale concurrent writes (`409`), and storage failures (`503`). See [Shared Quotas and Sessions](/guides/shared-storage/#development-storage-and-integration-tests) for real Redis/Valkey test configuration.

## What to test

For 0.11.0 upgrades, cover dependency overrides inside an open `with app.test_client()` block, `Form()` models whose `__post_init__` rejects a combination of fields (now `422`), fixed-length uploaded-file tuples, routes under an ASGI `root_path`, and JWT secrets that are not usable as HMAC keys (now `ValueError` at configuration time). Tests that patched names on `flasgo.request` or `flasgo.session`, or asserted `Request.__module__`, must target `flasgo._request` and `flasgo._session` instead.

For 0.10.0 upgrades, cover `after_request` hooks that now run for `404`, `405`, and authentication/authorization denials; authentication backends that raise (now `500`, not `401`); `ALLOWED_HOSTS` and `CSRF_TRUSTED_ORIGINS` entries that now fail at startup; parameter markers on the outer annotation; plain ASCII decimal numeric query, header, and form values; empty `1xx`/`204`/`304` bodies; and SSE heartbeats shorter than `idle_timeout`. Test suites that relied on the client replaying cross-origin redirects, keeping expired cookies, or passing unencoded non-ASCII paths may need updating.

For 0.9.1 upgrades, also cover registration-time route conflicts and specific routes over catch-all routes; wrong-typed booleans; repeated session/CSRF cookies, CSRF headers, Origin, and Referer; failed-handler session rollback; deliberate session changes in error handlers; and malformed responses that must not write session state. Verify that rejected WebSocket heads, origins, and client-IP quotas never invoke session storage.

For cancellation-resistant streaming fixtures, release blockers and await outstanding cleanup in `finally`, even when an assertion fails. Otherwise a regression test can hang its event loop during shutdown. Assert cleanup outcomes separately from successful body delivery. Metrics-enabled tests should use the client's lifespan context to exercise sampler startup/shutdown and the public `app.metrics_registry` for application metrics.

Alongside success cases, cover invalid hosts, signed CSRF failures, missing and insufficient permissions, invalid or expired JWTs, rate-limit exhaustion, oversized request and WebSocket bodies, typed `422` validation failures, validation-budget exhaustion, malformed JSON/forms, duplicate headers and cookies, unsafe outbound URLs, invalid response headers, startup failure, and background-task failure logging. If tracing is enabled, pass an application-owned test tracer provider so tests can inspect spans without exporting them.