How to use wait_for method in Playwright Python

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

Source: cmd_test.py

import asyncio
import discord
import googleapiclient.errors

import necrobot.exception
from necrobot.util import server
from necrobot.botbase.command import Command
from necrobot.botbase.commandtype import CommandType
from necrobot.botbase.necrobot import Necrobot
from necrobot.test import msgqueue
from necrobot.config import Config
from necrobot.util import console

from necrobot.gsheet.matchupsheet import MatchupSheet
from necrobot.gsheet import sheetlib
from necrobot.league.leaguemgr import LeagueMgr


class TestCommandType(CommandType):
    def __init__(self, bot_channel, cmd_name):
        CommandType.__init__(self, bot_channel, cmd_name)
        self.admin_only = True
        self.testing_command = True

    def get_send_func(self, channel):
        async def send(author, msg, wait_for=None):
            if wait_for is not None:
                wait_ev = await self.wait_event(wait_for)

            await channel.send("`{0}` {1}".format(author.display_name, msg))
            await Necrobot().force_command(channel=channel, author=author, message_str=msg)

            if wait_for is not None:
                # noinspection PyUnboundLocalVariable
                await wait_ev.wait()
        return send

    @staticmethod
    async def wait_event(msg_str: str):
        def starts_with_str(msg: discord.Message) -> bool:
            return msg_str in msg.content
        return await msgqueue.register_event(starts_with_str)


class TestCreateCategory(TestCommandType):
    def __init__(self, bot_channel):
        TestCommandType.__init__(self, bot_channel, 'testcreatecategory')
        self.help_text = "Make the 'Race Rooms' category channel."

    async def _do_execute(self, cmd: Command):
        await server.guild.create_channel_category(
            name=Config.MATCH_CHANNEL_CATEGORY_NAME
        )


class TestOverwriteGSheet(TestCommandType):
    def __init__(self, bot_channel):
        TestCommandType.__init__(self, bot_channel, 'testoverwritegsheet')
        self.help_text = "`{} league_tag`: Write a bunch of data into league\'s GSheet."

    async def _do_execute(self, cmd: Command):
        await cmd.channel.send(
            'Not currently implemented.'
        )
        # if len(cmd.args) != 1:
        #     await cmd.channel.send(
        #         'Incorrect number of arguments for {}.'.format(self.mention)
        #     )
        #     return
        #
        # league_tag = cmd.args[0]
        # try:
        #     league = await LeagueMgr().get_league(league_tag=league_tag)
        # except necrobot.exception.LeagueDoesNotExist:
        #     await cmd.channel.send(
        #         'Error: The league `{}` was not found.'.format(league_tag)
        #     )
        #     return
        #
        # # Get the matchup sheet
        # wks_id = 0
        # try:
        #     matchup_sheet = await sheetlib.get_sheet(
        #             gsheet_id=league.gsheet_id,
        #             wks_id=league.worksheet_id,
        #             sheet_type=sheetlib.SheetType.MATCHUP
        #         )  # type: MatchupSheet
        # except (googleapiclient.errors.Error, necrobot.exception.NecroException) as e:
        #     await cmd.channel.send(
        #         'Error accessing GSheet: `{0}`'.format(e)
        #     )
        #     return
        #
        # if matchup_sheet is None:
        #     await cmd.channel.send('Error: MatchupSheet is None.')
        #     return
        #
        # await matchup_sheet.overwrite_gsheet()


class TestDebugMembers(TestCommandType):
    def __init__(self, bot_channel):
        TestCommandType.__init__(self, bot_channel, 'testdebug')
        self.help_text = "For temporary debugging. Functionality depends on what the bug is."

    async def _do_execute(self, cmd: Command):
        console.info("--CHANNEL LIST:---------------------------------------")
        for channel in server.guild.channels:
            console.info(str(channel))
        console.info("--END CHANNEL LIST------------------------------------")

        console.info("--MEMBER LIST:---------------------------------------")
        for member in server.guild.members:
            console.info(str(member))
        console.info("--END MEMBER LIST------------------------------------")


