How to use wait_for method in Playwright Python

Best Python code snippet using playwright-python

Run Playwright Python automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

cmd_test.py

Source: cmd_test.py Github

copy
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
Full Screen

waitfor.py

Source: waitfor.py Github

copy
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
Full Screen

test_iosxr_command.py

Source: test_iosxr_command.py Github

copy
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
Full Screen

Accelerate Your Automation Test Cycles With LambdaTest

Leverage LambdaTest’s cloud-based platform to execute your automation tests in parallel and trim down your test execution time significantly. Your first 100 automation testing minutes are on us.

Try LambdaTest

Run Python Tests on LambdaTest Cloud Grid

Execute automation tests with Playwright Python on a cloud-based Grid of 3000+ real browsers and operating systems for both web and mobile applications.

Test now for Free
LambdaTestX

We use cookies to give you the best experience. Cookies help to provide a more personalized experience and relevant advertising for you, and web analytics for us. Learn More in our Cookies policy, Privacy & Terms of service

Allow Cookie
Sarah

I hope you find the best code examples for your project.

If you want to accelerate automated browser testing, try LambdaTest. Your first 100 automation testing minutes are FREE.

Sarah Elson (Product & Growth Lead)