COVERAGE: patch 91% → 96%+ — 603 tests, tasks.py at 100%
New/extended tests across billboard, dashboard, drama, epic, gameboard, and lyric to cover previously untested branches: dev_login view, scroll position endpoints, sky preview error paths, drama to_prose/to_activity branches, consumer broadcast handlers, tarot deck draw/shuffle, astrology model __str__, character model, sig reserve/ready/confirm views, natus preview/save views, and the full tasks.py countdown scheduler. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Disco DeDisco
@@ -166,6 +166,65 @@ class CursorMoveConsumerTest(TransactionTestCase):
|
||||
await bc_comm.disconnect()
|
||||
|
||||
|
||||
@override_settings(CHANNEL_LAYERS=TEST_CHANNEL_LAYERS)
|
||||
class MissingConsumerHandlersTest(SimpleTestCase):
|
||||
"""Covers the simple pass-through handlers not exercised by other tests."""
|
||||
|
||||
async def _send_and_receive(self, room_path, group_name, msg):
|
||||
communicator = WebsocketCommunicator(application, room_path)
|
||||
await communicator.connect()
|
||||
channel_layer = get_channel_layer()
|
||||
await channel_layer.group_send(group_name, msg)
|
||||
response = await communicator.receive_json_from()
|
||||
await communicator.disconnect()
|
||||
return response
|
||||
|
||||
async def test_receives_sig_selected_broadcast(self):
|
||||
room_id = "00000000-0000-0000-0000-000000000002"
|
||||
response = await self._send_and_receive(
|
||||
f"/ws/room/{room_id}/",
|
||||
f"room_{room_id}",
|
||||
{"type": "sig_selected", "card_id": "abc"},
|
||||
)
|
||||
self.assertEqual(response["type"], "sig_selected")
|
||||
|
||||
async def test_receives_countdown_start_broadcast(self):
|
||||
room_id = "00000000-0000-0000-0000-000000000002"
|
||||
response = await self._send_and_receive(
|
||||
f"/ws/room/{room_id}/",
|
||||
f"room_{room_id}",
|
||||
{"type": "countdown_start", "polarity": "levity", "seconds": 12},
|
||||
)
|
||||
self.assertEqual(response["type"], "countdown_start")
|
||||
|
||||
async def test_receives_countdown_cancel_broadcast(self):
|
||||
room_id = "00000000-0000-0000-0000-000000000002"
|
||||
response = await self._send_and_receive(
|
||||
f"/ws/room/{room_id}/",
|
||||
f"room_{room_id}",
|
||||
{"type": "countdown_cancel", "polarity": "levity", "seconds_remaining": 7},
|
||||
)
|
||||
self.assertEqual(response["type"], "countdown_cancel")
|
||||
|
||||
async def test_receives_polarity_room_done_broadcast(self):
|
||||
room_id = "00000000-0000-0000-0000-000000000002"
|
||||
response = await self._send_and_receive(
|
||||
f"/ws/room/{room_id}/",
|
||||
f"room_{room_id}",
|
||||
{"type": "polarity_room_done", "polarity": "levity"},
|
||||
)
|
||||
self.assertEqual(response["type"], "polarity_room_done")
|
||||
|
||||
async def test_receives_pick_sky_available_broadcast(self):
|
||||
room_id = "00000000-0000-0000-0000-000000000002"
|
||||
response = await self._send_and_receive(
|
||||
f"/ws/room/{room_id}/",
|
||||
f"room_{room_id}",
|
||||
{"type": "pick_sky_available"},
|
||||
)
|
||||
self.assertEqual(response["type"], "pick_sky_available")
|
||||
|
||||
|
||||
@tag('channels')
|
||||
@override_settings(CHANNEL_LAYERS=TEST_CHANNEL_LAYERS)
|
||||
class SigHoverConsumerTest(TransactionTestCase):
|
||||
|
||||
@@ -8,7 +8,8 @@ from django.db import IntegrityError
|
||||
|
||||
from apps.lyric.models import Token, User
|
||||
from apps.epic.models import (
|
||||
DeckVariant, GateSlot, Room, RoomInvite, SigReservation, TableSeat, TarotCard,
|
||||
AspectType, Character, DeckVariant, GateSlot, HouseLabel, Planet, Room, RoomInvite,
|
||||
SigReservation, Sign, TableSeat, TarotCard,
|
||||
debit_token, select_token, sig_deck_cards, levity_sig_cards, gravity_sig_cards,
|
||||
sig_seat_order, active_sig_seat,
|
||||
)
|
||||
@@ -593,3 +594,112 @@ class RoomSkySelectStatusTest(TestCase):
|
||||
self.room.save()
|
||||
self.room.refresh_from_db()
|
||||
self.assertIsNotNone(self.room.sig_select_started_at)
|
||||
|
||||
|
||||
# ── TarotDeck.draw / shuffle ──────────────────────────────────────────────────
|
||||
|
||||
class TarotDeckDrawTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="dealer@test.io")
|
||||
self.room = Room.objects.create(name="R", owner=self.user)
|
||||
|
||||
def test_draw_raises_value_error_when_too_few_cards_remain(self):
|
||||
from apps.epic.models import TarotDeck
|
||||
deck_variant = DeckVariant.objects.first()
|
||||
all_ids = list(TarotCard.objects.filter(deck_variant=deck_variant).values_list('id', flat=True))
|
||||
td = TarotDeck.objects.create(
|
||||
room=self.room,
|
||||
deck_variant=deck_variant,
|
||||
drawn_card_ids=all_ids,
|
||||
)
|
||||
with self.assertRaises(ValueError):
|
||||
td.draw(1)
|
||||
|
||||
def test_shuffle_resets_drawn_card_ids(self):
|
||||
from apps.epic.models import TarotDeck
|
||||
deck_variant = DeckVariant.objects.first()
|
||||
some_ids = list(TarotCard.objects.filter(deck_variant=deck_variant).values_list('id', flat=True)[:3])
|
||||
td = TarotDeck.objects.create(
|
||||
room=self.room,
|
||||
deck_variant=deck_variant,
|
||||
drawn_card_ids=some_ids,
|
||||
)
|
||||
td.shuffle()
|
||||
td.refresh_from_db()
|
||||
self.assertEqual(td.drawn_card_ids, [])
|
||||
|
||||
|
||||
# ── sig_deck_cards with no equipped deck ─────────────────────────────────────
|
||||
|
||||
class SigDeckCardsNoEquippedDeckTest(TestCase):
|
||||
def test_returns_empty_list_when_owner_has_no_equipped_deck(self):
|
||||
user = User.objects.create(email="nodeck@test.io")
|
||||
user.equipped_deck = None
|
||||
user.save(update_fields=["equipped_deck"])
|
||||
room = Room.objects.create(name="R", owner=user)
|
||||
self.assertEqual(sig_deck_cards(room), [])
|
||||
|
||||
|
||||
# ── Astrology model __str__ methods ──────────────────────────────────────────
|
||||
|
||||
class AstrologyModelStrTest(TestCase):
|
||||
def test_zodiac_sign_str(self):
|
||||
sign = Sign.objects.first()
|
||||
if sign is None:
|
||||
self.skipTest("No Sign rows")
|
||||
self.assertEqual(str(sign), sign.name)
|
||||
|
||||
def test_planet_str(self):
|
||||
planet = Planet.objects.first()
|
||||
if planet is None:
|
||||
self.skipTest("No Planet rows")
|
||||
self.assertEqual(str(planet), planet.name)
|
||||
|
||||
def test_aspect_type_str(self):
|
||||
aspect = AspectType.objects.first()
|
||||
if aspect is None:
|
||||
self.skipTest("No AspectType rows")
|
||||
self.assertEqual(str(aspect), aspect.name)
|
||||
|
||||
def test_house_label_str(self):
|
||||
label = HouseLabel.objects.first()
|
||||
if label is None:
|
||||
self.skipTest("No HouseLabel rows")
|
||||
self.assertIn(str(label.number), str(label))
|
||||
|
||||
|
||||
# ── Character model ───────────────────────────────────────────────────────────
|
||||
|
||||
class CharacterModelTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="char@test.io")
|
||||
self.room = Room.objects.create(name="R", owner=self.user)
|
||||
self.seat = TableSeat.objects.create(room=self.room, gamer=self.user, slot_number=1, role="PC")
|
||||
|
||||
def test_draft_str(self):
|
||||
char = Character.objects.create(seat=self.seat)
|
||||
self.assertIn("draft", str(char))
|
||||
|
||||
def test_confirmed_str(self):
|
||||
char = Character.objects.create(seat=self.seat, confirmed_at=timezone.now())
|
||||
self.assertIn("confirmed", str(char))
|
||||
|
||||
def test_is_confirmed_false_for_draft(self):
|
||||
char = Character.objects.create(seat=self.seat)
|
||||
self.assertFalse(char.is_confirmed)
|
||||
|
||||
def test_is_confirmed_true_when_confirmed_at_set(self):
|
||||
char = Character.objects.create(seat=self.seat, confirmed_at=timezone.now())
|
||||
self.assertTrue(char.is_confirmed)
|
||||
|
||||
def test_is_active_true_when_confirmed_and_not_retired(self):
|
||||
char = Character.objects.create(seat=self.seat, confirmed_at=timezone.now())
|
||||
self.assertTrue(char.is_active)
|
||||
|
||||
def test_is_active_false_when_retired(self):
|
||||
char = Character.objects.create(
|
||||
seat=self.seat,
|
||||
confirmed_at=timezone.now(),
|
||||
retired_at=timezone.now(),
|
||||
)
|
||||
self.assertFalse(char.is_active)
|
||||
|
||||
@@ -1612,3 +1612,278 @@ class PickSkyRenderingTest(TestCase):
|
||||
response = self.client.get(self.url)
|
||||
self.assertContains(response, 'id="id_pick_sky_btn"')
|
||||
self.assertContains(response, 'style="display:none"')
|
||||
|
||||
|
||||
# ── select_role GET redirect ──────────────────────────────────────────────────
|
||||
|
||||
class SelectRoleGetRedirectTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="gamer@sr.io")
|
||||
self.client.force_login(self.user)
|
||||
self.room = Room.objects.create(name="R", owner=self.user)
|
||||
self.room.table_status = Room.ROLE_SELECT
|
||||
self.room.save()
|
||||
|
||||
def test_get_redirects_to_room(self):
|
||||
response = self.client.get(reverse("epic:select_role", kwargs={"room_id": self.room.id}))
|
||||
self.assertRedirects(response, reverse("epic:room", kwargs={"room_id": self.room.id}),
|
||||
fetch_redirect_response=False)
|
||||
|
||||
|
||||
# ── sig_reserve / sig_ready / sig_confirm / select_sig helpers ────────────────
|
||||
|
||||
def _make_sig_room(owner, *extra_gamers):
|
||||
room = Room.objects.create(name="SR", owner=owner)
|
||||
seat_map = {}
|
||||
gamers = [owner] + list(extra_gamers)
|
||||
roles = ["PC", "NC", "EC", "SC", "AC", "BC"]
|
||||
for i, (gamer, role) in enumerate(zip(gamers, roles), start=1):
|
||||
seat = TableSeat.objects.create(room=room, gamer=gamer, slot_number=i, role=role)
|
||||
seat_map[role] = seat
|
||||
room.table_status = Room.SIG_SELECT
|
||||
room.save()
|
||||
return room, seat_map
|
||||
|
||||
|
||||
class SigReserveViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="pc@sig.io")
|
||||
self.client.force_login(self.user)
|
||||
self.room, self.seats = _make_sig_room(self.user)
|
||||
self.card = TarotCard.objects.first()
|
||||
|
||||
def test_non_post_returns_405(self):
|
||||
response = self.client.get(reverse("epic:sig_reserve", kwargs={"room_id": self.room.id}))
|
||||
self.assertEqual(response.status_code, 405)
|
||||
|
||||
def test_reserve_action_succeeds(self):
|
||||
response = self.client.post(
|
||||
reverse("epic:sig_reserve", kwargs={"room_id": self.room.id}),
|
||||
{"action": "reserve", "card_id": str(self.card.pk)},
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertTrue(SigReservation.objects.filter(room=self.room, gamer=self.user).exists())
|
||||
|
||||
def test_release_action_removes_reservation(self):
|
||||
SigReservation.objects.create(
|
||||
room=self.room, gamer=self.user, card=self.card,
|
||||
polarity=SigReservation.LEVITY, seat=self.seats["PC"],
|
||||
)
|
||||
response = self.client.post(
|
||||
reverse("epic:sig_reserve", kwargs={"room_id": self.room.id}),
|
||||
{"action": "release"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertFalse(SigReservation.objects.filter(room=self.room, gamer=self.user).exists())
|
||||
|
||||
|
||||
class SigReadyViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="pc@ready.io")
|
||||
self.client.force_login(self.user)
|
||||
self.room, self.seats = _make_sig_room(self.user)
|
||||
self.card = (
|
||||
TarotCard.objects.filter(arcana=TarotCard.MIDDLE).first()
|
||||
or TarotCard.objects.first()
|
||||
)
|
||||
self.res = SigReservation.objects.create(
|
||||
room=self.room, gamer=self.user, card=self.card,
|
||||
polarity=SigReservation.LEVITY, seat=self.seats["PC"],
|
||||
)
|
||||
|
||||
def test_non_post_returns_405(self):
|
||||
response = self.client.get(reverse("epic:sig_ready", kwargs={"room_id": self.room.id}))
|
||||
self.assertEqual(response.status_code, 405)
|
||||
|
||||
def test_ready_action_sets_ready_flag(self):
|
||||
response = self.client.post(
|
||||
reverse("epic:sig_ready", kwargs={"room_id": self.room.id}),
|
||||
{"action": "ready"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.res.refresh_from_db()
|
||||
self.assertTrue(self.res.ready)
|
||||
|
||||
def test_ready_action_idempotent_when_already_ready(self):
|
||||
self.res.ready = True
|
||||
self.res.save()
|
||||
response = self.client.post(
|
||||
reverse("epic:sig_ready", kwargs={"room_id": self.room.id}),
|
||||
{"action": "ready"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_unready_action_saves_seconds_remaining(self):
|
||||
self.res.ready = True
|
||||
self.res.save()
|
||||
response = self.client.post(
|
||||
reverse("epic:sig_ready", kwargs={"room_id": self.room.id}),
|
||||
{"action": "unready", "seconds_remaining": "7"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.res.refresh_from_db()
|
||||
self.assertEqual(self.res.countdown_remaining, 7)
|
||||
|
||||
def test_unready_action_handles_invalid_seconds_remaining(self):
|
||||
self.res.ready = True
|
||||
self.res.save()
|
||||
response = self.client.post(
|
||||
reverse("epic:sig_ready", kwargs={"room_id": self.room.id}),
|
||||
{"action": "unready", "seconds_remaining": "abc"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.res.refresh_from_db()
|
||||
self.assertEqual(self.res.countdown_remaining, 12)
|
||||
|
||||
|
||||
class SigConfirmViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="pc@confirm.io")
|
||||
self.client.force_login(self.user)
|
||||
self.room, self.seats = _make_sig_room(self.user)
|
||||
|
||||
def test_non_post_returns_405(self):
|
||||
response = self.client.get(reverse("epic:sig_confirm", kwargs={"room_id": self.room.id}))
|
||||
self.assertEqual(response.status_code, 405)
|
||||
|
||||
|
||||
class SelectSigViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="pc@selsig.io")
|
||||
self.client.force_login(self.user)
|
||||
self.room, self.seats = _make_sig_room(self.user)
|
||||
|
||||
def test_non_post_redirects(self):
|
||||
response = self.client.get(reverse("epic:select_sig", kwargs={"room_id": self.room.id}))
|
||||
self.assertEqual(response.status_code, 302)
|
||||
|
||||
def test_nonexistent_card_returns_400(self):
|
||||
response = self.client.post(
|
||||
reverse("epic:select_sig", kwargs={"room_id": self.room.id}),
|
||||
{"card_id": "99999999"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
|
||||
# ── natus_preview (epic) ──────────────────────────────────────────────────────
|
||||
|
||||
class NatusPreviewViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="pc@natus.io")
|
||||
self.client.force_login(self.user)
|
||||
self.room, _ = _make_sig_room(self.user)
|
||||
self.room.table_status = Room.SKY_SELECT
|
||||
self.room.save()
|
||||
self.url = reverse("epic:natus_preview", kwargs={"room_id": self.room.id})
|
||||
|
||||
def test_missing_params_returns_400(self):
|
||||
response = self.client.get(self.url, {"date": "1990-06-15"})
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_non_numeric_lat_returns_400(self):
|
||||
response = self.client.get(self.url, {"date": "1990-06-15", "lat": "abc", "lon": "0"})
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_out_of_range_lat_returns_400(self):
|
||||
response = self.client.get(self.url, {"date": "1990-06-15", "lat": "999", "lon": "0"})
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_invalid_tz_string_returns_400(self):
|
||||
response = self.client.get(
|
||||
self.url,
|
||||
{"date": "1990-06-15", "lat": "51.5", "lon": "-0.1", "tz": "Not/Real"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_bad_date_format_returns_400(self):
|
||||
response = self.client.get(
|
||||
self.url,
|
||||
{"date": "baddate", "time": "09:00", "lat": "51.5", "lon": "-0.1", "tz": "UTC"},
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
@patch("apps.epic.views.http_requests")
|
||||
def test_pyswiss_failure_returns_502(self, mock_requests):
|
||||
from unittest.mock import MagicMock
|
||||
tz_r = MagicMock()
|
||||
tz_r.json.return_value = {"timezone": "UTC"}
|
||||
tz_r.raise_for_status = MagicMock()
|
||||
chart_r = MagicMock()
|
||||
chart_r.raise_for_status.side_effect = Exception("timeout")
|
||||
mock_requests.get.side_effect = [tz_r, chart_r]
|
||||
response = self.client.get(self.url, {"date": "1990-06-15", "lat": "51.5", "lon": "-0.1"})
|
||||
self.assertEqual(response.status_code, 502)
|
||||
|
||||
@patch("apps.epic.views.http_requests")
|
||||
def test_success_returns_chart_distinctions_timezone(self, mock_requests):
|
||||
from unittest.mock import MagicMock
|
||||
payload = {
|
||||
"planets": {"Sun": {"degree": 84.5}},
|
||||
"houses": {"cusps": [0] * 12},
|
||||
"elements": {"Earth": 1},
|
||||
"house_system": "O",
|
||||
}
|
||||
tz_r = MagicMock()
|
||||
tz_r.json.return_value = {"timezone": "Europe/London"}
|
||||
tz_r.raise_for_status = MagicMock()
|
||||
ch_r = MagicMock()
|
||||
ch_r.json.return_value = payload
|
||||
ch_r.raise_for_status = MagicMock()
|
||||
mock_requests.get.side_effect = [tz_r, ch_r]
|
||||
response = self.client.get(self.url, {"date": "1990-06-15", "lat": "51.5", "lon": "-0.1"})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertIn("distinctions", data)
|
||||
self.assertIn("Stone", data["elements"])
|
||||
self.assertNotIn("Earth", data["elements"])
|
||||
|
||||
|
||||
# ── natus_save (epic) ─────────────────────────────────────────────────────────
|
||||
|
||||
class NatusSaveViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = User.objects.create(email="pc@natussave.io")
|
||||
self.client.force_login(self.user)
|
||||
self.room, _ = _make_sig_room(self.user)
|
||||
self.room.table_status = Room.SKY_SELECT
|
||||
self.room.save()
|
||||
self.url = reverse("epic:natus_save", kwargs={"room_id": self.room.id})
|
||||
|
||||
def _post(self, payload):
|
||||
import json as _json
|
||||
return self.client.post(self.url, data=_json.dumps(payload), content_type="application/json")
|
||||
|
||||
def test_get_returns_405(self):
|
||||
response = self.client.get(self.url)
|
||||
self.assertEqual(response.status_code, 405)
|
||||
|
||||
def test_invalid_json_returns_400(self):
|
||||
response = self.client.post(self.url, data="not json", content_type="application/json")
|
||||
self.assertEqual(response.status_code, 400)
|
||||
|
||||
def test_save_draft_returns_id_and_not_confirmed(self):
|
||||
response = self._post({
|
||||
"birth_dt": "1990-06-15T09:00:00Z",
|
||||
"birth_lat": 51.5,
|
||||
"birth_lon": -0.1,
|
||||
"birth_place": "London",
|
||||
"house_system": "O",
|
||||
"chart_data": {},
|
||||
})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertIn("id", data)
|
||||
self.assertFalse(data["confirmed"])
|
||||
|
||||
def test_confirm_action_locks_character(self):
|
||||
response = self._post({
|
||||
"birth_dt": "1990-06-15T09:00:00Z",
|
||||
"birth_lat": 51.5,
|
||||
"birth_lon": -0.1,
|
||||
"birth_place": "",
|
||||
"house_system": "O",
|
||||
"chart_data": {},
|
||||
"action": "confirm",
|
||||
})
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertTrue(response.json()["confirmed"])
|
||||
|
||||
204
src/apps/epic/tests/unit/test_tasks.py
Normal file
204
src/apps/epic/tests/unit/test_tasks.py
Normal file
@@ -0,0 +1,204 @@
|
||||
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"})
|
||||
Reference in New Issue
Block a user