class TestMatch(TestCommandType):
    def __init__(self, bot_channel):
        TestCommandType.__init__(self, bot_channel, 'testmatch')
        self.help_text = "Run a full match from test code. " \
                         "WARNING: This takes several minutes and is only for debugging."

    async def _do_execute(self, cmd: Command):
        fancy = len(cmd.args) == 1 and cmd.args[0] == 'fancy'
        send = self.get_send_func(cmd.channel)

        match = self.bot_channel.match
        racer_1 = match.racer_1.member
        racer_2 = match.racer_2.member
        admin = server.find_admin(ignore=[racer_1.name, racer_2.name])
        if racer_1 is None or racer_2 is None or admin is None:
            await cmd.channel.send(
                "Can't find one of the racers (as a Discord member) in this match."
            )
            return

        # Match info
        await send(racer_1, '.matchinfo', wait_for='')
        await send(admin, '.setmatchtype bestof 5', wait_for='This match has been set')
        await send(admin, '.setmatchtype repeat 3', wait_for='This match has been set')
        await asyncio.sleep(0)

        # Time suggestion
        await send(racer_1, '.suggest friday 8p', wait_for='This match is suggested')
        await send(racer_2, '.suggest tomorrow 12:30', wait_for='This match is suggested')
        await send(racer_1, '.confirm', wait_for='officially scheduled')
        await send(racer_2, '.unconfirm', wait_for='wishes to remove')
        await send(racer_1, '.unconfirm', wait_for='has been unscheduled')
        await asyncio.sleep(0)

        # Prematch admin commands
        await send(admin, '.f-schedule tomorrow 21:15', wait_for='This match is suggested')
        await send(admin, '.f-confirm', wait_for='Forced confirmation')
        await send(admin, '.postpone', wait_for='has been postponed')
        await send(admin, '.f-begin', wait_for='Please input')
        await asyncio.sleep(0)

        # Race 1
        await send(racer_1, '.ready', wait_for='is ready')
        await send(racer_2, '.ready', wait_for='The race will begin')
        await send(racer_1, '.unready', wait_for='is no longer ready')
        await send(racer_1, '.r', wait_for='GO!')
        await send(racer_1, '.d', wait_for='Please input')
        await asyncio.sleep(0)

        # Race 2
        await send(admin, '.reseed', wait_for='Changed seed')
        await send(racer_2, '.ready', wait_for='is ready')
        await send(racer_1, '.ready', wait_for='GO!')
        await send(admin, '.pause', wait_for='Race paused')
        await send(admin, '.unpause', wait_for='GO!')
        await send(racer_1, '.time', wait_for='The current race time')
        await send(racer_2, '.d', wait_for='has finished in')
        await send(racer_1, '.d', wait_for='Please input')
        await asyncio.sleep(0)

        # Race 3:
        if not fancy:
            await send(racer_1, '.ready', wait_for='is ready')
            await send(racer_2, '.ready', wait_for='GO!')
            await send(racer_1, '.d', wait_for='Match complete')
            await asyncio.sleep(0)

        else:
            await send(admin, '.cancelrace 1', wait_for='')
            await send(admin, '.recordrace "{0}"'.format(racer_2.display_name), wait_for='')
            await send(admin, '.changewinner 2 "{0}"'.format(racer_1.display_name), wait_for='')
            await send(admin, '.postpone', wait_for='has been postponed')
            await send(admin, '.f-begin', wait_for='Please input')
            await send(admin, '.changerules diamond u', wait_for='Changed rules')
            await send(admin, '.matchinfo', wait_for='')
            await send(admin, '.changerules cadence s', wait_for='Changed rules')
            await send(admin, '.matchinfo', wait_for='')
            await send(racer_1, '.r', wait_for='is ready')
            await send(racer_2, '.r', wait_for='GO!')
            await send(admin, '.pause', wait_for='Race paused')
            await send(admin, '.cancelrace', wait_for='Please input')
            await asyncio.sleep(0)

            # Race 4:
            await send(racer_1, '.r', wait_for='is ready')
            await send(racer_2, '.r', wait_for='GO!')
            await send(admin, '.pause', wait_for='Race paused')
            await send(admin, '.newrace', wait_for='Please input')
            await asyncio.sleep(0)

            # Race 5:
            await send(racer_1, '.r', wait_for='is ready')
            await send(racer_2, '.r', wait_for='GO!')
            await send(racer_1, '.d', wait_for='Match complete')
            await send(admin, '.matchinfo', wait_for='')
            await send(admin, '.newrace', wait_for='Please input')
            await asyncio.sleep(0)

            # Race 6:
            await send(racer_1, '.r', wait_for='is ready')
            await send(racer_2, '.r', wait_for='GO!')
            await send(racer_2, '.d', wait_for='Match complete')
            await asyncio.sleep(0)

            await send(admin, '.cancelrace 3')


