130 lines
4.5 KiB
Python
130 lines
4.5 KiB
Python
from unittest import mock
|
|
|
|
from django.core.cache import cache
|
|
from django.test import TestCase
|
|
|
|
from accounts.models import CustomUser
|
|
from events.models import Event
|
|
from games.models import EventGame, Game
|
|
|
|
from . import tasks
|
|
from .services.aandt_sync import ScopeSyncError, ScopeSyncResult
|
|
|
|
|
|
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_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"
|
|
cache.set(self.guard_key, self.loop_id)
|
|
self.addCleanup(cache.delete, self.guard_key)
|
|
|
|
def _run(self, status=None, exc=None):
|
|
run = tasks.sync_event_game_scopes_task.func
|
|
with (
|
|
mock.patch.object(
|
|
tasks,
|
|
"sync_event_game_scopes",
|
|
side_effect=exc,
|
|
return_value=ScopeSyncResult(status=status),
|
|
) as sync,
|
|
mock.patch.object(tasks, "sync_event_game_scopes_task") as task,
|
|
mock.patch.object(tasks, "logger"),
|
|
):
|
|
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.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_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()
|