# 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)