class TestRace(TestCommandType):
    def __init__(self, bot_channel):
        TestCommandType.__init__(self, bot_channel, 'testrace')
        self.help_text = "Run a full race from test code. " \
                         "WARNING: This takes several minutes and is only for debugging."

    async def _do_execute(self, cmd: Command):
        send = self.get_send_func(cmd.channel)

        alice = server.find_member(discord_name='incnone_testing')
        bob = server.find_member(discord_name='condorbot_alpha')
        carol = server.find_member(discord_name='condorbot')
        admin = server.find_member(discord_name='incnone')

        if alice is None or bob is None or carol is None or admin is None:
            await cmd.channel.send(
                "Can't find one of the racers (as a Discord member) in this match."
            )
            return

        # Race 1: some common stuff
        await send(alice, '.r', wait_for='Waiting on')
        await send(bob, '.e', wait_for='2 entrants')
        await send(alice, '.ready', wait_for='is already ready')
        await send(bob, '.ready', wait_for='The race will begin')
        await send(alice, '.unready', wait_for='is no longer ready')
        await send(alice, '.r', wait_for='GO!')
        await send(carol, '.notify', wait_for='will be alerted')
        await send(alice, '.f', wait_for='has forfeit')
        await send(alice, '.unforfeit', wait_for='no longer forfeit')
        await send(bob, '.death 3-2', wait_for='has forfeit')
        await send(bob, '.c spirits are fair')
        await asyncio.sleep(2)
        await send(alice, '.d', wait_for='The race is over')
        await send(alice, '.igt 4:20.69')
        await send(bob, '.re', wait_for='Race number')
        await send(alice, '.c i did it')

        # Race 2: admin stuff
        await send(alice, '.join')
        await send(carol, '.j')
        await send(admin, '.e', wait_for='3 entrants')
        await send(alice, '.missing', wait_for='Unentered')
        await send(bob, '.r')
        await send(alice, '.r')
        await send(carol, '.r', wait_for='1 remaining')
        await send(alice, '.poke')
        await send(bob, '.poke')
        await send(admin, '.kick "{0}"'.format(alice.display_name), wait_for='no longer entered')
        await send(alice, '.r', wait_for='is ready')
        await send(admin, '.reseed', wait_for='Changed seed')
        await send(admin, '.changerules Diamond u custom have a blast with diamond', wait_for="Couldn't parse")
        await send(admin, '.changerules Diamond u custom "have a blast with diamond"', wait_for="Changed rules")
        await send(admin, '.reseed', wait_for='This is not a seeded race')
        await send(admin, '.unenter', wait_for='GO!')
        await send(admin, '.pause', wait_for='Race paused')
        await send(bob, '.time', wait_for='The current race time')
        await asyncio.sleep(1)
        await send(carol, '.time', wait_for='The current race time')
        await send(admin, '.forceforfeit "{0}"'.format(carol.display_name), wait_for='has forfeit')
        await send(alice, '.missing', wait_for='Still racing')
        await send(alice, '.d')
        await asyncio.sleep(1)
        await send(admin, '.unpause', wait_for='GO!')
        await send(carol, '.d', wait_for='has finished')
        await send(bob, '.d 5-4 i can\'t deep blues', wait_for='has forfeit')
        await send(bob, '.notify off', wait_for='not be alerted')
        await send(carol, '.undone', wait_for='continues to race')
        await send(alice, '.d', wait_for='has finished')
        await send(carol, '.d', wait_for='The race is over')

        # Race 3
        await send(alice, '.re', wait_for='Race number')
        await send(alice, '.r')
        await send(carol, '.r', wait_for='GO!')
        await send(admin, '.forceforfeitall', wait_for='The race is over')

