python-tdd/src/apps/epic/tests/integrated/test_consumers.py

from channels.testing.websocket import WebsocketCommunicator
from channels.layers import get_channel_layer
from django.test import SimpleTestCase, override_settings

from core.asgi import application


TEST_CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels.layers.InMemoryChannelLayer",
    }
}


@override_settings(CHANNEL_LAYERS=TEST_CHANNEL_LAYERS)
class RoomConsumerTest(SimpleTestCase):
    async def test_can_connect_and_disconnect(self):
        communicator = WebsocketCommunicator(application, "/ws/room/00000000-0000-0000-0000-000000000001/")
        connected, _ = await communicator.connect()
        self.assertTrue(connected)
        await communicator.disconnect()

    async def test_receives_role_select_start_broadcast(self):
        communicator = WebsocketCommunicator(application, "/ws/room/00000000-0000-0000-0000-000000000001/")
        await communicator.connect()

        channel_layer = get_channel_layer()
        await channel_layer.group_send(
            "room_00000000-0000-0000-0000-000000000001",
            {"type": "role_select_start", "slot_order": [1, 2, 3, 4, 5, 6]},
        )

        response = await communicator.receive_json_from()
        self.assertEqual(response["type"], "role_select_start")
        self.assertEqual(response["slot_order"], [1, 2, 3, 4, 5, 6])

        await communicator.disconnect()

    async def test_receives_turn_changed_broadcast(self):
        communicator = WebsocketCommunicator(application, "/ws/room/00000000-0000-0000-0000-000000000001/")
        await communicator.connect()

        channel_layer = get_channel_layer()
        await channel_layer.group_send(
            "room_00000000-0000-0000-0000-000000000001",
            {"type": "turn_changed", "active_slot": 2},
        )

        response = await communicator.receive_json_from()
        self.assertEqual(response["type"], "turn_changed")
        self.assertEqual(response["active_slot"], 2)

        await communicator.disconnect()

    async def test_receives_roles_revealed_broadcast(self):
        communicator = WebsocketCommunicator(application, "/ws/room/00000000-0000-0000-0000-000000000001/")
        await communicator.connect()

        channel_layer = get_channel_layer()
        await channel_layer.group_send(
            "room_00000000-0000-0000-0000-000000000001",
            {"type": "roles_revealed", "assignments": {"1": "PC", "2": "BC"}},
        )

        response = await communicator.receive_json_from()
        self.assertEqual(response["type"], "roles_revealed")
        self.assertIn("assignments", response)

        await communicator.disconnect()

    async def test_receives_gate_update_broadcast(self):
        communicator = WebsocketCommunicator(application, "/ws/room/00000000-0000-0000-0000-000000000001/")
        await communicator.connect()

        channel_layer = get_channel_layer()
        await channel_layer.group_send(
            "room_00000000-0000-0000-0000-000000000001",
            {"type": "gate_update", "gate_state": "some_state"},
        )

        response = await communicator.receive_json_from()
        self.assertEqual(response["type"], "gate_update")
        self.assertEqual(response["gate_state"], "some_state")

        await communicator.disconnect()