Testing

Chanx provides specialized testing utilities that make it easier to write comprehensive tests for WebSocket consumers. These tools handle connection management, authentication, message exchange, and test cleanup.

Testing Overview

Testing WebSocket consumers differs from testing regular HTTP views:

Connections are long-lived instead of request/response
Authentication happens once at connection time
Multiple messages can be exchanged over a single connection
Asynchronous code requires special testing approaches
Group messaging requires multiple client testing

Chanx addresses these challenges with the WebsocketTestCase class and enhanced WebsocketCommunicator.

WebsocketTestCase

The WebsocketTestCase class extends Django's TransactionTestCase with WebSocket-specific functionality:

from chanx.testing import WebsocketTestCase

class TestMyConsumer(WebsocketTestCase):
    # Path to test (required)
    ws_path = "/ws/myendpoint/"

    async def test_connect(self):
        # Use the default authenticator communicator
        await self.auth_communicator.connect()

        # Verify connection was successful
        await self.auth_communicator.assert_authenticated_status_ok()

Key features of WebsocketTestCase:

Automatic Router Discovery: Finds your WebSocket application from ASGI configuration
Connection Tracking: Manages test communicators to ensure proper cleanup
Helper Methods: Provides utilities for common testing tasks
Default auth_communicator: Access the main communicator via self.auth_communicator
Multi-user testing support: Create additional communicators as needed

Authentication in Testing

There are several ways to implement authentication in your tests. Here's an example of JWT-based authentication in a custom test case:

from django.conf import settings
from accounts.factories.user import UserFactory
from accounts.models import User
from asgiref.sync import sync_to_async
from chanx.testing import WebsocketTestCase as BaseWebsocketTestCase
from rest_framework_simplejwt.tokens import RefreshToken


class WebsocketTestCase(BaseWebsocketTestCase):
    def setUp(self) -> None:
        # Create a user and authentication headers during setup
        self.user, self.ws_headers = self.create_user_and_ws_headers()
        super().setUp()

    def create_user_and_ws_headers(self) -> tuple[User, list[tuple[bytes, bytes]]]:
        # Create a user and generate JWT tokens
        user = UserFactory.create()
        user_refresh_token = RefreshToken.for_user(user)

        # Create cookie string with JWT tokens
        cookie_string = (
            f"jwt_auth_cookie={str(user_refresh_token.access_token)}; "
            f"jwt_auth_refresh_cookie={str(user_refresh_token)}"
        )

        # Create WebSocket headers with the cookie and other required headers
        ws_headers = [
            (b"cookie", cookie_string.encode()),
            (b"origin", settings.SERVER_URL.encode()),
            (b"x-forwarded-for", b"127.0.0.1"),
        ]
        return user, ws_headers

    async def acreate_user_and_ws_headers(self) -> tuple[User, list[tuple[bytes, bytes]]]:
        """Async version for creating users during tests"""
        return await sync_to_async(self.create_user_and_ws_headers)()

    def get_ws_headers(self) -> list[tuple[bytes, bytes]]:
        """Provide headers for the default auth_communicator"""
        return self.ws_headers

For session-based authentication, you can use Django's test client:

def get_ws_headers(self):
    # Create a session using Django's test client
    self.client.login(username="testuser", password="password")

    # Get the session cookie
    cookies = self.client.cookies
    return [
        (b"cookie", f"sessionid={cookies['sessionid'].value}".encode()),
    ]

Creating Multiple Communicators

For testing scenarios with multiple users, use the create_communicator method:

async def test_multi_user_scenario(self) -> None:
    # Get the default communicator for the first user
    first_comm = self.auth_communicator

    # Create a second user with different auth headers
    second_user, second_ws_headers = await self.acreate_user_and_ws_headers()

    # Create a communicator for the second user
    second_comm = self.create_communicator(
        headers=second_ws_headers,
    )

    # Connect both communicators
    await first_comm.connect()
    await first_comm.assert_authenticated_status_ok()

    await second_comm.connect()
    await second_comm.assert_authenticated_status_ok()

    # Test interactions between the users
    # ...

The create_communicator method is essential for multi-user testing. It:

