28 lines
1.3 KiB
Python
28 lines
1.3 KiB
Python
from unittest.mock import Mock, patch
|
|
|
|
from channels.exceptions import InvalidChannelLayerError
|
|
from django.test import SimpleTestCase
|
|
|
|
from realtime.broadcast import broadcast_phase_event, sync_broadcast_phase_event
|
|
|
|
|
|
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"})
|