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

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

@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:

@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:

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:

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:

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

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:

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