Skip to content

korvex.testing

The test client. See Testing.

An in-process test client for any ASGI app: no server, no network.

Guide: https://3d3n-pyc.github.io/korvex/testing/

TestClient 

TestClient(app: Callable[..., Awaitable[None]], *, base_url: str = 'http://testserver', headers: HeadersLike | None = None, cookies: Mapping[str, str] | None = None, raise_server_errors: bool = True, timeout: float | None = 5.0)

Calls an ASGI app in-process, from regular tests.

Same arguments as AsyncTestClient. As a context manager it runs the app's lifespan. Raises RuntimeError inside a running event loop: use AsyncTestClient there.

Guide: https://3d3n-pyc.github.io/korvex/testing/

cookies property 

cookies: dict[str, str]

The cookies sent with each request: those the app set and any added here.

headers property 

headers: Headers

Headers sent with each request.

close 

close() -> None

Closes the event loop, cancelling what the app left running.

request 

request(method: str, url: str, **kwargs: Any) -> TestResponse

Sends a request and returns the whole response; see AsyncTestClient.request().

get 

get(url: str, **kwargs: Any) -> TestResponse

A GET request; see request().

head 

head(url: str, **kwargs: Any) -> TestResponse

A HEAD request; see request().

options 

options(url: str, **kwargs: Any) -> TestResponse

A OPTIONS request; see request().

post 

post(url: str, **kwargs: Any) -> TestResponse

A POST request; see request().

put 

put(url: str, **kwargs: Any) -> TestResponse

A PUT request; see request().

patch 

patch(url: str, **kwargs: Any) -> TestResponse

A PATCH request; see request().

delete 

delete(url: str, **kwargs: Any) -> TestResponse

A DELETE request; see request().

stream 

stream(method: str, url: str, **kwargs: Any) -> _SyncStreamContext

Sends a request and returns the response before its body is read.

websocket 

websocket(url: str, **kwargs: Any) -> TestWebSocket

Opens a WebSocket, as a context manager; see AsyncTestClient.websocket().

AsyncTestClient 

AsyncTestClient(app: Callable[..., Awaitable[None]], *, base_url: str = 'http://testserver', headers: HeadersLike | None = None, cookies: Mapping[str, str] | None = None, raise_server_errors: bool = True, timeout: float | None = 5.0)

Calls an ASGI app in-process, from async tests.

As an async context manager it runs the app's lifespan (startup and shutdown hooks); without it, no hook runs.

  • base_url: scheme and host requests go to, unless a request gives a full URL
  • headers, cookies: sent with every request; a request's own replace the same name
  • raise_server_errors: raise in the test an error the app logs as unhandled (a 500)
  • timeout: seconds to wait for a response or WebSocket message, then TimeoutError; None waits forever

Guide: https://3d3n-pyc.github.io/korvex/testing/

request async 

request(method: str, url: str, *, query: Query | None = None, headers: HeadersLike | None = None, cookies: Mapping[str, str] | None = None, json: Any = _UNSET, form: Form | None = None, files: Files | None = None, body: bytes | str | None = None, follow_redirects: bool = False) -> TestResponse

Sends a request and returns the whole response.

Give the body as json, form (with files: {"f": ("a.png", data, "image/png")}) or raw body, one way only (TypeError otherwise). follow_redirects follows up to 20 redirects (a 303, or a 301/302 after a POST, becomes a GET without body).

stream 

stream(method: str, url: str, *, query: Query | None = None, headers: HeadersLike | None = None, cookies: Mapping[str, str] | None = None, json: Any = _UNSET, form: Form | None = None, files: Files | None = None, body: bytes | str | None = None) -> _StreamContext

Sends a request and returns the response before its body is read.

Leaving the block stops the app's handler, as a client going away would. Takes the arguments of request(), but not follow_redirects.

get async 

get(url: str, **kwargs: Any) -> TestResponse

A GET request; takes the same arguments as request().

head async 

head(url: str, **kwargs: Any) -> TestResponse

A HEAD request; see request().

options async 

options(url: str, **kwargs: Any) -> TestResponse

A OPTIONS request; see request().

post async 

post(url: str, **kwargs: Any) -> TestResponse

A POST request; see request().

put async 

put(url: str, **kwargs: Any) -> TestResponse

A PUT request; see request().

patch async 

patch(url: str, **kwargs: Any) -> TestResponse

A PATCH request; see request().

delete async 

delete(url: str, **kwargs: Any) -> TestResponse

A DELETE request; see request().

websocket 

websocket(url: str, *, subprotocols: Iterable[str] = (), query: Query | None = None, headers: HeadersLike | None = None, cookies: Mapping[str, str] | None = None) -> AsyncTestWebSocket

Opens a WebSocket, as an async context manager.

Entering raises WebSocketDisconnect if the app closes before accepting.

TestResponse 

TestResponse(status: int, headers: Headers, body: bytes, url: str)

A response: status, headers, body, text, json() and the final url.

text property 

text: str

The body, decoded with the charset of its Content-Type (UTF-8 by default).

json 

json() -> Any

The body, parsed as JSON.

TestWebSocket 

TestWebSocket(websocket: AsyncTestWebSocket, run: Callable[[Awaitable[Any]], Any])

The sync AsyncTestWebSocket.

AsyncTestWebSocket 

AsyncTestWebSocket(client: AsyncTestClient, scope: dict)

A WebSocket open on the app: subprotocol and headers are those it accepted with.

receive async 

receive(timeout: float | None = _UNSET) -> str | bytes

The next message from the app, text or bytes; WebSocketDisconnect once it closed.

timeout (seconds) overrides the client's.

close async 

close(code: int = 1000) -> None

Disconnects (unless the app closed first) and waits for the handler to end.