KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use _get_drone_for_process method in autotest

Best Python code snippet using autotest_python
drone_manager.py
Source:drone_manager.py
...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'])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        """213        Called at the end of a scheduler cycle to execute all queued actions214        on drones.215        """216        for drone in self._drones.values():217            drone.execute_queued_calls()218        try:219             self._results_drone.execute_queued_calls()220        except error.AutoservError:221            warning = ('Results repository failed to execute calls:\n' +222                       traceback.format_exc())223            email_manager.manager.enqueue_notify_email(224                'Results repository error', warning)225            self._results_drone.clear_call_queue()226    def get_orphaned_autoserv_processes(self):227        """228        Returns a set of Process objects for orphaned processes only.229        """230        return set(process for process in self._process_set231                   if process.ppid == 1)232    def get_process_for(self, execution_tag):233        """234        Return the process object for the given execution tag.235        """236        return self._processes.get(execution_tag, None)237    def kill_process(self, process):238        """239        Kill the given process.240        """241        logging.info('killing %s', process)242        drone = self._get_drone_for_process(process)243        drone.queue_call('kill_process', process)244    def _ensure_directory_exists(self, path):245        if not os.path.exists(path):246            os.makedirs(path)247    def _extract_num_processes(self, command):248        try:249            machine_list_index = command.index('-m') + 1250        except ValueError:251            return 1252        assert machine_list_index < len(command)253        machine_list = command[machine_list_index].split(',')254        return len(machine_list)255    def total_running_processes(self):256        return sum(drone.active_processes for drone in self.get_drones())257    def max_runnable_processes(self):258        """259        Return the maximum number of processes that can be run (in a single260        execution) given the current load on drones.261        """262        if not self._drone_queue:263            # all drones disabled264            return 0265        return max(drone.max_processes - drone.active_processes266                   for _, drone in self._drone_queue)267    def _least_loaded_drone(self, drones):268        drone_to_use = drones[0]269        for drone in drones[1:]:270            if drone.used_capacity() < drone_to_use.used_capacity():271                drone_to_use = drone272        return drone_to_use273    def _choose_drone_for_execution(self, num_processes):274        # cycle through drones is order of increasing used capacity until275        # we find one that can handle these processes276        checked_drones = []277        drone_to_use = None278        while self._drone_queue:279            used_capacity, drone = heapq.heappop(self._drone_queue)280            checked_drones.append(drone)281            if drone.active_processes + num_processes <= drone.max_processes:282                drone_to_use = drone283                break284        if not drone_to_use:285            drone_summary = ','.join('%s %s/%s' % (drone.hostname,286                                                   drone.active_processes,287                                                   drone.max_processes)288                                     for drone in checked_drones)289            logging.error('No drone has capacity to handle %d processes (%s)',290                          num_processes, drone_summary)291            drone_to_use = self._least_loaded_drone(checked_drones)292        drone_to_use.active_processes += num_processes293        # refill _drone_queue294        for drone in checked_drones:295            self._enqueue_drone(drone)296        return drone_to_use297    def _substitute_working_directory_into_command(self, command,298                                                   working_directory):299        for i, item in enumerate(command):300            if item is WORKING_DIRECTORY:301                command[i] = working_directory302    def execute_command(self, command, working_directory, pidfile_name,303                        log_file=None, paired_with_pidfile=None):304        """305        Execute the given command, taken as an argv list.306        @param command: command to execute as a list.  if any item is307                WORKING_DIRECTORY, the absolute path to the working directory308                will be substituted for it.309        @param working_directory: directory in which the pidfile will be written310        @param pidfile_name: name of the pidfile this process will write311        @param log_file (optional): path (in the results repository) to hold312                command output.313        @param paired_with_pidfile (optional): a PidfileId for an314                already-executed process; the new process will execute on the315                same drone as the previous process.316        """317        working_directory = self.absolute_path(working_directory)318        if not log_file:319            log_file = self.get_temporary_path('execute')320        log_file = self.absolute_path(log_file)321        self._substitute_working_directory_into_command(command,322                                                        working_directory)323        if paired_with_pidfile:324            drone = self._get_drone_for_pidfile_id(paired_with_pidfile)325        else:326            num_processes = self._extract_num_processes(command)327            drone = self._choose_drone_for_execution(num_processes)328        logging.info("command = %s" % command)329        logging.info('log file = %s:%s' % (drone.hostname, log_file))330        self._write_attached_files(command, drone)331        drone.queue_call('execute_command', command, working_directory,332                         log_file, pidfile_name)333        pidfile_path = self.absolute_path(os.path.join(working_directory,334                                                       pidfile_name))335        pidfile_id = PidfileId(pidfile_path)336        self.register_pidfile(pidfile_id)337        return pidfile_id338    def get_pidfile_id_from(self, execution_tag, pidfile_name):339        path = os.path.join(self.absolute_path(execution_tag), pidfile_name)340        return PidfileId(path)341    def register_pidfile(self, pidfile_id):342        """343        Indicate that the DroneManager should look for the given pidfile when344        refreshing.345        """346        self._pidfile_age[pidfile_id] = 0347    def get_pidfile_contents(self, pidfile_id, use_second_read=False):348        """349        Retrieve a PidfileContents object for the given pidfile_id.  If350        use_second_read is True, use results that were read after the processes351        were checked, instead of before.352        """353        self.register_pidfile(pidfile_id)354        if use_second_read:355            pidfile_map = self._pidfiles_second_read356        else:357            pidfile_map = self._pidfiles358        return pidfile_map.get(pidfile_id, PidfileContents())359    def is_process_running(self, process):360        """361        Check if the given process is in the running process list.362        """363        return process in self._process_set364    def get_temporary_path(self, base_name):365        """366        Get a new temporary path guaranteed to be unique across all drones367        for this scheduler execution.368        """369        self._temporary_path_counter += 1370        return os.path.join(drone_utility._TEMPORARY_DIRECTORY,371                            '%s.%s' % (base_name, self._temporary_path_counter))372    def absolute_path(self, path):373        return os.path.join(self._results_dir, path)374    def _copy_results_helper(self, process, source_path, destination_path,375                             to_results_repository=False):376        full_source = self.absolute_path(source_path)377        full_destination = self.absolute_path(destination_path)378        source_drone = self._get_drone_for_process(process)379        if to_results_repository:380            source_drone.send_file_to(self._results_drone, full_source,381                                      full_destination, can_fail=True)382        else:383            source_drone.queue_call('copy_file_or_directory', full_source,384                                    full_destination)385    def copy_to_results_repository(self, process, source_path,386                                   destination_path=None):387        """388        Copy results from the given process at source_path to destination_path389        in the results repository.390        """391        if destination_path is None:392            destination_path = source_path393        self._copy_results_helper(process, source_path, destination_path,394                                  to_results_repository=True)395    def copy_results_on_drone(self, process, source_path, destination_path):396        """397        Copy a results directory from one place to another on the drone.398        """399        self._copy_results_helper(process, source_path, destination_path)400    def _write_attached_files(self, command, drone):401        execution_tag = self._extract_execution_tag(' '.join(command))402        attached_files = self._attached_files.pop(execution_tag, {})403        for file_path, contents in attached_files.iteritems():404            drone.queue_call('write_to_file', self.absolute_path(file_path),405                             contents)406    def attach_file_to_execution(self, execution_tag, file_contents,407                                 file_path=None):408        """409        When the process for execution_tag is executed, the given file contents410        will be placed in a file on the drone.  Returns the path at which the411        file will be placed.412        """413        if not file_path:414            file_path = self.get_temporary_path('attach')415        files_for_execution = self._attached_files.setdefault(execution_tag, {})416        assert file_path not in files_for_execution417        files_for_execution[file_path] = file_contents418        return file_path419    def write_lines_to_file(self, file_path, lines, paired_with_process=None):420        """421        Write the given lines (as a list of strings) to a file.  If422        paired_with_process is given, the file will be written on the drone423        running the given Process.  Otherwise, the file will be written to the424        results repository.425        """426        full_path = os.path.join(self._results_dir, file_path)427        file_contents = '\n'.join(lines) + '\n'428        if paired_with_process:429            drone = self._get_drone_for_process(paired_with_process)430        else:431            drone = self._results_drone...
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.