ynerant/plateforme-tfjm2
1
0
Fork 0
mirror of https://gitlab.com/animath/si/plateforme.git synced 2024-11-27 02:53:02 +00:00
Code Issues Releases Wiki Activity
93a2e2436d
plateforme-tfjm2/draw/tests.py
Emmy D'Anello 0c80385958
Use a unique socket for the drawing system 
Signed-off-by: Emmy D'Anello <emmy.danello@animath.fr>
2023-05-11 17:07:53 +02:00

813 lines
41 KiB
Python
Raw Blame History

# 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': '3+4+5'})
# 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': [5, 4, 3],
'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, 5)
self.assertEqual(await p.teamdraw_set.acount(), 5)
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(4):
# 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)
# Reorder the pool since there are 5 teams
self.assertEqual((await communicator.receive_json_from())['type'], 'reorder_poule')
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, 3)
self.assertEqual(await p.teamdraw_set.acount(), 3)
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(3):
# 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, 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})
td: TeamDraw = await TeamDraw.objects.prefetch_related('participation__team').aget(pk=td.pk)
self.assertIsNone(td.purposed)
self.assertEqual(td.accepted, i + 1)
if i == 2:
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)
# 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, 5 - i)
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(5 - 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 == 4 - 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 == 0:
# 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=5)
p12 = Pool.objects.create(round=r1, letter=2, size=4)
p13 = Pool.objects.create(round=r1, letter=3, size=3)
p21 = Pool.objects.create(round=r2, letter=1, size=5)
p22 = Pool.objects.create(round=r2, letter=2, size=4)
p23 = Pool.objects.create(round=r2, letter=3, size=3)
tds = []
for i, team in enumerate(self.teams):
tds.append(TeamDraw.objects.create(participation=team.participation,
round=r1,
pool=p11 if i < 5 else p12 if i < 9 else p13))
tds.append(TeamDraw.objects.create(participation=team.participation,
round=r2,
pool=p21) if i < 5 else p22 if i < 9 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)
Reference in New Issue View Git Blame Copy Permalink