Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use _process_pidfiles method in autotest

Best Python code snippet using autotest_python

drone_manager.py

Source:drone_manager.py

...155                contents.num_tests_failed = int(lines[2])156        except ValueError, exc:157            return InvalidPidfile('Corrupt pid file: ' + str(exc.args))158        return contents159    def _process_pidfiles(self, drone, pidfiles, store_in_dict):160        for pidfile_path, contents in pidfiles.iteritems():161            pidfile_id = PidfileId(pidfile_path)162            contents = self._parse_pidfile(drone, contents)163            store_in_dict[pidfile_id] = contents164    def _add_process(self, drone, process_info):165        process = Process(drone.hostname, int(process_info['pid']),166                          int(process_info['ppid']))167        self._process_set.add(process)168        return process169    def _add_autoserv_process(self, drone, process_info):170        assert process_info['comm'] == 'autoserv'171        # only root autoserv processes have pgid == pid172        if process_info['pgid'] != process_info['pid']:173            return174        process = self._add_process(drone, process_info)175        execution_tag = self._execution_tag_for_process(drone, process_info)176        self._processes[execution_tag] = process177    def _enqueue_drone(self, drone):178        heapq.heappush(self._drone_queue, (drone.used_capacity(), drone))179    def refresh(self):180        """181        Called at the beginning of a scheduler cycle to refresh all process182        information.183        """184        self._reset()185        self._drop_old_pidfiles()186        pidfile_paths = [pidfile_id.path for pidfile_id in self._pidfile_age]187        all_results = self._call_all_drones('refresh', pidfile_paths)188        for drone, results_list in all_results.iteritems():189            results = results_list[0]190            drone.active_processes = len(results['autoserv_processes'])191            if drone.enabled:192                self._enqueue_drone(drone)193            for process_info in results['autoserv_processes']:194                self._add_autoserv_process(drone, process_info)195            for process_info in results['parse_processes']:196                self._add_process(drone, process_info)197            self._process_pidfiles(drone, results['pidfiles'], self._pidfiles)198            self._process_pidfiles(drone, results['pidfiles_second_read'],199                                   self._pidfiles_second_read)200    def _execution_tag_for_process(self, drone, process_info):201        execution_tag = self._extract_execution_tag(process_info['args'])202        if not execution_tag:203            # this process has no execution tag - just make up something unique204            return '%s.%s' % (drone, process_info['pid'])205        return execution_tag206    def _extract_execution_tag(self, command):207        match = re.match(r'.* -P (\S+) ', command)208        if not match:209            return None210        return match.group(1)211    def execute_actions(self):212        """...

Automation Testing Tutorials

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