Testing

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

Core Testing

The core testing module provides the foundation for Chanx's testing infrastructure with a framework-agnostic mixin.

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

Mixin providing enhanced WebSocket testing functionality for Chanx consumers.

This mixin provides Chanx-specific features that work across different WebSocket frameworks (Django Channels, fast-channels, etc.):

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 mixin automatically handles message serialization/deserialization and integrates with Chanx's completion signal system for reliable testing.

Concrete implementations should inherit from both this mixin and a framework-specific WebSocket communicator class (e.g., channels.testing.WebsocketCommunicator).

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[ChanxWebsocketConsumerMixin]

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 receive_json_from(timeout: float = 1) → Any

Receive and parse JSON data from the WebSocket.

Provided by the framework testing communicator (Channels/fast-channels).

Parameters:: timeout -- Maximum time to wait for data (seconds)
Returns:: Parsed JSON data as dictionary

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

Receive raw output from the WebSocket.

Provided by the framework testing communicator (Channels/fast-channels).

Parameters:: timeout -- Maximum time to wait for output (seconds)
Returns:: Raw output dictionary

async send_json_to(data: dict[str, Any]) → None

Send JSON data to the WebSocket.

Provided by the framework testing communicator (Channels/fast-channels).

Parameters:: data -- Dictionary to serialize and send as JSON

async send_message(message: BaseMessage) → None

Sends a Message object as JSON to the WebSocket.

Parameters:: message -- The Message instance to send

class chanx.core.testing.CapturedBroadcastEvent(event: BaseMessage, groups: Collection[str] | str | None)

Structure of a captured broadcast event.

event: BaseMessage

groups: Collection[str] | str | None

chanx.core.testing.capture_broadcast_events(consumer: type[ChanxWebsocketConsumerMixin], suppress: bool = True) → Generator[list[CapturedBroadcastEvent], None, None]

Capture broadcast events sent via broadcast_event() for testing purposes.

Similar to structlog's capture_logs(), this context manager captures calls to broadcast_event() by monkey-patching the broadcast_event method to spy on events.

Parameters:

consumer -- The consumer class to capture broadcast events from.
suppress -- If True (default), suppress actual broadcast event calls. If False, capture events and still call the original broadcast_event.

Returns:

A list that will be populated with captured broadcast events.

Django Channels Testing

The Django Channels testing module provides Django-specific testing utilities.

class chanx.channels.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[ChanxWebsocketConsumerMixin[Any]])

Base WebSocket communicator for testing Chanx consumers with Django Channels.

Combines Chanx testing features (send_message, receive_all_messages, message validation) with Django Channels' WebSocket communicator (connect, disconnect, send_json_to).

For Django-specific features like authentication, use DjangoWebsocketCommunicator.

application: Any

consumer: type[AsyncJsonWebsocketConsumer]

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

Extends WebsocketCommunicator with Django authentication and settings integration.

Adds Django-specific features:

wait_for_auth(): Handle Django authentication messages
assert_authenticated_status_ok(): Validate DRF authentication status
SEND_AUTHENTICATION_MESSAGE and CAMELIZE settings support

Use this for Django consumers with authentication.

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

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

Django test case for WebSocket testing with Chanx.

Integrates Chanx WebSocket testing with Django's test framework, providing:

Django TransactionTestCase inheritance for database transaction handling
Automatic Django ASGI application discovery from routing configuration
Django-style setUp/tearDown with automatic communicator cleanup
Integration with Django authentication headers via get_ws_headers()
Support for Django-specific WebSocket subprotocols

Usage: 1. Subclass WebsocketTestCase 2. Set ws_path to your Django WebSocket endpoint 3. Override get_ws_headers() for Django authentication 4. Use self.auth_communicator for primary connection testing 5. Use create_communicator() for multi-user Django scenarios

The test case automatically discovers WebSocket routing from Django's ASGI configuration and ensures proper cleanup of all connections after each test.

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

consumer: type[AsyncJsonWebsocketConsumer[Any]]

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

Creates a DjangoWebsocketCommunicator 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 DjangoWebsocketCommunicator 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.

router: Any = None

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.

ws_path: str = ''

FastAPI Testing

The FastAPI testing module provides FastAPI-specific testing utilities.

class chanx.fast_channels.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[ChanxWebsocketConsumerMixin[Any]])

Fast-channels WebSocket communicator for testing Chanx consumers.

Combines Chanx testing mixin features with fast-channels' WebSocket communicator, providing comprehensive testing capabilities for FastAPI applications:

Chanx features (from WebsocketCommunicatorMixin):

Structured message sending/receiving with BaseMessage objects
Automatic message collection until completion signals
Message validation using consumer's type adapters
Async context manager support for automatic cleanup
send_message(): Send BaseMessage objects directly
receive_all_json(): Collect all messages until timeout
receive_all_messages(): Collect and validate messages until stop action

FastAPI fast-channels features:

Full compatibility with FastAPI ASGI applications
fast-channels channel layer support
High-performance async operation

application: Any

consumer: type[AsyncJsonWebsocketConsumer]