58 lines
2.3 KiB
Python
58 lines
2.3 KiB
Python
from unittest import mock
|
|
|
|
from django.core.cache import cache
|
|
from django.test import TestCase, override_settings
|
|
|
|
from events.models import Event
|
|
from games.models import EventGame, Game
|
|
|
|
from .tasks import HEARTBEAT_LOCK_KEY, clock_heartbeat_task, start_clock_heartbeat
|
|
|
|
LOCMEM_CACHE = {"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}}
|
|
|
|
|
|
@override_settings(CACHES=LOCMEM_CACHE)
|
|
@mock.patch("clock.tasks.get_channel_layer")
|
|
class ClockHeartbeatTests(TestCase):
|
|
def setUp(self):
|
|
cache.clear()
|
|
event = Event.objects.create(name="Event", path="event")
|
|
clock = Game.objects.create(name="Clock", path="clock", custom_path=True)
|
|
self.event_game = EventGame.objects.create(event=event, game=clock)
|
|
cache.clear() # post_save signal can start a chain
|
|
|
|
@mock.patch("clock.tasks.clock_heartbeat_task")
|
|
def test_only_one_chain_is_started(self, task, _):
|
|
start_clock_heartbeat()
|
|
start_clock_heartbeat()
|
|
|
|
task.enqueue.assert_called_once_with(cache.get(HEARTBEAT_LOCK_KEY))
|
|
|
|
def test_owning_chain_reschedules_itself(self, channel_layer):
|
|
channel_layer.return_value.group_send = mock.AsyncMock()
|
|
cache.set(HEARTBEAT_LOCK_KEY, "chain")
|
|
with mock.patch("clock.tasks.clock_heartbeat_task") as patched:
|
|
clock_heartbeat_task.call("chain")
|
|
|
|
patched.using.return_value.enqueue.assert_called_once_with("chain")
|
|
channel_layer.return_value.group_send.assert_awaited_once()
|
|
self.assertEqual(cache.get(HEARTBEAT_LOCK_KEY), "chain")
|
|
|
|
def test_superseded_chain_stops(self, channel_layer):
|
|
cache.set(HEARTBEAT_LOCK_KEY, "new-chain")
|
|
with mock.patch("clock.tasks.clock_heartbeat_task") as patched:
|
|
clock_heartbeat_task.call("old-chain")
|
|
|
|
patched.using.assert_not_called()
|
|
channel_layer.assert_not_called()
|
|
self.assertEqual(cache.get(HEARTBEAT_LOCK_KEY), "new-chain")
|
|
|
|
def test_chain_dies_and_releases_lock_when_game_disabled(self, _):
|
|
EventGame.objects.filter(pk=self.event_game.pk).update(enabled=False)
|
|
cache.set(HEARTBEAT_LOCK_KEY, "chain")
|
|
with mock.patch("clock.tasks.clock_heartbeat_task") as patched:
|
|
clock_heartbeat_task.call("chain")
|
|
|
|
patched.using.assert_not_called()
|
|
self.assertIsNone(cache.get(HEARTBEAT_LOCK_KEY))
|