from unittest.mock import patch, MagicMock from django.test import TestCase from apps.epic.models import Room, SigReservation, TableSeat, TarotCard from apps.lyric.models import User from apps.epic.tasks import ( _cache_key, cancel_polarity_confirm, schedule_polarity_confirm, ) class CacheKeyTest(TestCase): def test_cache_key_format(self): self.assertEqual(_cache_key("room-1", "levity"), "sig_countdown_room-1_levity") class CancelPolarityConfirmTest(TestCase): def setUp(self): self.user = User.objects.create(email="owner@tasks.io") self.room = Room.objects.create(name="R", owner=self.user) def test_cancel_with_no_timer_is_a_noop(self): cancel_polarity_confirm(str(self.room.id), SigReservation.LEVITY) def test_cancel_clears_cache_entry(self): from django.core.cache import cache key = _cache_key(str(self.room.id), SigReservation.LEVITY) cache.set(key, "sometoken", timeout=60) cancel_polarity_confirm(str(self.room.id), SigReservation.LEVITY) self.assertIsNone(cache.get(key)) @patch("apps.epic.tasks._timers") def test_cancel_calls_timer_cancel_when_present(self, mock_timers): mock_timer = MagicMock() key = f"{self.room.id}_levity" mock_timers.pop.return_value = mock_timer cancel_polarity_confirm(str(self.room.id), SigReservation.LEVITY) mock_timer.cancel.assert_called_once() class FireFunctionTest(TestCase): """Tests for the _fire() callback executed by threading.Timer.""" def setUp(self): self.owner = User.objects.create(email="owner@fire.io") self.room = Room.objects.create(name="R", owner=self.owner) self.room.table_status = Room.SIG_SELECT self.room.save() roles = ["PC", "NC", "SC"] self.gamers = [self.owner] for i, role in enumerate(roles): if i == 0: TableSeat.objects.create(room=self.room, gamer=self.owner, slot_number=1, role=role) else: g = User.objects.create(email=f"g{i}@fire.io") self.gamers.append(g) TableSeat.objects.create(room=self.room, gamer=g, slot_number=i+1, role=role) # Gravity seats (no significators needed for levity test) grav_roles = ["BC", "EC", "AC"] for i, role in enumerate(grav_roles, start=4): g = User.objects.create(email=f"grav{i}@fire.io") TableSeat.objects.create(room=self.room, gamer=g, slot_number=i, role=role) def _set_token(self): from django.core.cache import cache import uuid token = str(uuid.uuid4()) cache.set(_cache_key(str(self.room.id), SigReservation.LEVITY), token, 120) return token @patch("apps.epic.tasks._group_send") def test_fire_does_nothing_if_token_mismatch(self, mock_send): from apps.epic.tasks import _fire self._set_token() _fire(str(self.room.id), SigReservation.LEVITY, "wrong-token") mock_send.assert_not_called() @patch("apps.epic.tasks._group_send") def test_fire_does_nothing_if_room_not_sig_select(self, mock_send): from apps.epic.tasks import _fire token = self._set_token() self.room.table_status = Room.ROLE_SELECT self.room.save() _fire(str(self.room.id), SigReservation.LEVITY, token) mock_send.assert_not_called() @patch("apps.epic.tasks._group_send") def test_fire_does_nothing_if_fewer_than_3_ready(self, mock_send): from apps.epic.tasks import _fire token = self._set_token() cards = list(TarotCard.objects.all()[:2]) seats = list(TableSeat.objects.filter(room=self.room, role__in=["PC", "NC"])) for i, seat in enumerate(seats): SigReservation.objects.create( room=self.room, gamer=seat.gamer, card=cards[i], polarity=SigReservation.LEVITY, seat=seat, ready=True, ) _fire(str(self.room.id), SigReservation.LEVITY, token) mock_send.assert_not_called() @patch("apps.epic.tasks._group_send") def test_fire_assigns_significators_and_broadcasts_when_all_ready(self, mock_send): from apps.epic.tasks import _fire token = self._set_token() cards = list(TarotCard.objects.all()[:3]) seats = list(TableSeat.objects.filter(room=self.room, role__in=["PC", "NC", "SC"])) for i, seat in enumerate(seats): SigReservation.objects.create( room=self.room, gamer=seat.gamer, card=cards[i], polarity=SigReservation.LEVITY, seat=seat, ready=True, ) _fire(str(self.room.id), SigReservation.LEVITY, token) self.assertTrue(mock_send.called) levity_seats = TableSeat.objects.filter(room=self.room, role__in=["PC", "NC", "SC"]) for i, seat in enumerate(levity_seats): seat.refresh_from_db() self.assertEqual(seat.significator, cards[i]) def test_fire_does_nothing_for_nonexistent_room(self): from apps.epic.tasks import _fire from django.core.cache import cache fake_id = "00000000-0000-0000-0000-000000000000" token = "known-token" cache.set(_cache_key(fake_id, SigReservation.LEVITY), token, 60) _fire(fake_id, SigReservation.LEVITY, token) @patch("apps.epic.tasks._group_send") def test_fire_does_nothing_if_all_sigs_already_assigned(self, mock_send): from apps.epic.tasks import _fire token = self._set_token() cards = list(TarotCard.objects.all()[:3]) seats = list(TableSeat.objects.filter(room=self.room, role__in=["PC", "NC", "SC"])) for i, seat in enumerate(seats): seat.significator = cards[i] seat.save(update_fields=["significator"]) _fire(str(self.room.id), SigReservation.LEVITY, token) mock_send.assert_not_called() @patch("apps.epic.tasks._group_send") def test_fire_broadcasts_pick_sky_when_all_polarity_sigs_assigned(self, mock_send): """When both levity AND gravity seats all have significators, fire() triggers SKY_SELECT.""" from apps.epic.tasks import _fire token = self._set_token() cards = list(TarotCard.objects.all()[:6]) # Give gravity seats significators so the all-assigned check passes gravity_seats = list(TableSeat.objects.filter(room=self.room, role__in=["BC", "EC", "AC"])) for i, seat in enumerate(gravity_seats): seat.significator = cards[i] seat.save(update_fields=["significator"]) # Create ready levity reservations (different cards from gravity) levity_seats = list(TableSeat.objects.filter(room=self.room, role__in=["PC", "NC", "SC"])) levity_cards = cards[3:6] for i, seat in enumerate(levity_seats): SigReservation.objects.create( room=self.room, gamer=seat.gamer, card=levity_cards[i], polarity=SigReservation.LEVITY, seat=seat, ready=True, ) _fire(str(self.room.id), SigReservation.LEVITY, token) call_types = [c.args[1]["type"] for c in mock_send.call_args_list] self.assertIn("polarity_room_done", call_types) self.assertIn("pick_sky_available", call_types) self.room.refresh_from_db() self.assertEqual(self.room.table_status, "SKY_SELECT") class SchedulePolarityConfirmTest(TestCase): def setUp(self): self.user = User.objects.create(email="owner@schedule.io") self.room = Room.objects.create(name="R", owner=self.user) def test_schedule_sets_cache_token(self): from django.core.cache import cache schedule_polarity_confirm(str(self.room.id), SigReservation.LEVITY, 60) token = cache.get(_cache_key(str(self.room.id), SigReservation.LEVITY)) self.assertIsNotNone(token) def test_schedule_registers_timer(self): from apps.epic.tasks import _timers key = f"{self.room.id}_{SigReservation.LEVITY}" schedule_polarity_confirm(str(self.room.id), SigReservation.LEVITY, 60) self.assertIn(key, _timers) _timers[key].cancel() # clean up def test_schedule_cancels_prior_timer_before_scheduling(self): from apps.epic.tasks import _timers schedule_polarity_confirm(str(self.room.id), SigReservation.LEVITY, 60) key = f"{self.room.id}_{SigReservation.LEVITY}" first_timer = _timers.get(key) schedule_polarity_confirm(str(self.room.id), SigReservation.LEVITY, 60) second_timer = _timers.get(key) self.assertIsNotNone(second_timer) self.assertIsNot(first_timer, second_timer) second_timer.cancel() class GroupSendTest(TestCase): @patch("apps.epic.tasks.async_to_sync") def test_group_send_calls_async_to_sync(self, mock_a2s): from apps.epic.tasks import _group_send mock_fn = MagicMock() mock_a2s.return_value = mock_fn _group_send("room-abc", {"type": "test"}) mock_a2s.assert_called_once() mock_fn.assert_called_once_with("room_room-abc", {"type": "test"})