from contextlib import ExitStack from unittest import mock from django.core.cache import cache from django.test import TestCase from rq.job import JobStatus from accounts.models import CustomUser from events.models import Event from games.models import EventGame, Game from . import tasks from .models import ImageCrop, Scope, Submission, SubmissionFile from .services.aandt_client.models import ( AandtApiInfosResponse200, AandtApiMatchesResponse200, AandtApiMembersResponse200, ) from .services.aandt_sync import ScopeSyncError, ScopeSyncResult, sync_event_game_scopes class SyncScopesApiTests(TestCase): def setUp(self): self.event = Event.objects.create(name="Event", path="event") self.game = Game.objects.create(name="Game", path="game") self.event_game = EventGame.objects.create(event=self.event, game=self.game) self.url = ( f"/api/events/{self.event.path}/submissions/{self.game.path}/sync-scopes" ) self.superuser = CustomUser.objects.create_user( username="admin", password="x", is_staff=True, is_superuser=True ) patcher = mock.patch("submissions.api.schedule_scope_sync", return_value=True) self.schedule = patcher.start() self.addCleanup(patcher.stop) def _post(self, body=None): return self.client.post(self.url, body or {}, content_type="application/json") def test_staff_is_forbidden(self): staff = CustomUser.objects.create_user( username="staff", password="x", is_staff=True ) self.client.force_login(staff) self.assertEqual(self._post().status_code, 403) self.schedule.assert_not_called() def test_asks_for_token_when_missing(self): self.client.force_login(self.superuser) response = self._post() self.assertEqual(response.status_code, 400) self.assertTrue(response.json()["token_required"]) self.schedule.assert_not_called() def test_stores_token_and_starts_sync(self): self.client.force_login(self.superuser) response = self._post({"apibot_token": " abc "}) self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), {"started": True}) self.event_game.refresh_from_db() self.assertEqual(self.event_game.apibot_token, "abc") self.schedule.assert_called_once_with(self.event_game.id) def test_status(self): Scope.objects.create(event_game=self.event_game, name="M1") Scope.objects.create(event_game=self.event_game, name="M2", is_active=False) self.client.force_login(self.superuser) with mock.patch( "submissions.api.get_scope_sync_state", return_value=tasks.ScopeSyncState(running=True, aandt_status="7inprog"), ): response = self.client.get(self.url) self.assertEqual(response.status_code, 200) data = response.json() self.assertTrue(data["running"]) self.assertFalse(data["has_token"]) self.assertEqual(data["aandt_status"], "7inprog") self.assertEqual((data["active_scope_count"], data["scope_count"]), (1, 2)) def test_stop(self): self.client.force_login(self.superuser) with mock.patch("submissions.api.stop_scope_sync", return_value=True) as stop: response = self.client.delete(self.url) self.assertEqual(response.json(), {"stopped": True}) stop.assert_called_once_with(self.event_game.id) def test_status_and_stop_forbidden_for_staff(self): staff = CustomUser.objects.create_user( username="staff", password="x", is_staff=True ) self.client.force_login(staff) self.assertEqual(self.client.get(self.url).status_code, 403) self.assertEqual(self.client.delete(self.url).status_code, 403) def test_custom_path_game_is_404(self): self.game.custom_path = True self.game.save() self.client.force_login(self.superuser) self.assertEqual(self._post({"apibot_token": "abc"}).status_code, 404) class SyncEventGameScopesTaskTests(TestCase): def setUp(self): event = Event.objects.create(name="Event", path="event") game = Game.objects.create(name="Game", path="game") self.event_game = EventGame.objects.create( event=event, game=game, apibot_token="abc" ) self.guard_key = tasks._scope_sync_guard_key(self.event_game.id) self.loop_id = "loop" self.job_key = tasks._scope_sync_job_key(self.event_game.id) self.state_key = tasks._scope_sync_state_key(self.event_game.id) cache.set(self.guard_key, self.loop_id) self.addCleanup( cache.delete_many, [self.guard_key, self.job_key, self.state_key] ) @staticmethod def _task_mock(task): # enqueue() returns a TaskResult, whose id ends up in the cache task.enqueue.return_value.id = "job-1" task.using.return_value.enqueue.return_value.id = "job-2" return task def _run(self, status=None, exc=None, during_sync=None): run = tasks.sync_event_game_scopes_task.func def sync_side_effect(_event_game): if during_sync: during_sync() if exc: raise exc return ScopeSyncResult(status=status) with ( mock.patch.object( tasks, "sync_event_game_scopes", side_effect=sync_side_effect ) as sync, mock.patch.object(tasks, "sync_event_game_scopes_task") as task, mock.patch.object(tasks, "logger"), ): self._task_mock(task) run(self.event_game.id, loop_id=self.loop_id) return sync, task def test_schedule_starts_only_one_loop(self): cache.delete(self.guard_key) with mock.patch.object(tasks, "sync_event_game_scopes_task") as task: self._task_mock(task) self.assertTrue(tasks.schedule_scope_sync(self.event_game.id)) self.assertFalse(tasks.schedule_scope_sync(self.event_game.id)) task.enqueue.assert_called_once() def test_reschedules_before_and_during_aandt(self): for status in ["3insc", "7inprog"]: with self.subTest(status=status): _, task = self._run(status=status) task.using.assert_called_once() def test_reschedules_after_api_failure(self): _, task = self._run(exc=ScopeSyncError("down")) task.using.assert_called_once() def test_reschedules_after_unexpected_error(self): _, task = self._run(exc=RuntimeError("boom")) task.using.assert_called_once() state = tasks.get_scope_sync_state(self.event_game.id) self.assertEqual(state.error, "boom") self.assertTrue(state.running) def test_records_job_and_state(self): self._run(status="7inprog") self.assertEqual(cache.get(self.job_key), "job-2") state = tasks.get_scope_sync_state(self.event_game.id) self.assertEqual(state.aandt_status, "7inprog") self.assertIsNotNone(state.next_run_at) def test_stop_cancels_scheduled_job(self): cache.set(self.job_key, "job-1") job = mock.Mock(get_status=mock.Mock(return_value=JobStatus.SCHEDULED)) with ( mock.patch.object(tasks.Job, "fetch", return_value=job) as fetch, mock.patch.object(tasks.django_rq, "get_connection"), ): self.assertTrue(tasks.stop_scope_sync(self.event_game.id)) fetch.assert_called_once_with("job-1", connection=mock.ANY) job.cancel.assert_called_once() self.assertIsNone(cache.get(self.guard_key)) self.assertFalse(tasks.stop_scope_sync(self.event_game.id)) def test_stop_during_run_does_not_reschedule(self): _, task = self._run( status="7inprog", during_sync=lambda: cache.delete(self.guard_key) ) task.using.assert_not_called() self.assertIsNone(cache.get(self.guard_key)) def test_stops_once_aandt_is_done(self): _, task = self._run(status="9done") task.using.assert_not_called() self.assertIsNone(cache.get(self.guard_key)) def test_stale_loop_stops(self): cache.set(self.guard_key, "another-loop") sync, task = self._run(status="7inprog") sync.assert_not_called() task.using.assert_not_called() class SyncScopeBroadcastTests(TestCase): def setUp(self): event = Event.objects.create(name="Event", path="event") game = Game.objects.create(name="Game", path="game") self.event_game = EventGame.objects.create( event=event, game=game, apibot_token="abc" ) self.player = CustomUser.objects.create_user( username="player", password="x", cas_user_id="42" ) def _sync(self, matches): api = "submissions.services.aandt_sync" responses = { "aandt_api_infos": AandtApiInfosResponse200.from_dict( {"aandt": {"status": "7inprog"}} ), "aandt_api_members": AandtApiMembersResponse200.from_dict( {"members": [{"id": 1, "user": {"id": 42}}]} ), "aandt_api_matches": AandtApiMatchesResponse200.from_dict( {"matches": matches} ), } with ExitStack() as stack: for module, parsed in responses.items(): stack.enter_context( mock.patch( f"{api}.{module}.sync_detailed", return_value=mock.Mock(parsed=parsed), ) ) stack.enter_context( mock.patch(f"{api}._push_submission_link", return_value=[]) ) stack.enter_context(mock.patch("submissions.signals.get_channel_layer")) send = stack.enter_context(mock.patch("submissions.signals.async_to_sync")) with self.captureOnCommitCallbacks(execute=True): sync_event_game_scopes(self.event_game) return [c.args[1]["scope"] for c in send.return_value.call_args_list] @staticmethod def _match(match_id, status): return { "id": match_id, "title": f"M{match_id}", "status": status, "players": {"1": 0}, } def test_match_done_and_next_one_playing_notifies_player(self): self._sync([self._match(1, "playing"), self._match(2, "waitPlayer")]) sent = self._sync([self._match(1, "done"), self._match(2, "playing")]) by_external_id = { Scope.objects.get(id=scope["id"]).external_id: scope for scope in sent } self.assertFalse(by_external_id[1]["is_active"]) self.assertTrue(by_external_id[2]["is_active"]) self.assertTrue(by_external_id[2]["assigned"]) def test_unchanged_scopes_send_nothing(self): self._sync([self._match(1, "playing")]) self.assertEqual(self._sync([self._match(1, "playing")]), []) def test_removed_match_is_closed_and_notified(self): self._sync([self._match(1, "playing")]) sent = self._sync([]) self.assertEqual(len(sent), 1) self.assertFalse(sent[0]["is_active"]) class ScopeAdminBroadcastTests(TestCase): def setUp(self): event = Event.objects.create(name="Event", path="event") game = Game.objects.create(name="Game", path="game") event_game = EventGame.objects.create(event=event, game=game) self.player = CustomUser.objects.create_user(username="player", password="x") self.scope = Scope.objects.create(event_game=event_game, name="Match 1") self.scope.assigned_users.set([self.player]) self.client.force_login( CustomUser.objects.create_user( username="admin", password="x", is_staff=True, is_superuser=True ) ) self.form = { "event_game": event_game.id, "name": "Match 1", "assigned_users": [self.player.id], "is_active": "on", } def _sent(self, request): with ( mock.patch("submissions.signals.get_channel_layer"), mock.patch("submissions.signals.async_to_sync") as send, self.captureOnCommitCallbacks(execute=True), ): response = request() self.assertEqual(response.status_code, 302) return [c.args[1]["scope"] for c in send.return_value.call_args_list] def _change(self, **changes): url = f"/admin/submissions/scope/{self.scope.id}/change/" data = {k: v for k, v in {**self.form, **changes}.items() if v is not None} return self._sent(lambda: self.client.post(url, data)) def test_closing_scope_notifies_player(self): sent = self._change(is_active=None) self.assertEqual( sent, [{"id": str(self.scope.id), "is_active": False, "assigned": True}] ) def test_unassigning_player_notifies_them(self): sent = self._change(assigned_users=[]) self.assertEqual( sent, [{"id": str(self.scope.id), "is_active": True, "assigned": False}] ) def test_unchanged_save_sends_nothing(self): self.assertEqual(self._change(name="Renamed"), []) def test_delete_notifies_player(self): url = f"/admin/submissions/scope/{self.scope.id}/delete/" sent = self._sent(lambda: self.client.post(url, {"post": "yes"})) self.assertEqual( sent, [{"id": str(self.scope.id), "is_active": False, "assigned": False}] ) class SubmissionHomeViewTests(TestCase): def setUp(self): self.event = Event.objects.create(name="Event", path="event") self.client.force_login( CustomUser.objects.create_user(username="player", password="x") ) def test_any_game_path_serves_the_app(self): # Unknown game / no scope: the frontend shows its own not-found state response = self.client.get(f"/{self.event.path}/submissions/unknown-game") self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "app.html") def test_unknown_event_is_404(self): response = self.client.get("/nope/submissions/unknown-game") self.assertEqual(response.status_code, 404) class ListUserSubmissionsApiTests(TestCase): def setUp(self): self.event = Event.objects.create(name="Event", path="event") game = Game.objects.create(name="Game", path="game") event_game = EventGame.objects.create(event=self.event, game=game) self.scope = Scope.objects.create(event_game=event_game, name="M1") other_scope = Scope.objects.create(event_game=event_game, name="M2") self.player = CustomUser.objects.create_user(username="player", password="x") other = CustomUser.objects.create_user(username="other", password="x") self.mine = Submission.objects.create( event=self.event, scope=self.scope, user=self.player ) self.theirs = Submission.objects.create( event=self.event, scope=self.scope, user=other ) Submission.objects.create(event=self.event, scope=other_scope, user=other) self.url = f"/api/events/{self.event.path}/submissions/submission/by/user" def _list(self): response = self.client.get(self.url, {"scope_id": self.scope.id}) self.assertEqual(response.status_code, 200) return response.json() def test_player_sees_only_own_submissions(self): self.client.force_login(self.player) data = self._list() self.assertEqual([s["id"] for s in data], [str(self.mine.id)]) self.assertEqual(data[0]["username"], "player") def test_staff_sees_all_submissions_of_scope(self): staff = CustomUser.objects.create_user( username="staff", password="x", is_staff=True ) self.client.force_login(staff) data = self._list() self.assertEqual( {s["id"] for s in data}, {str(self.mine.id), str(self.theirs.id)} ) class SubmissionBroadcastTests(TestCase): def setUp(self): event = Event.objects.create(name="Event", path="event") game = Game.objects.create(name="Game", path="game") event_game = EventGame.objects.create(event=event, game=game) scope = Scope.objects.create(event_game=event_game, name="M1") self.player = CustomUser.objects.create_user(username="player", password="x") submission = Submission.objects.create( event=event, scope=scope, user=self.player ) # bulk_create: skips the file broadcast, which would sign an S3 url [submission_file] = SubmissionFile.objects.bulk_create( [ SubmissionFile( submission=submission, file="shot.png", original_filename="shot.png", file_size=1, content_type="image/png", ) ] ) self.crop = ImageCrop.objects.create( submission_file=submission_file, x1=0, y1=0, x2=1, y2=1 ) def test_crop_update_reaches_owner_and_staff(self): with mock.patch("submissions.signals.get_channel_layer") as get_layer: get_layer.return_value.group_send = mock.AsyncMock() self.crop.save() groups = [c.args[0] for c in get_layer.return_value.group_send.call_args_list] self.assertEqual( groups, ["submissions_staff", f"submissions_user_{self.player.id}"] ) def test_crop_delete_reaches_owner_and_staff(self): crop_id = str(self.crop.id) with mock.patch("submissions.signals.get_channel_layer") as get_layer: get_layer.return_value.group_send = mock.AsyncMock() self.crop.delete() calls = get_layer.return_value.group_send.call_args_list self.assertEqual( [c.args[0] for c in calls], ["submissions_staff", f"submissions_user_{self.player.id}"], ) self.assertEqual(calls[0].args[1]["type"], "submission.crop.deleted") self.assertEqual(calls[0].args[1]["crop"]["id"], crop_id)