How to use wait_for method in testcontainers-python

Best Python code snippet using testcontainers-python_python

cmd_test.py

Source:cmd_test.py

1import asyncio2import discord3import googleapiclient.errors4import necrobot.exception5from necrobot.util import server6from necrobot.botbase.command import Command7from necrobot.botbase.commandtype import CommandType8from necrobot.botbase.necrobot import Necrobot9from necrobot.test import msgqueue10from necrobot.config import Config11from necrobot.util import console12from necrobot.gsheet.matchupsheet import MatchupSheet13from necrobot.gsheet import sheetlib14from necrobot.league.leaguemgr import LeagueMgr15class TestCommandType(CommandType):16    def __init__(self, bot_channel, cmd_name):17        CommandType.__init__(self, bot_channel, cmd_name)18        self.admin_only = True19        self.testing_command = True20    def get_send_func(self, channel):21        async def send(author, msg, wait_for=None):22            if wait_for is not None:23                wait_ev = await self.wait_event(wait_for)24            await channel.send("`{0}` {1}".format(author.display_name, msg))25            await Necrobot().force_command(channel=channel, author=author, message_str=msg)26            if wait_for is not None:27                # noinspection PyUnboundLocalVariable28                await wait_ev.wait()29        return send30    @staticmethod31    async def wait_event(msg_str: str):32        def starts_with_str(msg: discord.Message) -> bool:33            return msg_str in msg.content34        return await msgqueue.register_event(starts_with_str)35class TestCreateCategory(TestCommandType):36    def __init__(self, bot_channel):37        TestCommandType.__init__(self, bot_channel, 'testcreatecategory')38        self.help_text = "Make the 'Race Rooms' category channel."39    async def _do_execute(self, cmd: Command):40        await server.guild.create_channel_category(41            name=Config.MATCH_CHANNEL_CATEGORY_NAME42        )43class TestOverwriteGSheet(TestCommandType):44    def __init__(self, bot_channel):45        TestCommandType.__init__(self, bot_channel, 'testoverwritegsheet')46        self.help_text = "`{} league_tag`: Write a bunch of data into league\'s GSheet."47    async def _do_execute(self, cmd: Command):48        await cmd.channel.send(49            'Not currently implemented.'50        )51        # if len(cmd.args) != 1:52        #     await cmd.channel.send(53        #         'Incorrect number of arguments for {}.'.format(self.mention)54        #     )55        #     return56        #57        # league_tag = cmd.args[0]58        # try:59        #     league = await LeagueMgr().get_league(league_tag=league_tag)60        # except necrobot.exception.LeagueDoesNotExist:61        #     await cmd.channel.send(62        #         'Error: The league `{}` was not found.'.format(league_tag)63        #     )64        #     return65        #66        # # Get the matchup sheet67        # wks_id = 068        # try:69        #     matchup_sheet = await sheetlib.get_sheet(70        #             gsheet_id=league.gsheet_id,71        #             wks_id=league.worksheet_id,72        #             sheet_type=sheetlib.SheetType.MATCHUP73        #         )  # type: MatchupSheet74        # except (googleapiclient.errors.Error, necrobot.exception.NecroException) as e:75        #     await cmd.channel.send(76        #         'Error accessing GSheet: `{0}`'.format(e)77        #     )78        #     return79        #80        # if matchup_sheet is None:81        #     await cmd.channel.send('Error: MatchupSheet is None.')82        #     return83        #84        # await matchup_sheet.overwrite_gsheet()85class TestDebugMembers(TestCommandType):86    def __init__(self, bot_channel):87        TestCommandType.__init__(self, bot_channel, 'testdebug')88        self.help_text = "For temporary debugging. Functionality depends on what the bug is."89    async def _do_execute(self, cmd: Command):90        console.info("--CHANNEL LIST:---------------------------------------")91        for channel in server.guild.channels:92            console.info(str(channel))93        console.info("--END CHANNEL LIST------------------------------------")94        console.info("--MEMBER LIST:---------------------------------------")95        for member in server.guild.members:96            console.info(str(member))97        console.info("--END MEMBER LIST------------------------------------")98class TestMatch(TestCommandType):99    def __init__(self, bot_channel):100        TestCommandType.__init__(self, bot_channel, 'testmatch')101        self.help_text = "Run a full match from test code. " \102                         "WARNING: This takes several minutes and is only for debugging."103    async def _do_execute(self, cmd: Command):104        fancy = len(cmd.args) == 1 and cmd.args[0] == 'fancy'105        send = self.get_send_func(cmd.channel)106        match = self.bot_channel.match107        racer_1 = match.racer_1.member108        racer_2 = match.racer_2.member109        admin = server.find_admin(ignore=[racer_1.name, racer_2.name])110        if racer_1 is None or racer_2 is None or admin is None:111            await cmd.channel.send(112                "Can't find one of the racers (as a Discord member) in this match."113            )114            return115        # Match info116        await send(racer_1, '.matchinfo', wait_for='')117        await send(admin, '.setmatchtype bestof 5', wait_for='This match has been set')118        await send(admin, '.setmatchtype repeat 3', wait_for='This match has been set')119        await asyncio.sleep(0)120        # Time suggestion121        await send(racer_1, '.suggest friday 8p', wait_for='This match is suggested')122        await send(racer_2, '.suggest tomorrow 12:30', wait_for='This match is suggested')123        await send(racer_1, '.confirm', wait_for='officially scheduled')124        await send(racer_2, '.unconfirm', wait_for='wishes to remove')125        await send(racer_1, '.unconfirm', wait_for='has been unscheduled')126        await asyncio.sleep(0)127        # Prematch admin commands128        await send(admin, '.f-schedule tomorrow 21:15', wait_for='This match is suggested')129        await send(admin, '.f-confirm', wait_for='Forced confirmation')130        await send(admin, '.postpone', wait_for='has been postponed')131        await send(admin, '.f-begin', wait_for='Please input')132        await asyncio.sleep(0)133        # Race 1134        await send(racer_1, '.ready', wait_for='is ready')135        await send(racer_2, '.ready', wait_for='The race will begin')136        await send(racer_1, '.unready', wait_for='is no longer ready')137        await send(racer_1, '.r', wait_for='GO!')138        await send(racer_1, '.d', wait_for='Please input')139        await asyncio.sleep(0)140        # Race 2141        await send(admin, '.reseed', wait_for='Changed seed')142        await send(racer_2, '.ready', wait_for='is ready')143        await send(racer_1, '.ready', wait_for='GO!')144        await send(admin, '.pause', wait_for='Race paused')145        await send(admin, '.unpause', wait_for='GO!')146        await send(racer_1, '.time', wait_for='The current race time')147        await send(racer_2, '.d', wait_for='has finished in')148        await send(racer_1, '.d', wait_for='Please input')149        await asyncio.sleep(0)150        # Race 3:151        if not fancy:152            await send(racer_1, '.ready', wait_for='is ready')153            await send(racer_2, '.ready', wait_for='GO!')154            await send(racer_1, '.d', wait_for='Match complete')155            await asyncio.sleep(0)156        else:157            await send(admin, '.cancelrace 1', wait_for='')158            await send(admin, '.recordrace "{0}"'.format(racer_2.display_name), wait_for='')159            await send(admin, '.changewinner 2 "{0}"'.format(racer_1.display_name), wait_for='')160            await send(admin, '.postpone', wait_for='has been postponed')161            await send(admin, '.f-begin', wait_for='Please input')162            await send(admin, '.changerules diamond u', wait_for='Changed rules')163            await send(admin, '.matchinfo', wait_for='')164            await send(admin, '.changerules cadence s', wait_for='Changed rules')165            await send(admin, '.matchinfo', wait_for='')166            await send(racer_1, '.r', wait_for='is ready')167            await send(racer_2, '.r', wait_for='GO!')168            await send(admin, '.pause', wait_for='Race paused')169            await send(admin, '.cancelrace', wait_for='Please input')170            await asyncio.sleep(0)171            # Race 4:172            await send(racer_1, '.r', wait_for='is ready')173            await send(racer_2, '.r', wait_for='GO!')174            await send(admin, '.pause', wait_for='Race paused')175            await send(admin, '.newrace', wait_for='Please input')176            await asyncio.sleep(0)177            # Race 5:178            await send(racer_1, '.r', wait_for='is ready')179            await send(racer_2, '.r', wait_for='GO!')180            await send(racer_1, '.d', wait_for='Match complete')181            await send(admin, '.matchinfo', wait_for='')182            await send(admin, '.newrace', wait_for='Please input')183            await asyncio.sleep(0)184            # Race 6:185            await send(racer_1, '.r', wait_for='is ready')186            await send(racer_2, '.r', wait_for='GO!')187            await send(racer_2, '.d', wait_for='Match complete')188            await asyncio.sleep(0)189            await send(admin, '.cancelrace 3')190class TestRace(TestCommandType):191    def __init__(self, bot_channel):192        TestCommandType.__init__(self, bot_channel, 'testrace')193        self.help_text = "Run a full race from test code. " \194                         "WARNING: This takes several minutes and is only for debugging."195    async def _do_execute(self, cmd: Command):196        send = self.get_send_func(cmd.channel)197        alice = server.find_member(discord_name='incnone_testing')198        bob = server.find_member(discord_name='condorbot_alpha')199        carol = server.find_member(discord_name='condorbot')200        admin = server.find_member(discord_name='incnone')201        if alice is None or bob is None or carol is None or admin is None:202            await cmd.channel.send(203                "Can't find one of the racers (as a Discord member) in this match."204            )205            return206        # Race 1: some common stuff207        await send(alice, '.r', wait_for='Waiting on')208        await send(bob, '.e', wait_for='2 entrants')209        await send(alice, '.ready', wait_for='is already ready')210        await send(bob, '.ready', wait_for='The race will begin')211        await send(alice, '.unready', wait_for='is no longer ready')212        await send(alice, '.r', wait_for='GO!')213        await send(carol, '.notify', wait_for='will be alerted')214        await send(alice, '.f', wait_for='has forfeit')215        await send(alice, '.unforfeit', wait_for='no longer forfeit')216        await send(bob, '.death 3-2', wait_for='has forfeit')217        await send(bob, '.c spirits are fair')218        await asyncio.sleep(2)219        await send(alice, '.d', wait_for='The race is over')220        await send(alice, '.igt 4:20.69')221        await send(bob, '.re', wait_for='Race number')222        await send(alice, '.c i did it')223        # Race 2: admin stuff224        await send(alice, '.join')225        await send(carol, '.j')226        await send(admin, '.e', wait_for='3 entrants')227        await send(alice, '.missing', wait_for='Unentered')228        await send(bob, '.r')229        await send(alice, '.r')230        await send(carol, '.r', wait_for='1 remaining')231        await send(alice, '.poke')232        await send(bob, '.poke')233        await send(admin, '.kick "{0}"'.format(alice.display_name), wait_for='no longer entered')234        await send(alice, '.r', wait_for='is ready')235        await send(admin, '.reseed', wait_for='Changed seed')236        await send(admin, '.changerules Diamond u custom have a blast with diamond', wait_for="Couldn't parse")237        await send(admin, '.changerules Diamond u custom "have a blast with diamond"', wait_for="Changed rules")238        await send(admin, '.reseed', wait_for='This is not a seeded race')239        await send(admin, '.unenter', wait_for='GO!')240        await send(admin, '.pause', wait_for='Race paused')241        await send(bob, '.time', wait_for='The current race time')242        await asyncio.sleep(1)243        await send(carol, '.time', wait_for='The current race time')244        await send(admin, '.forceforfeit "{0}"'.format(carol.display_name), wait_for='has forfeit')245        await send(alice, '.missing', wait_for='Still racing')246        await send(alice, '.d')247        await asyncio.sleep(1)248        await send(admin, '.unpause', wait_for='GO!')249        await send(carol, '.d', wait_for='has finished')250        await send(bob, '.d 5-4 i can\'t deep blues', wait_for='has forfeit')251        await send(bob, '.notify off', wait_for='not be alerted')252        await send(carol, '.undone', wait_for='continues to race')253        await send(alice, '.d', wait_for='has finished')254        await send(carol, '.d', wait_for='The race is over')255        # Race 3256        await send(alice, '.re', wait_for='Race number')257        await send(alice, '.r')258        await send(carol, '.r', wait_for='GO!')...

