124 lines
4.6 KiB
Python
124 lines
4.6 KiB
Python
import unittest
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
from channels.exceptions import InvalidChannelLayerError
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import SimpleTestCase, TestCase
|
|
|
|
try:
|
|
from channels.testing import WebsocketCommunicator
|
|
except Exception: # pragma: no cover - optional test dependency
|
|
WebsocketCommunicator = None
|
|
|
|
from fupogfakta.models import GameSession, Player
|
|
from partyhub.asgi import application
|
|
from realtime.broadcast import broadcast_phase_event, sync_broadcast_phase_event
|
|
from realtime.consumers import GameConsumer
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class BroadcastPhaseEventTests(SimpleTestCase):
|
|
@patch("realtime.broadcast.get_channel_layer", return_value=None)
|
|
async def test_broadcast_phase_event_noops_without_channel_layer(self, _mock_get_channel_layer):
|
|
await broadcast_phase_event("ABCD", "phase.scoreboard", {"phase": "scoreboard"})
|
|
|
|
@patch("realtime.broadcast.async_to_sync")
|
|
def test_sync_broadcast_phase_event_noops_when_channel_layer_is_unavailable(self, mock_async_to_sync):
|
|
mock_async_to_sync.return_value.side_effect = InvalidChannelLayerError("missing channel layer")
|
|
|
|
sync_broadcast_phase_event("ABCD", "phase.scoreboard", {"phase": "scoreboard"})
|
|
|
|
@patch("realtime.broadcast.async_to_sync")
|
|
def test_sync_broadcast_phase_event_still_broadcasts_when_channel_layer_exists(self, mock_async_to_sync):
|
|
sender = Mock()
|
|
mock_async_to_sync.return_value = sender
|
|
|
|
sync_broadcast_phase_event("ABCD", "phase.scoreboard", {"phase": "scoreboard"})
|
|
|
|
sender.assert_called_once_with("ABCD", "phase.scoreboard", {"phase": "scoreboard"})
|
|
|
|
|
|
class GameConsumerPhaseEventTests(SimpleTestCase):
|
|
async def test_phase_event_restores_external_type_field(self):
|
|
consumer = GameConsumer()
|
|
consumer.send_json = AsyncMock()
|
|
|
|
await consumer.phase_event(
|
|
{
|
|
"event_type": "phase.test_event",
|
|
"payload": {"hello": "world"},
|
|
}
|
|
)
|
|
|
|
consumer.send_json.assert_awaited_once_with(
|
|
{
|
|
"type": "phase.test_event",
|
|
"hello": "world",
|
|
}
|
|
)
|
|
|
|
|
|
@unittest.skipIf(WebsocketCommunicator is None, "channels.testing dependencies unavailable")
|
|
class GameConsumerConnectTest(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username="host", password="pw")
|
|
self.session = GameSession.objects.create(host=self.user, code="AABBCC")
|
|
self.player = Player.objects.create(session=self.session, nickname="Tester")
|
|
|
|
async def test_player_connect_and_ping(self):
|
|
token = self.player.session_token
|
|
communicator = WebsocketCommunicator(
|
|
application,
|
|
f"/ws/game/AABBCC/?session_token={token}",
|
|
)
|
|
connected, _ = await communicator.connect()
|
|
self.assertTrue(connected)
|
|
|
|
await communicator.send_json_to({"type": "ping"})
|
|
response = await communicator.receive_json_from()
|
|
self.assertEqual(response["type"], "pong")
|
|
|
|
await communicator.disconnect()
|
|
|
|
async def test_connect_without_token_rejected(self):
|
|
communicator = WebsocketCommunicator(application, "/ws/game/AABBCC/")
|
|
connected, code = await communicator.connect()
|
|
self.assertFalse(connected)
|
|
self.assertEqual(code, 4001)
|
|
|
|
async def test_connect_invalid_token_rejected(self):
|
|
communicator = WebsocketCommunicator(
|
|
application,
|
|
"/ws/game/AABBCC/?session_token=invalid-token",
|
|
)
|
|
connected, code = await communicator.connect()
|
|
self.assertFalse(connected)
|
|
self.assertEqual(code, 4003)
|
|
|
|
async def test_host_connect_without_token(self):
|
|
communicator = WebsocketCommunicator(
|
|
application,
|
|
"/ws/game/AABBCC/?role=host",
|
|
)
|
|
connected, _ = await communicator.connect()
|
|
self.assertTrue(connected)
|
|
await communicator.disconnect()
|
|
|
|
async def test_broadcast_reaches_connected_client(self):
|
|
token = self.player.session_token
|
|
communicator = WebsocketCommunicator(
|
|
application,
|
|
f"/ws/game/AABBCC/?session_token={token}",
|
|
)
|
|
connected, _ = await communicator.connect()
|
|
self.assertTrue(connected)
|
|
|
|
await broadcast_phase_event("AABBCC", "phase.test_event", {"hello": "world"})
|
|
|
|
message = await communicator.receive_json_from(timeout=2)
|
|
self.assertEqual(message["type"], "phase.test_event")
|
|
self.assertEqual(message["hello"], "world")
|
|
|
|
await communicator.disconnect()
|