Creates WebsocketCommunicator instances with custom configuration
Automatically tracks communicators for proper cleanup
Supports custom headers for authentication
Lets you test group messaging scenarios

WebsocketCommunicator Features

Chanx extends the standard Channels WebsocketCommunicator with additional features:

# Connect with timeout
connected, _ = await communicator.connect(timeout=3)

# Wait for authentication message
auth_message = await communicator.wait_for_auth()

# Assert authentication succeeded
await communicator.assert_authenticated_status_ok()

# Send message objects directly
from myapp.messages import ChatMessage
await communicator.send_message(ChatMessage(payload="Hello"))

# Receive all messages until completion
messages = await communicator.receive_all_json()

# Receive messages including group completion
messages = await communicator.receive_all_json(wait_group=True)

# Verify connection closed properly
await communicator.assert_closed()

Testing Message Exchange

Here's a complete example of testing message exchange with modern Python assertions:

from typing import Any, cast
from chanx.messages.incoming import PingMessage
from chanx.messages.outgoing import PongMessage
from myapp.messages import ChatMessage, ChatResponse

class TestChatConsumer(WebsocketTestCase):
    ws_path = "/ws/chat/room1/"

    async def test_ping_pong(self) -> None:
        # Connect and authenticate
        await self.auth_communicator.connect()
        await self.auth_communicator.assert_authenticated_status_ok()

        # Send ping message
        await self.auth_communicator.send_message(PingMessage())

        # Receive all messages until completion
        responses = await self.auth_communicator.receive_all_json()

        # Check for pong response
        assert len(responses) == 1

        # You can either check raw JSON
        assert responses[0]["action"] == "pong"

        # Or validate with the message model
        pong_message = PongMessage.model_validate(responses[0])
        assert isinstance(pong_message, PongMessage)

    async def test_chat_message(self) -> None:
        await self.auth_communicator.connect()
        await self.auth_communicator.assert_authenticated_status_ok()

        # Send chat message
        message_content = "Test message"
        await self.auth_communicator.send_message(
            ChatMessage(payload={"content": message_content})
        )

        # Get responses up to completion marker
        responses = await self.auth_communicator.receive_all_json()

        # Verify the response
        assert len(responses) == 1
        response = responses[0]
        assert response["action"] == "chat_response"
        assert response["payload"]["content"] == f"Echo: {message_content}"

Testing Group Messaging

Use multiple communicators to test group messaging:

async def test_group_message_broadcast(self) -> None:
    """Test that messages are broadcast to all group members"""
    # Create a second user with different auth headers
    second_user, second_ws_headers = await self.acreate_user_and_ws_headers()

    # Create communicators for both users in the same room
    first_comm = self.auth_communicator
    second_comm = self.create_communicator(headers=second_ws_headers)

    # Connect both communicators
    await first_comm.connect()
    await first_comm.assert_authenticated_status_ok()

    await second_comm.connect()
    await second_comm.assert_authenticated_status_ok()

    # Send a message from the first user
    message_content = "This is a group message"
    await first_comm.send_message(
        ChatMessage(payload={"content": message_content})
    )

    # Verify that the first user (sender) receives the message
    first_responses = await first_comm.receive_all_json(wait_group=True)
    assert len(first_responses) == 1
    assert first_responses[0]["action"] == "chat_group"
    assert first_responses[0]["payload"]["content"] == message_content
    assert first_responses[0]["is_mine"] == True  # Sent by this user

    # Verify that the second user receives the same message
    second_responses = await second_comm.receive_all_json(wait_group=True)
    assert len(second_responses) == 1
    assert second_responses[0]["action"] == "chat_group"
    assert second_responses[0]["payload"]["content"] == message_content
    assert second_responses[0]["is_mine"] == False  # Not sent by this user

Testing Object Permissions

Test consumer access with object-level permissions:

