86 lines
3.2 KiB
Python
86 lines
3.2 KiB
Python
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()
|