<commit_before>import asyncio
from collections import defaultdict
import datetime
from unittest.mock import Mock
from unittest.mock import patch

from ctf_gameserver.lib.database import transaction_cursor
from ctf_gameserver.lib.flag import generate as generate_flag
from ctf_gameserver.lib.test_util import DatabaseTestCase
from ctf_gameserver.submission.submission import serve


class ServerTest(DatabaseTestCase):

    fixtures = ['tests/submission/fixtures/server.json']
    flag_prefix = 'FAUST_'
    flag_secret = b'topsecret'
    metrics = defaultdict(Mock)

    async def connect(self):
        task = asyncio.create_task(serve('localhost', 6666, self.connection, {
            'flag_secret': self.flag_secret,
            'team_regex': None,
            'competition_name': 'Test CTF',
            'flag_prefix': self.flag_prefix,
            'metrics': self.metrics
        }))

        for _ in range(50):
            try:
                reader, writer = await asyncio.open_connection('localhost', 6666)
                break
            except OSError:
                await asyncio.sleep(0.1)

        return (task, reader, writer)

    @patch('ctf_gameserver.submission.submission._match_net_number')
    def test_basic(self, net_number_mock):
        async def coroutine():
            net_number_mock.return_value = 103

            with transaction_cursor(self.connection) as cursor:
                cursor.execute('UPDATE scoring_gamecontrol SET start = datetime("now"), '
                               '                               end = datetime("now", "+1 hour")')

            with transaction_cursor(self.connection) as cursor:
                cursor.execute('SELECT COUNT(*) FROM scoring_capture')
                capture_count = cursor.fetchone()[0]
            self.assertEqual(capture_count, 0)

            task, reader, writer = await self.connect()
            await reader.readuntil(b'\n\n')

            expiration_time = datetime.datetime.now() + datetime.timedelta(seconds=60)
            flag = generate_flag(expiration_time, 4, 102, self.flag_secret, self.flag_prefix).encode('ascii')
            writer.write(flag + b'\n')
<commit_msg>Fix submission server in GitHub Actions<commit_after>import asyncio
from collections import defaultdict
import datetime
from unittest.mock import Mock
from unittest.mock import patch

from ctf_gameserver.lib.database import transaction_cursor
from ctf_gameserver.lib.flag import generate as generate_flag
from ctf_gameserver.lib.test_util import DatabaseTestCase
from ctf_gameserver.submission.submission import serve


class ServerTest(DatabaseTestCase):

    fixtures = ['tests/submission/fixtures/server.json']
    flag_prefix = 'FAUST_'
    flag_secret = b'topsecret'
    metrics = defaultdict(Mock)

    async def connect(self):
        # For this to work on GitHub Actions (in Docker), we need to use the v4 address instead of
        # "localhost"
        task = asyncio.create_task(serve('127.0.0.1', 6666, self.connection, {
            'flag_secret': self.flag_secret,
            'team_regex': None,
            'competition_name': 'Test CTF',
            'flag_prefix': self.flag_prefix,
            'metrics': self.metrics
        }))

        for _ in range(50):
            try:
                reader, writer = await asyncio.open_connection('127.0.0.1', 6666)
                break
            except OSError:
                await asyncio.sleep(0.1)

        return (task, reader, writer)

    @patch('ctf_gameserver.submission.submission._match_net_number')
    def test_basic(self, net_number_mock):
        async def coroutine():
            net_number_mock.return_value = 103

            with transaction_cursor(self.connection) as cursor:
                cursor.execute('UPDATE scoring_gamecontrol SET start = datetime("now"), '
                               '                               end = datetime("now", "+1 hour")')

            with transaction_cursor(self.connection) as cursor:
                cursor.execute('SELECT COUNT(*) FROM scoring_capture')
                capture_count = cursor.fetchone()[0]
            self.assertEqual(capture_count, 0)

            task, reader, writer = await self.connect()
            await reader.readuntil(b'\n\n')

            expiration_time = datetime.datetime.now() + datetime.timedelta(seconds=60)
            flag = generate_flag(expiration_time, 4, 102, self.flag_secret, self.flag_prefix).encode('ascii')
            writer.write(flag + b'\n')

{'fraction_non_alphanumeric': 0.0939791157520551, 'fraction_numerical': 0.01355254387913797, 'mean_word_length': 2.465742879137798, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6696389', 'n_tokens_mistral': 1291, 'n_tokens_neox': 1199, 'n_words': 288}