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)

Chanx extended WebsocketCommunicator for testing WebSocket consumers.

Provides enhanced testing capabilities for WebSocket applications including:

Structured message sending and receiving
Authentication handling and verification
Automatic message collection until completion signals
Group message broadcast testing
Connection state tracking
Message validation and error handling

Typical usage patterns:

Connection testing: connect(), assert_authenticated_status_ok()
Message exchange: send_message(), receive_all_json()
Authentication flows: wait_for_auth(), assert_authenticated_status_ok()
Error handling: send invalid messages and check error responses
Group messaging: create multiple communicators and test broadcasts

The communicator automatically handles message serialization/deserialization and provides convenience methods to simplify common WebSocket testing tasks.

async assert_authenticated_status_ok(max_auth_time: float = 0.5) → None

Assert that the WebSocket connection was authenticated successfully.

Waits for an authentication message and verifies that its status code is 200 OK.

Parameters:: max_auth_time -- Maximum time to wait for authentication message (in seconds)
Raises:: AssertionError -- If the authentication status is not 200 OK

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)

async receive_all_json(timeout: float = 5, *, wait_group: bool = False) → 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)
wait_group -- wait until the complete group messages are received

Returns:

List of received JSON messages

async receive_until_action(stop_action: str, timeout: float = 5, *, inclusive: bool = False) → list[dict[str, Any]]

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)
inclusive -- Whether to include the stop_action message in results

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

async wait_for_auth(send_authentication_message: bool | None = None, max_auth_time: float = 0.5, after_auth_time: float = 0.1) → AuthenticationMessage | None

Waits for and returns an authentication message if enabled in settings.

Parameters:

send_authentication_message -- Whether to expect auth message, defaults to setting
max_auth_time -- Maximum time to wait for authentication (in seconds)
after_auth_time -- Wait time sleep after authentication (in seconds)

Returns:

Authentication message or None if auth is disabled

WebSocket Test Case

class chanx.testing.WebsocketTestCase(*args: Any, **kwargs: Any)

Base test case for WebSocket testing with Chanx.

Provides a framework for testing WebSocket consumers with built-in support for:

Automatic WebSocket application discovery
Connection management and cleanup
Authentication testing
Message sending and receiving
Group broadcast testing

To use this class:

Subclass WebsocketTestCase
Set the ws_path class attribute to your WebSocket endpoint
Optionally override get_ws_headers() for authentication
Use self.auth_communicator for the main connection
Use create_communicator() only when testing multi-user scenarios

ws_path

WebSocket endpoint path to test (required)

Type:: str

router

WebSocket application router (auto-discovered)

Type:: Any

auth_communicator: Default WebSocket communicator for the main connection

property auth_communicator: WebsocketCommunicator: Returns a connected WebsocketCommunicator instance. The instance is created using create_communicator if not already exists.

create_communicator(*, router: Any | None = None, ws_path: str | None = None, headers: list[tuple[bytes, bytes]] | None = None, subprotocols: list[str] | None = None) → WebsocketCommunicator

Creates a WebsocketCommunicator for testing WebSocket connections.

Creates and tracks a communicator instance for interacting with WebSocket consumers in tests, allowing you to create multiple communicators to test various scenarios including: - Multi-user WebSocket interactions - Testing group message broadcasting - Testing authentication with different credentials - Simulating concurrent connections

The method tracks all created communicators and automatically handles their cleanup during tearDown() to prevent resource leaks.

Parameters:

router -- Application to use (defaults to self.router)
ws_path -- WebSocket path to connect to (defaults to self.ws_path)
headers -- HTTP headers to include (defaults to self.ws_headers) Use different headers for testing multiple authenticated users
subprotocols -- WebSocket subprotocols to use (defaults to self.subprotocols)

Returns:

A configured WebsocketCommunicator instance ready for connecting

Raises:

AttributeError -- If ws_path is not set and not provided

get_subprotocols() → list[str]: Returns WebSocket subprotocols to use. Override this method to provide custom subprotocols.

get_ws_headers() → list[tuple[bytes, bytes]]: Returns WebSocket headers for authentication/configuration. Override this method to provide custom headers.

setUp() → None

Set up the test environment before each test method.

Initializes WebSocket headers and subprotocols by calling the corresponding getter methods, and prepares for tracking communicators.

tearDown() → None

Clean up after each test method.

Ensures all WebSocket connections created during the test are properly disconnected to prevent resource leaks and test isolation issues.

Key Methods

Connection Management

async WebsocketCommunicator.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)

async WebsocketCommunicator.disconnect(code=1000, timeout=1): Closes the socket

async WebsocketCommunicator.assert_closed() → None: Asserts that the WebSocket has been closed.

Authentication

async WebsocketCommunicator.wait_for_auth(send_authentication_message: bool | None = None, max_auth_time: float = 0.5, after_auth_time: float = 0.1) → AuthenticationMessage | None

Waits for and returns an authentication message if enabled in settings.

Parameters:

send_authentication_message -- Whether to expect auth message, defaults to setting
max_auth_time -- Maximum time to wait for authentication (in seconds)
after_auth_time -- Wait time sleep after authentication (in seconds)

Returns:

Authentication message or None if auth is disabled

async WebsocketCommunicator.assert_authenticated_status_ok(max_auth_time: float = 0.5) → None

Assert that the WebSocket connection was authenticated successfully.

Waits for an authentication message and verifies that its status code is 200 OK.

Parameters:: max_auth_time -- Maximum time to wait for authentication message (in seconds)
Raises:: AssertionError -- If the authentication status is not 200 OK

Message Handling

async WebsocketCommunicator.send_message(message: BaseMessage) → None

Sends a Message object as JSON to the WebSocket.

Parameters:: message -- The Message instance to send

async WebsocketCommunicator.receive_all_json(timeout: float = 5, *, wait_group: bool = False) → 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)
wait_group -- wait until the complete group messages are received

Returns:

List of received JSON messages

async WebsocketCommunicator.receive_until_action(stop_action: str, timeout: float = 5, *, inclusive: bool = False) → list[dict[str, Any]]

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)
inclusive -- Whether to include the stop_action message in results

Returns:

List of received JSON messages (excluding completion messages)

Test Case Management

WebsocketTestCase.create_communicator(*, router: Any | None = None, ws_path: str | None = None, headers: list[tuple[bytes, bytes]] | None = None, subprotocols: list[str] | None = None) → WebsocketCommunicator

Creates a WebsocketCommunicator for testing WebSocket connections.

Creates and tracks a communicator instance for interacting with WebSocket consumers in tests, allowing you to create multiple communicators to test various scenarios including: - Multi-user WebSocket interactions - Testing group message broadcasting - Testing authentication with different credentials - Simulating concurrent connections

The method tracks all created communicators and automatically handles their cleanup during tearDown() to prevent resource leaks.

Parameters:

router -- Application to use (defaults to self.router)
ws_path -- WebSocket path to connect to (defaults to self.ws_path)
headers -- HTTP headers to include (defaults to self.ws_headers) Use different headers for testing multiple authenticated users
subprotocols -- WebSocket subprotocols to use (defaults to self.subprotocols)

Returns:

A configured WebsocketCommunicator instance ready for connecting

Raises:

AttributeError -- If ws_path is not set and not provided

WebsocketTestCase.get_ws_headers() → list[tuple[bytes, bytes]]: Returns WebSocket headers for authentication/configuration. Override this method to provide custom headers.

WebsocketTestCase.get_subprotocols() → list[str]: Returns WebSocket subprotocols to use. Override this method to provide custom subprotocols.