KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _call_all_drones method in autotest

Best Python code snippet using autotest_python

drone_manager.py

Source:drone_manager.py

...85        self._results_drone = drones.get_drone(results_repository_hostname)86        # don't initialize() the results drone - we don't want to clear out any87        # directories and we don't need ot kill any processes88    def reinitialize_drones(self):89        self._call_all_drones('initialize', self._results_dir)90    def shutdown(self):91        for drone in self.get_drones():92            drone.shutdown()93    def _add_drone(self, hostname):94        logging.info('Adding drone %s' % hostname)95        drone = drones.get_drone(hostname)96        self._drones[drone.hostname] = drone97        return drone98    def _remove_drone(self, hostname):99        self._drones.pop(hostname, None)100    def refresh_drone_configs(self):101        """102        Reread global config options for all drones.103        """104        config = global_config.global_config105        section = scheduler_config.CONFIG_SECTION106        config.parse_config_file()107        for hostname, drone in self._drones.iteritems():108            disabled = config.get_config_value(109                section, '%s_disabled' % hostname, default='')110            drone.enabled = not bool(disabled)111            drone.max_processes = config.get_config_value(112                section, '%s_max_processes' % hostname, type=int,113                default=scheduler_config.config.max_processes_per_drone)114    def get_drones(self):115        return self._drones.itervalues()116    def _get_drone_for_process(self, process):117        return self._drones[process.hostname]118    def _get_drone_for_pidfile_id(self, pidfile_id):119        pidfile_contents = self.get_pidfile_contents(pidfile_id)120        assert pidfile_contents.process is not None121        return self._get_drone_for_process(pidfile_contents.process)122    def _drop_old_pidfiles(self):123        for pidfile_id, age in self._pidfile_age.items():124            if age > self._MAX_PIDFILE_AGE:125                del self._pidfile_age[pidfile_id]126            else:127                self._pidfile_age[pidfile_id] += 1128    def _reset(self):129        self._processes = {}130        self._process_set = set()131        self._pidfiles = {}132        self._pidfiles_second_read = {}133        self._drone_queue = []134    def _call_all_drones(self, method, *args, **kwargs):135        all_results = {}136        for drone in self.get_drones():137            all_results[drone] = drone.call(method, *args, **kwargs)138        return all_results139    def _parse_pidfile(self, drone, raw_contents):140        contents = PidfileContents()141        if not raw_contents:142            return contents143        lines = raw_contents.splitlines()144        if len(lines) > 3:145            return InvalidPidfile('Corrupt pid file (%d lines):\n%s' %146                                  (len(lines), lines))147        try:148            pid = int(lines[0])149            contents.process = Process(drone.hostname, pid)150            # if len(lines) == 2, assume we caught Autoserv between writing151            # exit_status and num_failed_tests, so just ignore it and wait for152            # the next cycle153            if len(lines) == 3:154                contents.exit_status = int(lines[1])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'])...

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.