1import sst
2import sst.actions
3from time import time
4
5
6CALLS = 0
7
8
9def get_condition(result=True, wait=0, raises=False,
10                  cond_args=None, cond_kwargs=None):
11    initial = time()
12
13    def condition(*args, **kwargs):
14        global CALLS
15        CALLS += 1
16        if cond_args is not None:
17            if cond_args != args:
18                # can't raise an assertion error here!
19                raise TypeError('wrong args passed')
20        if cond_kwargs is not None:
21            if cond_kwargs != kwargs:
22                # can't raise an assertion error here!
23                raise TypeError('wrong args passed')
24        if time() > initial + wait:
25            return result
26        if raises:
27            raise AssertionError('not yet')
28        return False
29    return condition
30
31sst.actions.set_base_url('http://localhost:%s/' % sst.DEVSERVER_PORT)
32sst.actions.go_to('/')
33sst.actions.set_wait_timeout(0.1)
34
35sst.actions.wait_for(get_condition(True))
36sst.actions.fails(sst.actions.wait_for, get_condition(False))
37
38sst.actions.wait_for(get_condition(raises=True))
39sst.actions.fails(sst.actions.wait_for, get_condition(False, raises=True))
40
41sst.actions.wait_for(sst.actions.assert_url, '/')
42sst.actions.fails(sst.actions.wait_for, sst.actions.assert_url, '/thing')
43
44sst.actions.wait_for(sst.actions.assert_url, url='/')
45sst.actions.fails(sst.actions.wait_for, sst.actions.assert_url, url='/thing')
46
47CALLS = 0
48sst.actions.set_wait_timeout(0.1, 0.01)
49sst.actions.fails(sst.actions.wait_for, get_condition(wait=0.2))
50assert CALLS > 6
51
52sst.actions.fails(sst.actions.wait_for, get_condition(wait=0.2, raises=True))
53
54sst.actions.set_wait_timeout(0.5)
55sst.actions.wait_for(get_condition(wait=0.2))
56sst.actions.wait_for(get_condition(wait=0.2, raises=True))
57
58sst.actions.set_wait_timeout(0.3, 0.1)
59CALLS = 0
60sst.actions.wait_for(get_condition(wait=0.2))
61assert CALLS <= 3
62
63sst.actions.set_wait_timeout(5, 0.1)
64
65sst.actions.wait_for_and_refresh(get_condition(True))
66sst.actions.fails(sst.actions.wait_for_and_refresh, get_condition(False))
67
68
69# Test the return value when wait_for succeeds.
70def identity(value):
71    return value
72
73assert sst.actions.wait_for(identity, None) is None
74assert sst.actions.wait_for(identity, True) is True
75assert sst.actions.wait_for(identity, 'test') == 'test'
76assert sst.actions.wait_for(identity, 5) == 5
77assert sst.actions.wait_for(identity, [5]) == [5]
78assert sst.actions.wait_for(identity, {'test': 5}) == {'test': 5}
79assert sst.actions.wait_for(identity, (5)) == (5)
80

Source: waitfor.py

import sst
import sst.actions
from time import time


CALLS = 0


def get_condition(result=True, wait=0, raises=False,
                  cond_args=None, cond_kwargs=None):
    initial = time()

    def condition(*args, **kwargs):
        global CALLS
        CALLS += 1
        if cond_args is not None:
            if cond_args != args:
                # can't raise an assertion error here!
                raise TypeError('wrong args passed')
        if cond_kwargs is not None:
            if cond_kwargs != kwargs:
                # can't raise an assertion error here!
                raise TypeError('wrong args passed')
        if time() > initial + wait:
            return result
        if raises:
            raise AssertionError('not yet')
        return False
    return condition

sst.actions.set_base_url('http://localhost:%s/' % sst.DEVSERVER_PORT)
sst.actions.go_to('/')
sst.actions.set_wait_timeout(0.1)

sst.actions.wait_for(get_condition(True))
sst.actions.fails(sst.actions.wait_for, get_condition(False))

sst.actions.wait_for(get_condition(raises=True))
sst.actions.fails(sst.actions.wait_for, get_condition(False, raises=True))

sst.actions.wait_for(sst.actions.assert_url, '/')
sst.actions.fails(sst.actions.wait_for, sst.actions.assert_url, '/thing')

