from datetime import timedelta from unittest import mock from django.core.cache import cache from django.test import TestCase, override_settings from django.utils import timezone from accounts.models import CustomUser from events.models import Event from games.models import EventGame, Game from .models import PixelPlacement, PlaceSettings from .tasks import schedule_wipe, wipe_canvas_task from .wire import to_epoch_ms URL = "/api/events/event/place/stats" CANVAS_URL = "/api/events/event/place/canvas" LOCMEM_CACHE = {"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}} @mock.patch("place.stats.read_canvas") class PlaceStatsTests(TestCase): def setUp(self): self.event = Event.objects.create(name="Event", path="event") place = Game.objects.create(name="Place", path="place", custom_path=True) EventGame.objects.create(event=self.event, game=place) PlaceSettings.objects.create(event=self.event, width=4, height=2) self.alice = CustomUser.objects.create_user(username="alice", password="x") self.bob = CustomUser.objects.create_user(username="bob", password="x") self.admin = CustomUser.objects.create_user( username="admin", password="x", is_staff=True ) def place(self, player, x, y, color_index): PixelPlacement.objects.create( event=self.event, player=player, x=x, y=y, color_index=color_index ) def test_requires_staff(self, _): self.client.force_login(self.alice) response = self.client.get(URL) self.assertEqual(response.status_code, 401) def test_empty_game(self, read_canvas): read_canvas.return_value = b"" self.client.force_login(self.admin) data = self.client.get(URL).json() self.assertEqual(data["total_placements"], 0) self.assertEqual(data["total_players"], 0) self.assertEqual(data["coverage"], 0.0) self.assertEqual(data["avg_placements_per_player"], 0.0) self.assertEqual(data["colors"][0]["on_canvas"], 8) self.assertEqual(data["per_hour"], []) def test_stats(self, read_canvas): self.place(self.alice, 0, 0, 5) self.place(self.bob, 0, 0, 3) self.place(self.alice, 0, 0, 5) self.place(self.alice, 1, 0, 3) # Outside the current grid, e.g. from before a shrink self.place(self.bob, 9, 9, 3) # Short canvas, Redis only padded up to the last written pixel read_canvas.return_value = bytes([5, 3]) self.client.force_login(self.admin) data = self.client.get(URL).json() self.assertEqual(data["total_placements"], 5) self.assertEqual(data["total_players"], 2) self.assertEqual(data["pixels_touched"], 2) self.assertEqual(data["coverage"], 0.25) self.assertEqual(data["overwrites"], 2) self.assertEqual(data["avg_placements_per_player"], 2.5) self.assertEqual( data["top_players"], [ {"username": "alice", "placements": 3}, {"username": "bob", "placements": 2}, ], ) self.assertEqual( data["most_contested_pixels"][0], {"x": 0, "y": 0, "placements": 3, "players": 2}, ) colors = {row["color_index"]: row for row in data["colors"]} self.assertEqual(colors[5]["placements"], 2) self.assertEqual(colors[3]["placements"], 3) self.assertEqual(colors[5]["on_canvas"], 1) self.assertEqual(colors[0]["on_canvas"], 6) self.assertEqual(sum(row["placements"] for row in data["per_hour"]), 5) @override_settings(CACHES=LOCMEM_CACHE) @mock.patch("place.signals.rebuild_canvas_from_history") class PlacementEndTests(TestCase): def setUp(self): cache.clear() self.event = Event.objects.create(name="Event", path="event") place = Game.objects.create(name="Place", path="place", custom_path=True) EventGame.objects.create(event=self.event, game=place) self.alice = CustomUser.objects.create_user(username="alice", password="x") def settings(self, ends_at): with mock.patch("place.signals.schedule_wipe"): return PlaceSettings.objects.create( event=self.event, width=2, height=1, placement_ends_at=ends_at ) @mock.patch("place.api.read_canvas") def test_canvas_is_blank_once_ended(self, read_canvas, _): read_canvas.return_value = bytes([5, 3]) place_settings = self.settings(timezone.now() + timedelta(hours=1)) self.client.force_login(self.alice) self.assertEqual(self.client.get(CANVAS_URL).content[4:], bytes([5, 3])) place_settings.placement_ends_at = timezone.now() - timedelta(seconds=1) with mock.patch("place.signals.schedule_wipe"): place_settings.save() self.assertEqual(self.client.get(CANVAS_URL).content[4:], bytes(2)) @mock.patch("place.tasks.wipe_canvas_task") def test_wipe_is_scheduled_once_per_end_time(self, task, _): ends_at = timezone.now() + timedelta(hours=1) place_settings = self.settings(ends_at) self.assertTrue(schedule_wipe(place_settings)) self.assertFalse(schedule_wipe(place_settings)) task.using.assert_called_once_with(run_after=ends_at) task.using.return_value.enqueue.assert_called_once_with( place_settings.id, to_epoch_ms(ends_at) ) @mock.patch("place.tasks.wipe_canvas_task") def test_no_wipe_without_future_end(self, task, _): self.assertFalse(schedule_wipe(self.settings(None))) past = PlaceSettings(placement_ends_at=timezone.now() - timedelta(seconds=1)) self.assertFalse(schedule_wipe(past)) task.using.assert_not_called() @mock.patch("place.tasks.get_channel_layer") def test_task_broadcasts_wipe(self, get_channel_layer, _): ends_at = timezone.now() place_settings = self.settings(ends_at) layer = get_channel_layer.return_value layer.group_send = mock.AsyncMock() wipe_canvas_task.call(place_settings.id, to_epoch_ms(ends_at)) layer.group_send.assert_awaited_once_with("place_room_event", {"type": "wipe"}) @mock.patch("place.tasks.get_channel_layer") def test_task_skips_moved_end_time(self, get_channel_layer, _): ends_at = timezone.now() place_settings = self.settings(ends_at + timedelta(hours=1)) layer = get_channel_layer.return_value layer.group_send = mock.AsyncMock() wipe_canvas_task.call(place_settings.id, to_epoch_ms(ends_at)) layer.group_send.assert_not_awaited()