Testing

The testing module provides specialized testing utilities for WebSocket consumers, including enhanced communicators and base test case classes with authentication support and message collection capabilities.

WebSocket Communicator

class chanx.testing.WebsocketCommunicator(application: Any, path: str, headers: list[tuple[bytes, bytes]] | None = None, subprotocols: list[str] | None = None, spec_version: int | None = None, *, consumer: type[AsyncJsonWebsocketConsumer[Any]])

Enhanced WebSocket communicator for testing Chanx consumers.

Extends the base WebSocket communicator with Chanx-specific features:

  • Structured message sending and receiving with BaseMessage objects

  • Automatic message collection until completion signals

  • Message validation using consumer's type adapters

  • Connection state tracking

  • Async context manager support for automatic cleanup

Key methods: - send_message(): Send BaseMessage objects directly - receive_all_json(): Collect all messages until timeout - receive_all_messages(): Collect and validate messages until stop action - connect()/disconnect(): Enhanced connection management

The communicator automatically handles message serialization/deserialization and integrates with Chanx's completion signal system for reliable testing.

action_key: str = 'action'
application: Any
async assert_closed() → None

Asserts that the WebSocket has been closed.

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

Connects to the WebSocket and tracks connection state.

Parameters:

timeout -- Maximum time to wait for connection (in seconds)

Returns:

Tuple of (connected, status_code)

consumer: type[AsyncJsonWebsocketConsumer[Any]]
async disconnect(code: int = 1000, timeout: float = 1) → None

Closes the socket

Parameters:

  • code -- Optional code to disconnect

  • timeout -- Maximum time to wait for connection (in seconds)

async receive_all_json(timeout: float = 1) → list[dict[str, Any]]

Receives and collects all JSON messages until an ACTION_COMPLETE message is received or timeout occurs.

Parameters:

timeout -- Maximum time to wait for messages (in seconds)

Returns:

List of received JSON messages

async receive_all_messages(stop_action: Literal['complete', 'event_complete', 'group_complete'] | str = 'complete', timeout: float = 1) → list[BaseMessage]

Receives and collects JSON messages until a specific action is received.

Automatically filters out completion messages (ACTION_COMPLETE and GROUP_ACTION_COMPLETE).

Parameters:

  • stop_action -- The action type to stop collecting at

  • timeout -- Maximum time to wait for messages (in seconds)

Returns:

List of received JSON messages (excluding completion messages)

async send_message(message: BaseMessage) → None

Sends a Message object as JSON to the WebSocket.

Parameters:

message -- The Message instance to send