test_acos_command.py

Source:test_acos_command.py

...55        self.assertTrue(result['stdout'][0].startswith('Thunder Series'))56        self.assertNotEqual(result['stdout'][0], "test")57        self.assertIsNotNone(result['stdout'][1])58        self.assertIn('Storage', result['stdout'][1])59    def test_acos_command_wait_for(self):60        wait_for = 'result[0] contains "ACOS"'61        set_module_args(dict(commands=['show version'], wait_for=wait_for))62        result = self.execute_module()63        self.assertIn('ACOS', result['stdout'][0])64    def test_acos_command_wait_for_fails(self):65        wait_for = 'result[0] contains "test"'66        set_module_args(dict(commands=['show version'], wait_for=wait_for))67        self.execute_module(failed=True)68        self.assertEqual(self.run_commands.call_count, 12)69    def test_acos_command_retries(self):70        wait_for = 'result[0] contains "ACOS"'71        set_module_args(72            dict(commands=['show version'], wait_for=wait_for, retries=2))73        self.execute_module()...

test_junos_command.py

Source:test_junos_command.py

...51        set_module_args(dict(commands=['show version', 'show version'], display='text'))52        result = self.execute_module()53        self.assertEqual(len(result['stdout']), 2)54        self.assertTrue(result['stdout'][0].startswith('Hostname'))55    def test_junos_command_wait_for(self):56        wait_for = 'result[0] contains "Hostname"'57        set_module_args(dict(commands=['show version'], wait_for=wait_for, display='text'))58        self.execute_module()59    def test_junos_command_wait_for_fails(self):60        wait_for = 'result[0] contains "test string"'61        set_module_args(dict(commands=['show version'], wait_for=wait_for, display='text'))62        self.execute_module(failed=True)63        self.assertEqual(self.run_commands.call_count, 10)64    def test_junos_command_retries(self):65        wait_for = 'result[0] contains "test string"'66        set_module_args(dict(commands=['show version'], wait_for=wait_for, retries=2, display='text'))67        self.execute_module(failed=True)68        self.assertEqual(self.run_commands.call_count, 2)69    def test_junos_command_match_any(self):...

test_vyos_command.py

Source:test_vyos_command.py

...51        set_module_args(dict(commands=['show version', 'show version']))52        result = self.execute_module()53        self.assertEqual(len(result['stdout']), 2)54        self.assertTrue(result['stdout'][0].startswith('Version:      VyOS'))55    def test_vyos_command_wait_for(self):56        wait_for = 'result[0] contains "VyOS maintainers"'57        set_module_args(dict(commands=['show version'], wait_for=wait_for))58        self.execute_module()59    def test_vyos_command_wait_for_fails(self):60        wait_for = 'result[0] contains "test string"'61        set_module_args(dict(commands=['show version'], wait_for=wait_for))62        self.execute_module(failed=True)63        self.assertEqual(self.run_commands.call_count, 10)64    def test_vyos_command_retries(self):65        wait_for = 'result[0] contains "test string"'66        set_module_args(dict(commands=['show version'], wait_for=wait_for, retries=2))67        self.execute_module(failed=True)68        self.assertEqual(self.run_commands.call_count, 2)69    def test_vyos_command_match_any(self):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.