How to use _launch method in Playwright Python

Best Python code snippet using playwright-python

rapp.py

Source:rapp.py

1#2# License: BSD3#   https://raw.github.com/robotics-in-py/rocon_app_platform/license/LICENSE4#5##############################################################################6# Imports7##############################################################################8import copy9import os10import subprocess11import tempfile12import rospy13import traceback14import rocon_python_utils15import rocon_app_manager_msgs.msg as rapp_manager_msgs16import rocon_std_msgs.msg as rocon_std_msgs17from .exceptions import MissingCapabilitiesException18from . import utils19##############################################################################20# Class21##############################################################################22class Rapp(object):23    '''24        Got many inspiration and imported from willow_app_manager implementation25    '''26    # standard args that can be put inside a rapp launcher, the rapp manager27    # will fill these args in when starting the rapp28    def __init__(self, rapp_specification):29        self._connections = _create_empty_connection_type_dictionary()30        self._raw_data = rapp_specification31        self.data = rapp_specification.data32        self.data['status'] = 'Ready'33        self.data['published_interfaces'] = []34        self.data['published_parameters'] = {}35        self.data['implementations'] = []36    def __repr__(self):37        string = ''38        for d in self.data:39            string += d + ' : ' + str(self.data[d]) + '\n'40        return string41    def to_msg(self):42        '''43          Converts this app definition to ros msg format.44          :returns: ros message format of Rapp45          :rtype: rocon_app_manager_msgs.Rapp46        '''47        a = rapp_manager_msgs.Rapp()48        a.name = self.data['name']49        a.display_name = self.data['display_name']50        a.description = self.data['description']51        a.compatibility = self.data['compatibility']52        a.status = self.data['status']53        a.icon = rocon_python_utils.ros.icon_to_msg(self.data['icon'])54        a.implementations = []55        key = 'public_interface'56        if key in self.data:57            a.public_interface = [rocon_std_msgs.KeyValue(key, str(val)) for key, val in self.data[key].items()]58        key = 'public_parameters'59        if key in self.data:60            a.public_parameters = [rocon_std_msgs.KeyValue(str(key), str(val)) for key, val in self.data[key].items()]61        return a62    def install(self, dependency_checker):63        '''64          Installs all dependencies of the specified rapp65          :param dependency_checker: DependencyChecker object for installation of the rapp dependencies66          :type dependency_checker: :py:class:`rocon_app_utilities.rapp_repositories.DependencyChecker`67          :returns: A C{tuple} of a flag for the installation success and a string containing the reason of failure68          :rtype: C{tuple}69        '''70        success = False71        # Trigger the installation of all rapp dependencies72        rapps = []73        rapps.append(self.data['name'])74        try:75            dependency_checker.install_rapp_dependencies(rapps)76        except Exception as e:77            return success, str(e)78        # Update the rospack cache79        devnull = open(os.devnull, 'w')80        subprocess.call(['rospack', 'profile'], stdout=devnull, stderr=subprocess.STDOUT)81        devnull = devnull.close()82        success = True83        return success, str()84    def published_interfaces_to_msg_list(self):85        '''86        Convert the published interfaces as a rocon_app_manager_msgs.PublishedInterface list.87        '''88        return self.data['published_interfaces']89    def published_parameters_to_msg_list(self):90        '''91        Convert the published parameters as a rocon_std_msgs.KeyValue list.92        '''93        return utils.dict_to_key_value_msg(self.data['published_parameters'])94    def start(self, launch_arg_mappings, remappings=[], parameters=[], force_screen=False, caps_list=None):95        '''96          Some important jobs here.97          1) run the rapp launcher under the unique robot name namespace98          This guarantees that flipped entities generate unique node id's that won't collide when communicating99          with each other (refer to https://github.com/robotics-in-py/rocon_multimaster/issues/136).100          2) Apply remapping rules while ignoring the namespace underneath.101          :param utils.LaunchArgMappings launch_arg_mappings : args to be passed down from the rapp manager to rapp102          :param remapping: rules for the app flips.103          :type remapping: list of rocon_std_msgs.msg.Remapping values.104          :param parameters: requested public_parameters105          :type parameters: list of rocon_std_msgs.msg.KeyValue106          :param bool force_screen: whether to roslaunch the app with --screen or not107          :param caps_list: this holds the list of available capabilities, if app needs capabilities108          :type caps_list: CapsList109        '''110        data = self.data111        application_namespace = launch_arg_mappings.application_namespace112        try:113            published_parameters = utils.apply_requested_public_parameters(data['public_parameters'], parameters)114            temp = tempfile.NamedTemporaryFile(mode='w+t', delete=False)115            self._launch = utils.prepare_launcher(data, published_parameters, force_screen, launch_arg_mappings, temp)116            # Better logic for the future, 1) get remap rules from capabilities. 2) get remap rules from requets. 3) apply them all. It would be clearer to understand the logic and easily upgradable117            if 'required_capabilities' in data:  # apply capability-specific remappings needed118                utils.apply_remapping_rules_from_capabilities(self._launch, data, caps_list)119            self._connections, published_interfaces = utils.apply_remapping_rules_from_start_app_request(self._launch, data, remappings, application_namespace)120            utils.resolve_chain_remappings(self._launch.config.nodes)121            self._launch.start()122            data['status'] = 'Running'123            data['published_parameters'] = published_parameters124            data['published_interfaces'] = published_interfaces125            connections = copy.deepcopy(self._connections)126            return True, "Success", connections127            # return True, "Success", self._connections['subscribers'], self._connections['publishers'], \128            #    self._connections['services'], self._connections['action_clients'], self._connections['action_servers']129        except rospy.ServiceException as e:130            rospy.logerr("App Manager : Couldn't get cap remappings. Error: " + str(e))131            return False, "Error while launching " + data['name'], _create_empty_connection_type_dictionary()132        except MissingCapabilitiesException as e:133            rospy.logerr("Rapp Manager : couldn't get capability remappings. Error: " + str(e))134            return False, "Error while launching " + data['name'], _create_empty_connection_type_dictionary()135        except Exception as e:136            traceback.print_stack()137            rospy.logerr("Rapp Manager : error while launching " + data['launch'] + " : " + str(e))138            data['status'] = "Error while launching " + data['launch']139            return False, "Error while launching " + data['name'], _create_empty_connection_type_dictionary()140        finally:141            os.unlink(temp.name)142    def stop(self):143        data = self.data144        connections = copy.deepcopy(self._connections)145        try:146            if self._launch:147                try:148                    self._launch.shutdown()149                finally:150                    self._launch = None151                    data['status'] = 'Ready'152                    data['published_parameters'] = {}153                    data['published_interfaces'] = []154                rospy.loginfo("Rapp Manager : stopped rapp [%s]" % data['name'] + "'.")155        except Exception as e:156            print str(e)157            error_msg = "Error while stopping rapp '" + data['name'] + "'."158            rospy.loginfo(error_msg)159            data['status'] = 'Error'160            return False, error_msg, connections161        return True, "Success", connections162    def is_running(self):163        '''164         Is the rapp both launched and currently running?165         Actually three possible states 1) not launched 2) running, 3) stopped166         Could acutally return a tertiary value, but rapp manager doesn't need167         to make any decision making about that (for now), so just return168         running or not.169         Used by the rapp_manager_script.170         :returns: True if the rapp is executing or False otherwise.171         :rtype: Bool172        '''173        if not self._launch:174            return False175        elif self._launch.pm and self._launch.pm.done:176            # time.sleep(1.0)  # do we need this sleep?177            return False178        return True179##############################################################################180# Utilities181##############################################################################182def convert_rapps_from_rapp_specs(rapp_specs):183    '''184      Converts rocon_app_utilities.Rapp into rocon_app_manager.Rapp185      :param rapp_specs: dict of rapp specification186      :type rapp_specs: {ancestor_name: rocon_app_utilities.Rapp}187      :returns: runnable rapps188      :rtype: {ancestor_name:rocon_app_manager.Rapp}189    '''190    runnable_rapps = {}191    for name, spec in rapp_specs.items():192        r = Rapp(spec)193        r.data['ancestor_name'] = spec.ancestor_name194        runnable_rapps[name] = r195    return runnable_rapps196def _create_empty_connection_type_dictionary():197    '''198      Initialise connections which use as public interface199      :returns: dict of connections with empty list200      :rtype: {connection_type: list}201    '''202    PUBLIC_CONNECTION_TYPES = ['publishers', 'subscribers', 'services', 'action_clients', 'action_servers']203    connections = {}204    for connection_type in PUBLIC_CONNECTION_TYPES:205        connections[connection_type] = []...

