mirror of
https://gitlab.com/animath/si/plateforme.git
synced 2024-11-26 20:47:09 +00:00
815 lines
41 KiB
Python
815 lines
41 KiB
Python
# Copyright (C) 2023 by Animath
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
import asyncio
|
|
from random import shuffle
|
|
|
|
from asgiref.sync import sync_to_async
|
|
from channels.auth import AuthMiddlewareStack
|
|
from channels.routing import URLRouter
|
|
from channels.testing import WebsocketCommunicator
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import User
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.contrib.sites.models import Site
|
|
from django.test import TestCase
|
|
from django.urls import reverse
|
|
from participation.models import Team, Tournament
|
|
|
|
from . import routing
|
|
from .models import Draw, Pool, Round, TeamDraw
|
|
|
|
|
|
class TestDraw(TestCase):
|
|
def setUp(self):
|
|
self.superuser = User.objects.create_superuser(
|
|
username="admin",
|
|
email="admin@example.com",
|
|
password="toto1234",
|
|
)
|
|
|
|
self.tournament = Tournament.objects.create(
|
|
name="Test",
|
|
)
|
|
self.teams = []
|
|
for i in range(12):
|
|
t = Team.objects.create(
|
|
name=f"Team {i + 1}",
|
|
trigram=3 * chr(65 + i),
|
|
)
|
|
t.participation.tournament = self.tournament
|
|
t.participation.valid = True
|
|
t.participation.save()
|
|
self.teams.append(t)
|
|
shuffle(self.teams)
|
|
|
|
async def test_draw(self): # noqa: C901
|
|
"""
|
|
Simulate a full draw operation.
|
|
"""
|
|
await sync_to_async(self.async_client.force_login)(self.superuser)
|
|
|
|
tid = self.tournament.id
|
|
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Connect to Websocket
|
|
headers = [(b'cookie', self.async_client.cookies.output(header='', sep='; ').encode())]
|
|
communicator = WebsocketCommunicator(AuthMiddlewareStack(URLRouter(routing.websocket_urlpatterns)),
|
|
"/ws/draw/", headers)
|
|
connected, subprotocol = await communicator.connect()
|
|
self.assertTrue(connected)
|
|
|
|
# Define language
|
|
await communicator.send_json_to({'tid': tid, 'type': 'set_language', 'language': 'en'})
|
|
|
|
# Ensure that Draw has not started
|
|
self.assertFalse(await Draw.objects.filter(tournament=self.tournament).aexists())
|
|
|
|
# Must be an error since 1+1+1 != 12
|
|
await communicator.send_json_to({'tid': tid, 'type': 'start_draw', 'fmt': '1+1+1'})
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'alert')
|
|
self.assertEqual(resp['alert_type'], 'danger')
|
|
self.assertEqual(resp['message'], "The sum must be equal to the number of teams: expected 12, got 3")
|
|
self.assertFalse(await Draw.objects.filter(tournament=self.tournament).aexists())
|
|
|
|
# Now start the draw
|
|
await communicator.send_json_to({'tid': tid, 'type': 'start_draw', 'fmt': '4+5+3'})
|
|
|
|
# Receive data after the start
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'alert')
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_poules', 'round': 1,
|
|
'poules': [{'letter': 'A', 'teams': []},
|
|
{'letter': 'B', 'teams': []},
|
|
{'letter': 'C', 'teams': []}]})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_poules', 'round': 2,
|
|
'poules': [{'letter': 'A', 'teams': []},
|
|
{'letter': 'B', 'teams': []},
|
|
{'letter': 'C', 'teams': []}]})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'alert')
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'draw_start', 'fmt': [3, 4, 5],
|
|
'trigrams': ['AAA', 'BBB', 'CCC', 'DDD', 'EEE', 'FFF',
|
|
'GGG', 'HHH', 'III', 'JJJ', 'KKK', 'LLL']})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 1, 'poule': None, 'team': None})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'notification')
|
|
|
|
# Ensure that now tournament has started
|
|
self.assertTrue(await Draw.objects.filter(tournament=self.tournament).aexists())
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Try to relaunch the draw
|
|
await communicator.send_json_to({'tid': tid, 'type': 'start_draw', 'fmt': '3+4+5'})
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'alert')
|
|
self.assertEqual(resp['alert_type'], 'danger')
|
|
self.assertEqual(resp['message'], "The draw is already started.")
|
|
|
|
draw: Draw = await Draw.objects.prefetch_related(
|
|
'current_round__current_pool__current_team__participation__team').aget(tournament=self.tournament)
|
|
r: Round = draw.current_round
|
|
|
|
for i, team in enumerate(self.teams):
|
|
# Launch a new dice
|
|
await communicator.send_json_to({'tid': tid, 'type': "dice", 'trigram': team.trigram})
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], "dice")
|
|
self.assertEqual(resp['team'], team.trigram)
|
|
self.assertGreaterEqual(resp['result'], 1)
|
|
self.assertLessEqual(resp['result'], 100)
|
|
td: TeamDraw = await r.teamdraw_set.aget(participation=team.participation)
|
|
if i != len(self.teams) - 1:
|
|
self.assertEqual(resp['result'], td.passage_dice)
|
|
|
|
# Try to relaunch the dice
|
|
await communicator.send_json_to({'tid': tid, 'type': "dice", 'trigram': team.trigram})
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'alert')
|
|
self.assertEqual(resp['message'], "You've already launched the dice.")
|
|
|
|
# Force exactly one duplicate
|
|
await td.arefresh_from_db()
|
|
td.passage_dice = 101 + i if i != 2 else 101
|
|
await td.asave()
|
|
|
|
# Manage duplicates
|
|
while dup_count := await r.teamdraw_set.filter(passage_dice__isnull=True).acount():
|
|
for i in range(dup_count):
|
|
# Dice
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'dice')
|
|
self.assertIsNone(resp['result'])
|
|
# Alert
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'alert')
|
|
|
|
for i in range(dup_count):
|
|
await communicator.send_json_to({'tid': tid, 'type': "dice", 'trigram': None})
|
|
await communicator.receive_json_from()
|
|
|
|
# Reset dices
|
|
for _i in range(12):
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'dice')
|
|
self.assertIsNone(resp['result'])
|
|
|
|
# Hide and re-display the dice
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': True})
|
|
|
|
# Set pools for the two rounds
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_poules')
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_poules')
|
|
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
# Manage the first pool
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 1, 'poule': 'A', 'team': None})
|
|
r: Round = await Round.objects.prefetch_related('current_pool__current_team__participation__team')\
|
|
.aget(number=1, draw=draw)
|
|
p = r.current_pool
|
|
self.assertEqual(p.letter, 1)
|
|
self.assertEqual(p.size, 3)
|
|
self.assertEqual(await p.teamdraw_set.acount(), 3)
|
|
self.assertEqual(p.current_team, None)
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
i = 0
|
|
async for td in p.teamdraw_set.prefetch_related('participation__team').all():
|
|
# Launch a new dice
|
|
await communicator.send_json_to({'tid': tid, 'type': "dice", 'trigram': td.participation.team.trigram})
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], "dice")
|
|
trigram = td.participation.team.trigram
|
|
self.assertEqual(resp['team'], trigram)
|
|
self.assertGreaterEqual(resp['result'], 1)
|
|
self.assertLessEqual(resp['result'], 100)
|
|
if i != p.size - 1:
|
|
await td.arefresh_from_db()
|
|
self.assertEqual(resp['result'], td.choice_dice)
|
|
|
|
# Try to relaunch the dice
|
|
await communicator.send_json_to({'tid': tid, 'type': "dice", 'trigram': trigram})
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'alert')
|
|
self.assertEqual(resp['message'], "You've already launched the dice.")
|
|
|
|
# Force exactly one duplicate
|
|
await td.arefresh_from_db()
|
|
td.passage_dice = 101 + i if i != 1 else 101
|
|
await td.asave()
|
|
i += 1
|
|
|
|
# Manage duplicates
|
|
while dup_count := await p.teamdraw_set.filter(choice_dice__isnull=True).acount():
|
|
for i in range(dup_count):
|
|
# Dice
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'dice')
|
|
self.assertIsNone(resp['result'])
|
|
# Alert
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'alert')
|
|
|
|
for i in range(dup_count):
|
|
await communicator.send_json_to({'tid': tid, 'type': "dice", 'trigram': None})
|
|
await communicator.receive_json_from()
|
|
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
# Check current pool
|
|
p: Pool = await Pool.objects.prefetch_related('current_team__participation__team').aget(round=r, letter=1)
|
|
td = p.current_team
|
|
trigram = td.participation.team.trigram
|
|
self.assertEqual(td.choose_index, 0)
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 1,
|
|
'poule': 'A', 'team': td.participation.team.trigram})
|
|
# Dice is hidden for everyone first
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': False})
|
|
# The draw box is displayed for the current team and for volunteers
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': True})
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Try to launch a dice while it is not the time
|
|
await communicator.send_json_to({'tid': tid, 'type': 'dice', 'trigram': None})
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'alert')
|
|
self.assertEqual(resp['message'], "This is not the time for this.")
|
|
|
|
# Draw a problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'draw_problem'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
purposed = td.purposed
|
|
self.assertIsNotNone(td.purposed)
|
|
self.assertIn(td.purposed, range(1, len(settings.PROBLEMS) + 1))
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Try to redraw a problem while it is not the time
|
|
await communicator.send_json_to({'tid': tid, 'type': 'draw_problem'})
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'alert')
|
|
self.assertEqual(resp['message'], "This is not the time for this.")
|
|
|
|
# Reject the first problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'reject'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'reject_problem', 'round': 1, 'team': trigram, 'rejected': [purposed]})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNone(td.purposed)
|
|
self.assertEqual(td.rejected, [purposed])
|
|
|
|
for i in range(2):
|
|
# Next team
|
|
p: Pool = await Pool.objects.prefetch_related('current_team__participation__team').aget(round=r, letter=1)
|
|
td = p.current_team
|
|
trigram = td.participation.team.trigram
|
|
self.assertEqual(td.choose_index, i + 1)
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 1, 'poule': 'A', 'team': trigram})
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Draw a problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'draw_problem'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNotNone(td.purposed)
|
|
self.assertIn(td.purposed, range(1, len(settings.PROBLEMS) + 1))
|
|
|
|
# Assume that this is the problem 1 for teams 2 et 4 and the problem 2 for teams 3 and 5
|
|
td.purposed = 1 + (i % 2)
|
|
await td.asave()
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Accept the problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'accept'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_problem', 'round': 1, 'team': trigram, 'problem': 1 + (i % 2)})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNone(td.purposed)
|
|
self.assertEqual(td.accepted, 1 + (i % 2))
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Go back to the first team of the pool
|
|
p: Pool = await Pool.objects.prefetch_related('current_team__participation__team').aget(round=r, letter=1)
|
|
td = p.current_team
|
|
trigram = td.participation.team.trigram
|
|
self.assertEqual(td.choose_index, 0)
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 1, 'poule': 'A', 'team': trigram})
|
|
|
|
# Draw and reject 100 times a problem
|
|
for _i in range(100):
|
|
# Draw a problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'draw_problem'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNotNone(td.purposed)
|
|
# Problems 1 and 2 are not available
|
|
self.assertIn(td.purposed, range(3, len(settings.PROBLEMS) + 1))
|
|
|
|
# Reject
|
|
await communicator.send_json_to({'tid': tid, 'type': 'reject'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': False})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'reject_problem')
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNone(td.purposed)
|
|
self.assertIn(purposed, td.rejected)
|
|
|
|
# Ensures that this is still the first team
|
|
p: Pool = await Pool.objects.prefetch_related('current_team__participation__team').aget(round=r, letter=1)
|
|
self.assertEqual(p.current_team, td)
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 1, 'poule': 'A', 'team': trigram})
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Ensures that there is a penalty
|
|
self.assertGreaterEqual(td.penalty, 1)
|
|
|
|
# Draw a last problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'draw_problem'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNotNone(td.purposed)
|
|
self.assertIn(td.purposed, range(1, len(settings.PROBLEMS) + 1))
|
|
|
|
# Accept the problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'accept'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_problem', 'round': 1, 'team': trigram, 'problem': td.purposed})
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNone(td.purposed)
|
|
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 1, 'poule': 'B', 'team': None})
|
|
|
|
# Start pool 2
|
|
r: Round = await Round.objects.prefetch_related('current_pool__current_team__participation__team')\
|
|
.aget(number=1, draw=draw)
|
|
p = r.current_pool
|
|
self.assertEqual(p.letter, 2)
|
|
self.assertEqual(p.size, 4)
|
|
self.assertEqual(await p.teamdraw_set.acount(), 4)
|
|
self.assertEqual(p.current_team, None)
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
i = 0
|
|
async for td in p.teamdraw_set.prefetch_related('participation__team').all():
|
|
# Launch a new dice
|
|
await communicator.send_json_to({'tid': tid, 'type': "dice", 'trigram': td.participation.team.trigram})
|
|
await communicator.receive_json_from()
|
|
await td.arefresh_from_db()
|
|
td.choice_dice = 101 + i # Avoid duplicates
|
|
await td.asave()
|
|
i += 1
|
|
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
for i in range(4):
|
|
# Next team
|
|
p: Pool = await Pool.objects.prefetch_related('current_team__participation__team').aget(round=r, letter=2)
|
|
td = p.current_team
|
|
trigram = td.participation.team.trigram
|
|
self.assertEqual(td.choose_index, i)
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 1, 'poule': 'B', 'team': trigram})
|
|
if i == 0:
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': True})
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Draw a problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'draw_problem'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNotNone(td.purposed)
|
|
self.assertIn(td.purposed, range(1, len(settings.PROBLEMS) + 1))
|
|
# Lower problems are already accepted
|
|
self.assertGreaterEqual(td.purposed, i + 1)
|
|
|
|
# Assume that this is the problem is i for the team i
|
|
td.purposed = i + 1
|
|
await td.asave()
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Accept the problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'accept'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_problem', 'round': 1, 'team': trigram, 'problem': i + 1})
|
|
if i < 3:
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': True})
|
|
else:
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNone(td.purposed)
|
|
self.assertEqual(td.accepted, i + 1)
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Start pool 3
|
|
r: Round = await Round.objects.prefetch_related('current_pool__current_team__participation__team')\
|
|
.aget(number=1, draw=draw)
|
|
p = r.current_pool
|
|
self.assertEqual(p.letter, 3)
|
|
self.assertEqual(p.size, 5)
|
|
self.assertEqual(await p.teamdraw_set.acount(), 5)
|
|
self.assertEqual(p.current_team, None)
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 1, 'poule': 'C', 'team': None})
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
i = 0
|
|
async for td in p.teamdraw_set.prefetch_related('participation__team').all():
|
|
# Launch a new dice
|
|
await communicator.send_json_to({'tid': tid, 'type': "dice", 'trigram': td.participation.team.trigram})
|
|
await communicator.receive_json_from()
|
|
await td.arefresh_from_db()
|
|
td.choice_dice = 101 + i # Avoid duplicates
|
|
await td.asave()
|
|
i += 1
|
|
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
for i in range(5):
|
|
# Next team
|
|
p: Pool = await Pool.objects.prefetch_related('current_team__participation__team').aget(round=r, letter=3)
|
|
td = p.current_team
|
|
trigram = td.participation.team.trigram
|
|
self.assertEqual(td.choose_index, i)
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 1, 'poule': 'C', 'team': trigram})
|
|
if i == 0:
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': True})
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Draw a problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'draw_problem'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNotNone(td.purposed)
|
|
self.assertIn(td.purposed, range(1, len(settings.PROBLEMS) + 1))
|
|
# Lower problems are already accepted
|
|
self.assertGreaterEqual(td.purposed, 1 + i // 2)
|
|
|
|
# Assume that this is the problem is i / 2 for the team i (there are 5 teams)
|
|
# We force to have duplicates
|
|
td.purposed = 1 + i // 2
|
|
await td.asave()
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Accept the problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'accept'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_problem', 'round': 1, 'team': trigram, 'problem': 1 + i // 2})
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNone(td.purposed)
|
|
self.assertEqual(td.accepted, 1 + i // 2)
|
|
if i == 4:
|
|
break
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Reorder the pool since there are 5 teams
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'reorder_poule')
|
|
|
|
# Start round 2
|
|
draw: Draw = await Draw.objects.prefetch_related(
|
|
'current_round__current_pool__current_team__participation__team').aget(tournament=self.tournament)
|
|
r = draw.current_round
|
|
p = r.current_pool
|
|
self.assertEqual(r.number, 2)
|
|
self.assertEqual(p.letter, 1)
|
|
|
|
for j in range(12):
|
|
# Reset dices
|
|
self.assertIsNone((await communicator.receive_json_from())['result'])
|
|
|
|
# Get pools
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'set_poules')
|
|
self.assertEqual(resp['round'], 2)
|
|
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': True})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'export_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
for i in range(3):
|
|
# Iterate on each pool
|
|
r: Round = await Round.objects.prefetch_related('current_pool__current_team__participation__team') \
|
|
.aget(draw=draw, number=2)
|
|
p = r.current_pool
|
|
self.assertEqual(p.letter, i + 1)
|
|
self.assertEqual(p.size, i + 3)
|
|
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 2, 'poule': chr(65 + i), 'team': None})
|
|
|
|
j = 0
|
|
async for td in p.teamdraw_set.prefetch_related('participation__team').all():
|
|
# Launch a new dice
|
|
await communicator.send_json_to({'tid': tid, 'type': "dice", 'trigram': td.participation.team.trigram})
|
|
await communicator.receive_json_from()
|
|
await td.arefresh_from_db()
|
|
td.choice_dice = 101 + j # Avoid duplicates
|
|
await td.asave()
|
|
j += 1
|
|
|
|
resp = await communicator.receive_json_from()
|
|
self.assertEqual(resp['type'], 'set_info')
|
|
|
|
for j in range(3 + i):
|
|
# Next team
|
|
p: Pool = await Pool.objects.prefetch_related('current_team__participation__team').aget(round=r,
|
|
letter=i + 1)
|
|
td = p.current_team
|
|
trigram = td.participation.team.trigram
|
|
self.assertEqual(td.choose_index, j)
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'set_active', 'round': 2, 'poule': chr(65 + i),
|
|
'team': trigram})
|
|
if j == 0:
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': True})
|
|
|
|
# Render page
|
|
resp = await self.async_client.get(reverse('draw:index'))
|
|
self.assertEqual(resp.status_code, 200)
|
|
|
|
# Draw a problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'draw_problem'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': False})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNotNone(td.purposed)
|
|
self.assertIn(td.purposed, range(1, len(settings.PROBLEMS) + 1))
|
|
# Check that the problem is different from the previous day
|
|
old_td = await TeamDraw.objects.aget(round__number=1, round__draw=draw,
|
|
participation_id=td.participation_id)
|
|
self.assertNotEqual(td.purposed, old_td.accepted)
|
|
|
|
# Accept the problem
|
|
await communicator.send_json_to({'tid': tid, 'type': 'accept'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'buttons_visibility', 'visible': False})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_problem')
|
|
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
|
|
self.assertIsNone(td.purposed)
|
|
if j == 2 + i:
|
|
break
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'box_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
if i == 2:
|
|
# Reorder the pool since there are 5 teams
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'reorder_poule')
|
|
if i < 2:
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'dice_visibility', 'visible': True})
|
|
else:
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'export_visibility', 'visible': True})
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_info')
|
|
|
|
self.assertEqual((await communicator.receive_json_from())['type'], 'set_active')
|
|
|
|
# Export the draw
|
|
await communicator.send_json_to({'tid': tid, 'type': 'export'})
|
|
self.assertEqual(await communicator.receive_json_from(),
|
|
{'tid': tid, 'type': 'export_visibility', 'visible': False})
|
|
|
|
# Cancel all steps and reset all
|
|
for i in range(1000):
|
|
await communicator.send_json_to({'tid': tid, 'type': 'cancel'})
|
|
|
|
# Purge receive queue
|
|
while True:
|
|
try:
|
|
await communicator.receive_json_from()
|
|
except asyncio.TimeoutError:
|
|
break
|
|
|
|
if await Draw.objects.filter(tournament_id=tid).aexists():
|
|
print((await Draw.objects.filter(tournament_id=tid).aexists()))
|
|
current_state = (await Draw.objects.filter(tournament_id=tid).prefetch_related(
|
|
'current_round__current_pool__current_team__participation__team').aget()).get_state()
|
|
raise AssertionError("Draw wasn't aborted after 1000 steps, current state: " + current_state)
|
|
|
|
# Abort while the tournament is already aborted
|
|
await communicator.send_json_to({'tid': tid, 'type': "abort"})
|
|
|
|
def test_admin_pages(self):
|
|
"""
|
|
Check that admin pages are rendering successfully.
|
|
"""
|
|
self.client.force_login(self.superuser)
|
|
|
|
draw = Draw.objects.create(tournament=self.tournament)
|
|
r1 = Round.objects.create(draw=draw, number=1)
|
|
r2 = Round.objects.create(draw=draw, number=2)
|
|
p11 = Pool.objects.create(round=r1, letter=1, size=3)
|
|
p12 = Pool.objects.create(round=r1, letter=2, size=4)
|
|
p13 = Pool.objects.create(round=r1, letter=3, size=5)
|
|
p21 = Pool.objects.create(round=r2, letter=1, size=3)
|
|
p22 = Pool.objects.create(round=r2, letter=2, size=4)
|
|
p23 = Pool.objects.create(round=r2, letter=3, size=5)
|
|
tds = []
|
|
for i, team in enumerate(self.teams):
|
|
tds.append(TeamDraw.objects.create(participation=team.participation,
|
|
round=r1,
|
|
pool=p11 if i < 3 else p12 if i < 7 else p13))
|
|
tds.append(TeamDraw.objects.create(participation=team.participation,
|
|
round=r2,
|
|
pool=p21) if i < 3 else p22 if i < 7 else p23)
|
|
|
|
p11.current_team = tds[0]
|
|
p11.save()
|
|
r1.current_pool = p11
|
|
r1.save()
|
|
draw.current_round = r1
|
|
draw.save()
|
|
|
|
response = self.client.get(reverse("admin:index") + "draw/draw/")
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
response = self.client.get(reverse("admin:index")
|
|
+ f"draw/draw/{draw.pk}/change/")
|
|
self.assertEqual(response.status_code, 200)
|
|
response = self.client.get(reverse("admin:index") +
|
|
f"r/{ContentType.objects.get_for_model(Draw).id}/"
|
|
f"{draw.pk}/")
|
|
self.assertRedirects(response, "http://" + Site.objects.get().domain +
|
|
str(draw.get_absolute_url()), 302, 200)
|
|
|
|
response = self.client.get(reverse("admin:index") + "draw/round/")
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
response = self.client.get(reverse("admin:index")
|
|
+ f"draw/round/{r1.pk}/change/")
|
|
self.assertEqual(response.status_code, 200)
|
|
response = self.client.get(reverse("admin:index")
|
|
+ f"draw/round/{r2.pk}/change/")
|
|
self.assertEqual(response.status_code, 200)
|
|
response = self.client.get(reverse("admin:index") +
|
|
f"r/{ContentType.objects.get_for_model(Round).id}/"
|
|
f"{r1.pk}/")
|
|
self.assertRedirects(response, "http://" + Site.objects.get().domain +
|
|
str(r1.get_absolute_url()), 302, 200)
|
|
|
|
response = self.client.get(reverse("admin:index") + "draw/pool/")
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
response = self.client.get(reverse("admin:index")
|
|
+ f"draw/pool/{p11.pk}/change/")
|
|
self.assertEqual(response.status_code, 200)
|
|
response = self.client.get(reverse("admin:index") +
|
|
f"r/{ContentType.objects.get_for_model(Pool).id}/"
|
|
f"{p11.pk}/")
|
|
self.assertRedirects(response, "http://" + Site.objects.get().domain +
|
|
str(p11.get_absolute_url()), 302, 200)
|
|
|
|
response = self.client.get(reverse("admin:index") + "draw/teamdraw/")
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
response = self.client.get(reverse("admin:index")
|
|
+ f"draw/teamdraw/{tds[0].pk}/change/")
|
|
self.assertEqual(response.status_code, 200)
|
|
response = self.client.get(reverse("admin:index") +
|
|
f"r/{ContentType.objects.get_for_model(TeamDraw).id}/"
|
|
f"{tds[0].pk}/")
|
|
self.assertRedirects(response, "http://" + Site.objects.get().domain +
|
|
str(tds[0].get_absolute_url()), 302, 200)
|