async def test_room_access_permission(self) -> None:
    """Test that only room members can access the room consumer"""
    # Create a room and add the default user as a member
    room = await Room.objects.acreate(name="Test Room")
    await RoomMember.objects.acreate(room=room, user=self.user)

    # Create a non-member user
    non_member, non_member_headers = await self.acreate_user_and_ws_headers()

    # Test successful access with member
    member_comm = self.auth_communicator
    room_path = f"/ws/rooms/{room.id}/"
    connected, _ = await member_comm.connect(ws_path=room_path)
    assert connected == True

    # Verify authentication succeeded
    auth_message = await member_comm.wait_for_auth()
    assert auth_message.payload.status_code == 200

    # Test failed access with non-member
    non_member_comm = self.create_communicator(headers=non_member_headers)
    connected, _ = await non_member_comm.connect(ws_path=room_path)
    assert connected == True  # Initial connection succeeds

    # But authentication fails due to permission check
    auth_message = await non_member_comm.wait_for_auth()
    assert auth_message.payload.status_code == 403

    # Connection should be closed
    await non_member_comm.assert_closed()

Mocking in WebSocket Tests

For isolated tests, mock external dependencies:

from unittest.mock import patch, AsyncMock

async def test_database_integration(self) -> None:
    # Mock the database operation
    with patch('myapp.services.message_service.save_message') as mock_save:
        mock_save.return_value = AsyncMock(id=123, content="Test")

        # Connect and send a message
        await self.auth_communicator.connect()
        await self.auth_communicator.assert_authenticated_status_ok()

        await self.auth_communicator.send_message(
            ChatMessage(payload={"content": "Test message"})
        )

        # Verify the mock was called
        mock_save.assert_called_once()
        args, kwargs = mock_save.call_args
        assert kwargs["content"] == "Test message"

        # Check response
        responses = await self.auth_communicator.receive_all_json()
        assert len(responses) == 1

Testing Custom Apps

Here's a complete example of a test for a chat application with custom test case:

from typing import Any, cast
from chanx.testing import WebsocketTestCase
from chat.messages.chat import ChatIncomingMessage, NewChatMessage, MessagePayload
from chat.messages.group import MemberMessage
from chat.models import ChatMember, ChatMessage, GroupChat

class ChatTestCase(WebsocketTestCase):
    async def setUp(self) -> None:
        await super().setUp()
        # Create a group chat and add the user as a member
        self.group_chat = await GroupChat.objects.acreate(name="Test Group")
        self.member = await ChatMember.objects.acreate(
            user=self.user,
            group_chat=self.group_chat,
            chat_role=ChatMember.ChatMemberRole.ADMIN,
        )

    async def test_connect_and_send_message(self) -> None:
        """Test connection and sending a message to a group chat"""
        # Connect to the chat endpoint
        self.ws_path = f"/ws/chat/{self.group_chat.pk}/"
        await self.auth_communicator.connect()
        await self.auth_communicator.assert_authenticated_status_ok()

        # Test sending a chat message
        message_content = "Hello group chat!"
        await self.auth_communicator.send_message(
            NewChatMessage(payload=MessagePayload(content=message_content))
        )

        # Receive the message that was broadcast
        messages = await self.auth_communicator.receive_all_json(wait_group=True)

        # Check the message was received and has the correct content
        assert len(messages) == 1
        assert messages[0]["action"] == "member_message"
        assert messages[0]["payload"]["content"] == message_content

        # Verify the message was stored in the database
        db_messages = await ChatMessage.objects.acount()
        assert db_messages == 1

Best Practices

Subclass WebsocketTestCase: Create a custom test base class for your app
Set up authenticating fixtures: Provide proper authentication in setUp
Use modern assert statements: Use Python's built-in assert for cleaner tests
Test both success and failure: Verify both positive and negative cases
Test group broadcasts: Create multiple communicators to test group messaging
Use wait_group=True: When testing group messages, use the wait_group parameter
Mock external services: Use AsyncMock for external dependencies
Test database persistence: Verify messages are properly stored/retrieved
Test lifecycle events: Check connections, authentication, and disconnections
Use async test methods: Write all test methods as async coroutines

Next Steps

Consumers - Learn about WebSocket consumers
Messages System - Understand message validation
WebSocket Playground - Try the interactive WebSocket playground
Chat Application Example - See complete test examples in the chat application