sst.actions.wait_for(sst.actions.assert_url, url='/')
sst.actions.fails(sst.actions.wait_for, sst.actions.assert_url, url='/thing')

CALLS = 0
sst.actions.set_wait_timeout(0.1, 0.01)
sst.actions.fails(sst.actions.wait_for, get_condition(wait=0.2))
assert CALLS > 6

sst.actions.fails(sst.actions.wait_for, get_condition(wait=0.2, raises=True))

sst.actions.set_wait_timeout(0.5)
sst.actions.wait_for(get_condition(wait=0.2))
sst.actions.wait_for(get_condition(wait=0.2, raises=True))

sst.actions.set_wait_timeout(0.3, 0.1)
CALLS = 0
sst.actions.wait_for(get_condition(wait=0.2))
assert CALLS <= 3

sst.actions.set_wait_timeout(5, 0.1)

sst.actions.wait_for_and_refresh(get_condition(True))
sst.actions.fails(sst.actions.wait_for_and_refresh, get_condition(False))


# Test the return value when wait_for succeeds.
def identity(value):
    return value

assert sst.actions.wait_for(identity, None) is None
assert sst.actions.wait_for(identity, True) is True
assert sst.actions.wait_for(identity, 'test') == 'test'
assert sst.actions.wait_for(identity, 5) == 5
assert sst.actions.wait_for(identity, [5]) == [5]
assert sst.actions.wait_for(identity, {'test': 5}) == {'test': 5}
assert sst.actions.wait_for(identity, (5)) == (5)

1# (c) 2016 Red Hat Inc.
2#
3# This file is part of Ansible
4#
5# Ansible is free software: you can redistribute it and/or modify
6# it under the terms of the GNU General Public License as published by
7# the Free Software Foundation, either version 3 of the License, or
8# (at your option) any later version.
9#
10# Ansible is distributed in the hope that it will be useful,
11# but WITHOUT ANY WARRANTY; without even the implied warranty of
12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13# GNU General Public License for more details.
14#
15# You should have received a copy of the GNU General Public License
16# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
17
18# Make coding more python3-ish
19from __future__ import (absolute_import, division, print_function)
20__metaclass__ = type
21
22import json
23
24from ansible.compat.tests.mock import patch
25from ansible.modules.network.iosxr import iosxr_command
26from .iosxr_module import TestIosxrModule, load_fixture, set_module_args
27
28
29class TestIosxrCommandModule(TestIosxrModule):
30
31    module = iosxr_command
32
33    def setUp(self):
34        self.mock_run_commands = patch('ansible.modules.network.iosxr.iosxr_command.run_commands')
35        self.run_commands = self.mock_run_commands.start()
36
37    def tearDown(self):
38        self.mock_run_commands.stop()
39
40    def load_fixtures(self, commands=None):
41
42        def load_from_file(*args, **kwargs):
43            module, commands = args
44            output = list()
45
46            for item in commands:
47                try:
48                    obj = json.loads(item)
49                    command = obj['command']
50                except ValueError:
51                    command = item
52                filename = str(command).replace(' ', '_')
53                output.append(load_fixture(filename))
54            return output
55
56        self.run_commands.side_effect = load_from_file
57
58    def test_iosxr_command_simple(self):
59        set_module_args(dict(commands=['show version']))
60        result = self.execute_module()
61        self.assertEqual(len(result['stdout']), 1)
62        self.assertTrue(result['stdout'][0].startswith('Cisco IOS XR Software'))
63
64    def test_iosxr_command_multiple(self):
65        set_module_args(dict(commands=['show version', 'show version']))
66        result = self.execute_module()
67        self.assertEqual(len(result['stdout']), 2)
68        self.assertTrue(result['stdout'][0].startswith('Cisco IOS XR Software'))
69
70    def test_iosxr_command_wait_for(self):
71        wait_for = 'result[0] contains "Cisco IOS"'
72        set_module_args(dict(commands=['show version'], wait_for=wait_for))
73        self.execute_module()
74
75    def test_iosxr_command_wait_for_fails(self):
76        wait_for = 'result[0] contains "test string"'
77        set_module_args(dict(commands=['show version'], wait_for=wait_for))
78        self.execute_module(failed=True)
79        self.assertEqual(self.run_commands.call_count, 10)
80
81    def test_iosxr_command_retries(self):
82        wait_for = 'result[0] contains "test string"'
83        set_module_args(dict(commands=['show version'], wait_for=wait_for, retries=2))
84        self.execute_module(failed=True)
85        self.assertEqual(self.run_commands.call_count, 2)
86
87    def test_iosxr_command_match_any(self):
88        wait_for = ['result[0] contains "Cisco IOS"',
89                    'result[0] contains "test string"']
90        set_module_args(dict(commands=['show version'], wait_for=wait_for, match='any'))
91        self.execute_module()
92
93    def test_iosxr_command_match_all(self):
94        wait_for = ['result[0] contains "Cisco IOS"',
95                    'result[0] contains "XR Software"']
96        set_module_args(dict(commands=['show version'], wait_for=wait_for, match='all'))
97        self.execute_module()
98
99    def test_iosxr_command_match_all_failure(self):
100        wait_for = ['result[0] contains "Cisco IOS"',
101                    'result[0] contains "test string"']
102        commands = ['show version', 'show version']
103        set_module_args(dict(commands=commands, wait_for=wait_for, match='all'))
104        self.execute_module(failed=True)
105

