Testing

Testing utilities for Fast Channels applications.

Application Communicator

class fast_channels.testing.ApplicationCommunicator(application, scope)

Bases: ApplicationCommunicator

Enhanced ASGI application communicator with proper type annotations.

This class extends the base ApplicationCommunicator from asgiref.testing to provide better type safety and documentation for testing ASGI applications.

async receive_nothing(timeout: float = 0.1, interval: float = 0.01) → bool

Assert that no output is received from the application.

Parameters:

timeout – Maximum time to wait for no output in seconds.
interval – Time interval between checks in seconds.

Returns:

True if no output was received, False otherwise.

async receive_output(timeout: float = 1) → Any

Receive output from the ASGI application.

Parameters:: timeout – Maximum time to wait for output in seconds.
Returns:: The message received from the application.
Raises:: TimeoutError – If no output is received within the timeout period.

async send_input(message: Any) → None

Send a message to the ASGI application.

Parameters:: message – The message to send to the application.

stop(exceptions: bool = True) → None

Stop the application communicator.

Parameters:: exceptions – Whether to raise exceptions that occurred during execution.

async wait(timeout: float = 1) → None

Wait for the application to finish processing.

Parameters:: timeout – Maximum time to wait in seconds.
Raises:: TimeoutError – If the application doesn’t finish within the timeout.

WebSocket Communicator

class fast_channels.testing.WebsocketCommunicator(application: MiddlewareProtocol | Callable[[MutableMapping[str, Any], Callable[[], Awaitable[Mapping[str, Any]]], Callable[[Mapping[str, Any]], Awaitable[None]]], Awaitable[None]], path: str, headers: Iterable[tuple[bytes, bytes]] | None = None, subprotocols: Iterable[str] | None = None, spec_version: int | None = None)

Bases: ApplicationCommunicator

ApplicationCommunicator subclass that has WebSocket shortcut methods.

It will construct the scope for you, so you need to pass the application (uninstantiated) along with the initial connection parameters.

async connect(timeout: float = 1) → tuple[bool, int | str | None]

Trigger the connection code.

Parameters:: timeout – Timeout for connection in seconds.
Returns:: On an accepted connection, returns (True, <chosen-subprotocol>) On a rejected connection, returns (False, <close-code>)

async disconnect(code: int = 1000, timeout: float = 1) → None: Closes the socket

async receive_from(timeout: float = 1) → str | bytes: Receives a data frame from the view. Will fail if the connection closes instead. Returns either a bytestring or a unicode string depending on what sort of frame you got.

async receive_json_from(timeout: float = 1) → Any: Receives a JSON text frame payload and decodes it

async send_json_to(data: Any) → None: Sends JSON data as a text frame

async send_to(text_data: str | None = None, bytes_data: bytes | None = None) → None

Sends a WebSocket frame to the application.

Parameters:

text_data – Text data to send (mutually exclusive with bytes_data).
bytes_data – Binary data to send (mutually exclusive with text_data).