from datetime import timedelta from io import BytesIO from unittest import mock from django.core.cache import cache from django.test import TestCase, override_settings from django.utils import timezone from PIL import Image 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" FINAL_URL = "/api/events/event/place/final.png" 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) @mock.patch("place.export.read_canvas") class FinalPngTests(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=2, height=1, # Ended, the real board is still exported placement_ends_at=timezone.now() - timedelta(hours=1), ) self.alice = CustomUser.objects.create_user(username="alice", password="x") self.admin = CustomUser.objects.create_user( username="admin", password="x", is_staff=True ) def test_requires_staff(self, _): self.client.force_login(self.alice) response = self.client.get(FINAL_URL) self.assertEqual(response.status_code, 401) def test_final_png(self, read_canvas): # Short canvas, Redis only padded up to the last written pixel read_canvas.return_value = bytes([5]) self.client.force_login(self.admin) response = self.client.get(FINAL_URL) self.assertEqual(response["Content-Type"], "image/png") image = Image.open(BytesIO(response.content)) self.assertEqual(image.size, (16, 8)) self.assertEqual(image.getpixel((0, 0)), (229, 0, 0)) self.assertEqual(image.getpixel((15, 0)), (255, 255, 255)) @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()