Source: test_iosxr_command.py

# (c) 2016 Red Hat Inc.
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.

# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

import json

from ansible.compat.tests.mock import patch
from ansible.modules.network.iosxr import iosxr_command
from .iosxr_module import TestIosxrModule, load_fixture, set_module_args


class TestIosxrCommandModule(TestIosxrModule):

    module = iosxr_command

    def setUp(self):
        self.mock_run_commands = patch('ansible.modules.network.iosxr.iosxr_command.run_commands')
        self.run_commands = self.mock_run_commands.start()

    def tearDown(self):
        self.mock_run_commands.stop()

    def load_fixtures(self, commands=None):

        def load_from_file(*args, **kwargs):
            module, commands = args
            output = list()

            for item in commands:
                try:
                    obj = json.loads(item)
                    command = obj['command']
                except ValueError:
                    command = item
                filename = str(command).replace(' ', '_')
                output.append(load_fixture(filename))
            return output

        self.run_commands.side_effect = load_from_file

    def test_iosxr_command_simple(self):
        set_module_args(dict(commands=['show version']))
        result = self.execute_module()
        self.assertEqual(len(result['stdout']), 1)
        self.assertTrue(result['stdout'][0].startswith('Cisco IOS XR Software'))

    def test_iosxr_command_multiple(self):
        set_module_args(dict(commands=['show version', 'show version']))
        result = self.execute_module()
        self.assertEqual(len(result['stdout']), 2)
        self.assertTrue(result['stdout'][0].startswith('Cisco IOS XR Software'))

    def test_iosxr_command_wait_for(self):
        wait_for = 'result[0] contains "Cisco IOS"'
        set_module_args(dict(commands=['show version'], wait_for=wait_for))
        self.execute_module()

    def test_iosxr_command_wait_for_fails(self):
        wait_for = 'result[0] contains "test string"'
        set_module_args(dict(commands=['show version'], wait_for=wait_for))
        self.execute_module(failed=True)
        self.assertEqual(self.run_commands.call_count, 10)

    def test_iosxr_command_retries(self):
        wait_for = 'result[0] contains "test string"'
        set_module_args(dict(commands=['show version'], wait_for=wait_for, retries=2))
        self.execute_module(failed=True)
        self.assertEqual(self.run_commands.call_count, 2)

    def test_iosxr_command_match_any(self):
        wait_for = ['result[0] contains "Cisco IOS"',
                    'result[0] contains "test string"']
        set_module_args(dict(commands=['show version'], wait_for=wait_for, match='any'))
        self.execute_module()

    def test_iosxr_command_match_all(self):
        wait_for = ['result[0] contains "Cisco IOS"',
                    'result[0] contains "XR Software"']
        set_module_args(dict(commands=['show version'], wait_for=wait_for, match='all'))
        self.execute_module()

    def test_iosxr_command_match_all_failure(self):
        wait_for = ['result[0] contains "Cisco IOS"',
                    'result[0] contains "test string"']
        commands = ['show version', 'show version']
        set_module_args(dict(commands=commands, wait_for=wait_for, match='all'))
        self.execute_module(failed=True)