test_distributed.py

Source:test_distributed.py

...81    module = DistributedDataParallel(module, **args)82    output = step(module, input)83    grads = {n: p.grad for n, p in module.named_parameters()}84    return output.detach(), module.state_dict(), grads85def _launch(inputs,86            modules=None,87            args=None,88            step=Steps._step,89            device_type="cpu"):90    procs = []91    with tempfile.TemporaryDirectory() as tmpdir, context.Pool(2) as pool:92        if modules is None:93            modules = [MyModule(), MyModule()]94        if args is None:95            args = {}96        file = os.path.join(tmpdir, "init")97        for i, (input, module) in enumerate(zip(inputs, modules)):98            p = pool.apply_async(99                _run,100                args=(file, input, module, i, args, step,101                      device_type))102            procs.append(p)103        return [p.get() for p in procs]104def _device_types():105    retval = ["cpu"]106    if torch.cuda.is_available() and torch.cuda.device_count() >= 2:107        retval.append("cuda")108    return retval109@pytest.mark.skipif(110    sys.platform == 'win32',111    reason='DDP not fully supported on Windows')112class TestDistributedDataParallel:113    def test_save_load(self):114        module = MyModule()115        with_ddp = DistributedDataParallel(module)116        assert module.state_dict().keys() == with_ddp.state_dict().keys()117        module.load_state_dict(with_ddp.state_dict())118        assert np.array_equal(module.state_dict()["param0"],119                              with_ddp.state_dict()["param0"])120        assert np.array_equal(module.state_dict()["param1"],121                              with_ddp.state_dict()["param1"])122        assert np.array_equal(module.state_dict()["buffer"],123                              with_ddp.state_dict()["buffer"])124    @pytest.mark.parametrize('device_type', _device_types())125    def test_sync_init_params(self, device_type):126        module0 = MyModule()127        module0.param0.data = torch.tensor([1.])128        r0, r1 = _launch(129            inputs=[torch.tensor([1.]), torch.tensor([2.])],130            modules=[module0, MyModule()],131            device_type=device_type)132        assert r0[0].item() == 1133        assert r1[0].item() == 2134        assert r0[1]["param0"].item() == 1.0135        assert r1[1]["param0"].item() == 1.0136    @pytest.mark.parametrize('device_type', _device_types())137    def test_all_reduce(self, device_type):138        r0, r1 = _launch(139            inputs=[torch.tensor([1.]), torch.tensor([2.])],140            device_type=device_type)141        assert r0[0].item() == -1142        assert r1[0].item() == -2143        assert r0[2]["module.param0"].item() == 1.5144        assert r1[2]["module.param0"].item() == 1.5145        assert r0[2]["module.param1"] is None146        assert r1[2]["module.param1"] is None147    @pytest.mark.parametrize('device_type', _device_types())148    def test_specific_reduce(self, device_type):149        r0, r1 = _launch(150            inputs=[torch.tensor([1.]), torch.tensor([2.])],151            args={"reduce_function": Collectives._to_zero},152            device_type=device_type)153        assert r0[2]["module.param0"].item() == 0.0154        assert r1[2]["module.param0"].item() == 0.0155    @pytest.mark.parametrize('device_type', _device_types())156    def test_nosync_buffer(self, device_type):157        r0, r1 = _launch(158            inputs=[torch.tensor([1.]), torch.tensor([2.])],159            args={"broadcast_buffers": False},160            device_type=device_type)161        assert r0[0].item() == -1162        assert r1[0].item() == -2163        assert r0[1]["buffer"].item() == 1164        assert r1[1]["buffer"].item() == 2165    @pytest.mark.parametrize('device_type', _device_types())166    def test_sync_buffer(self, device_type):167        r0, r1 = _launch(168            inputs=[torch.tensor([1.]), torch.tensor([2.])],169            args={"broadcast_buffers": True},170            device_type=device_type)171        assert r0[0].item() == -1172        assert r1[0].item() == -2173        assert r0[1]["buffer"].item() == 1174        assert r1[1]["buffer"].item() == 1175    @pytest.mark.parametrize('device_type', _device_types())176    def test_specific_broadcast(self, device_type):177        r0, r1 = _launch(178            inputs=[torch.tensor([1.]), torch.tensor([2.])],179            args={"broadcast_function": Collectives._to_zero,180                  "broadcast_buffers": True},181            device_type=device_type)182        assert r0[1]["buffer"].item() == 0.0183        assert r1[1]["buffer"].item() == 0.0184    @pytest.mark.parametrize('device_type', _device_types())185    def test_define_by_run(self, device_type):186        r0, r1 = _launch(187            inputs=[torch.tensor([1.]), torch.tensor([-1])],188            device_type=device_type)189        assert r0[0].item() == -1190        assert r1[0].item() == 1191        assert r0[2]["module.param0"].item() == 0.5192        assert r1[2]["module.param0"].item() == 0.5193        assert r0[2]["module.param1"].item() == 0.5194        assert r1[2]["module.param1"].item() == 0.5195    @pytest.mark.parametrize('device_type', _device_types())196    def test_no_sync(self, device_type):197        r0, r1 = _launch(198            inputs=[torch.tensor([1.]), torch.tensor([2.])],199            step=Steps._step_with_no_sync,200            device_type=device_type)201        assert r0[0].item() == -1202        assert r1[0].item() == -2203        assert r0[2]["module.param0"].item() == 1204        assert r1[2]["module.param0"].item() == 2205        assert r0[2]["module.param1"] is None206        assert r1[2]["module.param1"] is None207    @pytest.mark.parametrize('device_type', _device_types())208    def test_hook(self, device_type):209        r0, r1 = _launch(210            inputs=[torch.tensor([1.]), torch.tensor([2.])],211            step=Steps._step_with_hook,212            device_type=device_type)213        assert r0[0].item() == -1214        assert r1[0].item() == -2215        assert r0[2]["module.param0"].item() == 0216        assert r1[2]["module.param0"].item() == 0217        assert r0[2]["module.param1"].item() == 0218        assert r1[2]["module.param1"].item() == 0219    @pytest.mark.parametrize('device_type', _device_types())220    @pytest.mark.skipif(221        not pytorch_pfn_extras.requires("1.6.0"),222        reason="Variable._execution_engine.queue_callback does not work "223               "with checkpointing when torch < 1.6.0")224    def test_checkpoint(self, device_type):225        r0, r1 = _launch(226            inputs=[torch.tensor([[1.]]), torch.tensor([[2.]])],227            modules=[MyModuleWithCheckpoint(), MyModuleWithCheckpoint()],228            step=Steps._step_with_hook,229            device_type=device_type)230        grad0 = r0[2]231        grad1 = r1[2]232        for key in grad0.keys():233            assert np.array_equal(grad0[key].cpu().numpy(),...

torque.py

Source:torque.py

...22    23    24    def submit( self, cmd ):25        cmds = [ r'echo \"%s\" | qsub' % (cmd,) ]26        failed, output, error = self._launch( cmds )27        if failed:28            if error.find( 'check pbs_server daemon' ) != -1:29                from exceptions import SchedulerDaemonNotStarted30                raise SchedulerDaemonNotStarted, "pbs_server"31            msg = "error in executing cmds %s. output: %s, error: %s" % (32                cmds, output, error )33            raise RuntimeError, msg34        return output35    36    def status( self, jobid ):37        38        cmds = [ 'qstat -f %s' % (jobid,) ]39        failed, output, error  = self._launch( cmds )40        if failed:41            if error.find( 'Unknown Job Id' ) != -1:42                return self.statusByTracejob( job )43            msg = "error in executing cmds %s. output: %s, error: %s" % (44                cmds, output, error )45            raise RuntimeError, msg46        47        lines = output.split( '\n' )48        lines = lines[1:] # first line removed49        if len(lines) == 0: return self.statusByTracejob( job )50        d = {}51        for line in lines:52            try:53                k,v = line.split( '=' )54            except:55                continue56            d[ k.strip() ] = v.strip()57            continue58        errorpath = d['Error_Path']59        dummy, errorfilename = os.path.split(errorpath)60        outputpath = d['Output_Path']61        dummy, outputfilename = os.path.split(outputpath)62        state = d['job_state']63        import time64        start_time = d.get('start_time') or time.ctime()65        66        ret = {67            'remote_outputfilename': outputfilename,68            'remote_errorfilename': errorfilename,69            'state': _state( state ),70            'timeStart': start_time,71            }72        if ret['state'] == 'finished':73            output, error = self._readoutputerror(74                outputfilename, errorfilename )75            ret.update(76                { 'exit_code': d['exit_status'],77                  'timeCompletion': d['etime'],78                  'output': output,79                  'error': error,80                  } )81            pass82        return ret83    def statusByTracejob( self, job ):84        jobid = job.id_incomputingserver85        d = {}86        87        tag = 'Exit_status'88        words = self._tracejob_search( jobid, tag )89        status = words[3]90        key, value = status.split( '=' )91        assert key.lower() == 'exit_status'92        d [ 'exit_code' ] =  value93        tag = 'job was terminated'94        words = self._tracejob_search( jobid, tag )95        d[ 'timeCompletion' ] = ' '.join( words[0:2] )96        output, error = self._readoutputerror(97            job.outputfilename, job.errorfilename )98        d.update( {99            'output': output,100            'error': error,101            } )102            103        return d104    def _readoutputerror(self, outputfilename, errorfilename ):105        return self._read( outputfilename ), self._read( errorfilename )106    def _read(self, filename):107        'read file in the remote job directory'108        cmds = [ 'tail %r' % (filename,) ]109        failed, output, error = self._launch( cmds )110        if failed:111            msg = "error in executing cmds %s. output: %s, error: %s" % (112                cmds, output, error )113            raise RuntimeError, msg114        maxlen = self.outputstr_maxlen115        return output[-maxlen+1:]116    def _tracejob_search(self, jobid, tag):117        cmds = [ 'tracejob %s | grep %r' % (jobid, tag) ]118        119        failed, output, error = self._launch( cmds )120        if failed:121            msg = "error in executing cmds %s. output: %s, error: %s" % (122                cmds, output, error )123            raise RuntimeError, msg124        # remove trailing \n to make parsing easier125        if output.endswith( '\n' ): output = output[:-1] 126        lines = output.split( '\n' )127        words = lines[-1].split( )128        debug.log( 'words: %s' % words )129        return words130    131    def _launch(self, cmds):132        if self.prefix: cmds = [ self.prefix ] + cmds133        return self.launcher( ' && '.join( cmds ) )134    pass # end of Scheduler135import os136_states = {137    'C': 'finished',138    'R': 'running',139    'Q': 'queued',140    'E': 'exiting', #after having run141    'H': 'onhold',142    'W': 'waiting',143    'S': 'suspend',144    }145    ...

test_samples.py

Source:test_samples.py

...63                    'ASK_VERIFY_REQUESTS': 'false'}64        if os.name == 'nt':65            self.env['SYSTEMROOT'] = os.getenv('SYSTEMROOT')66            self.env['PATH'] = os.getenv('PATH')67    def _launch(self, sample):68        prefix = os.path.join(project_root, 'samples/')69        path = prefix + sample70        process = subprocess.Popen([self.python, path], env=self.env)71        time.sleep(1)72        self.assertIsNone(process.poll(),73                          msg='Poll should work,'74                          'otherwise we failed to launch')75        self.process = process76    def _post(self, route='/', data={}):77        url = 'http://127.0.0.1:5000' + str(route)78        print('POSTing to %s' % url)79        response = post(url, json=data)80        self.assertEqual(200, response.status_code)81        return response82    @staticmethod83    def _get_text(http_response):84        data = http_response.json()85        return data.get('response', {})\86                   .get('outputSpeech', {})\87                   .get('text', None)88    @staticmethod89    def _get_reprompt(http_response):90        data = http_response.json()91        return data.get('response', {})\92                   .get('reprompt', {})\93                   .get('outputSpeech', {})\94                   .get('text', None)95    def tearDown(self):96        try:97            self.process.terminate()98            self.process.communicate(timeout=1)99        except Exception as e:100            try:101                print('[%s]...trying to kill.' % str(e))102                self.process.kill()103                self.process.communicate(timeout=1)104            except Exception as e:105                print('Error killing test python process: %s' % str(e))106                print('*** it is recommended you manually kill with PID %s',107                      self.process.pid)108    def test_helloworld(self):109        """ Test the HelloWorld sample project """110        self._launch('helloworld/helloworld.py')111        response = self._post(data=launch)112        self.assertTrue('hello' in self._get_text(response))113    def test_session_sample(self):114        """ Test the Session sample project """115        self._launch('session/session.py')116        response = self._post(data=launch)117        self.assertTrue('favorite color' in self._get_text(response))118    def test_audio_simple_demo(self):119        """ Test the SimpleDemo Audio sample project """120        self._launch('audio/simple_demo/ask_audio.py')121        response = self._post(data=launch)122        self.assertTrue('audio example' in self._get_text(response))123    def test_audio_playlist_demo(self):124        """ Test the Playlist Audio sample project """125        self._launch('audio/playlist_demo/playlist.py')126        response = self._post(data=launch)127        self.assertTrue('playlist' in self._get_text(response))128    def test_blueprints_demo(self):129        """ Test the sample project using Flask Blueprints """130        self._launch('blueprint_demo/demo.py')131        response = self._post(route='/ask', data=launch)132        self.assertTrue('hello' in self._get_text(response))133    def test_history_buff(self):134        """ Test the History Buff sample """135        self._launch('historybuff/historybuff.py')136        response = self._post(data=launch)137        self.assertTrue('History buff' in self._get_text(response))138    def test_spacegeek(self):139        """ Test the Spacegeek sample """140        self._launch('spacegeek/spacegeek.py')141        response = self._post(data=launch)142        # response is random143        self.assertTrue(len(self._get_text(response)) > 1)144    def test_tidepooler(self):145        """ Test the Tide Pooler sample """146        self._launch('tidepooler/tidepooler.py')147        response = self._post(data=launch)...

Playwright tutorial

LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.

Chapters:

What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.