Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _least_loaded_drone method in autotest

Best Python code snippet using autotest_python

drone_manager.py

Source:drone_manager.py

...483        runnable_processes = [484                wrapper.drone.max_processes - wrapper.drone.active_processes485                for wrapper in usable_drone_wrappers]486        return max([0] + runnable_processes)487    def _least_loaded_drone(self, drones):488        return min(drones, key=lambda d: d.used_capacity())489    def pick_drone_to_use(self, num_processes=1, prefer_ssp=False):490        """Return a drone to use.491        Various options can be passed to optimize drone selection.492        num_processes is the number of processes the drone is intended493        to run.494        prefer_ssp indicates whether drones supporting server-side495        packaging should be preferred.  The returned drone is not496        guaranteed to support it.497        This public API is exposed for luciferlib to wrap.498        Returns a drone instance (see drones.py).499        """500        return self._choose_drone_for_execution(501                num_processes=num_processes,502                username=None,  # Always allow all drones503                drone_hostnames_allowed=None,  # Always allow all drones504                require_ssp=prefer_ssp,505        )506    def _choose_drone_for_execution(self, num_processes, username,507                                    drone_hostnames_allowed,508                                    require_ssp=False):509        """Choose a drone to execute command.510        @param num_processes: Number of processes needed for execution.511        @param username: Name of the user to execute the command.512        @param drone_hostnames_allowed: A list of names of drone allowed.513        @param require_ssp: Require server-side packaging to execute the,514                            command, default to False.515        @return: A drone object to be used for execution.516        """517        # cycle through drones is order of increasing used capacity until518        # we find one that can handle these processes519        checked_drones = []520        usable_drones = []521        # Drones do not support server-side packaging, used as backup if no522        # drone is found to run command requires server-side packaging.523        no_ssp_drones = []524        drone_to_use = None525        while self._drone_queue:526            drone = heapq.heappop(self._drone_queue).drone527            checked_drones.append(drone)528            logging.info('Checking drone %s', drone.hostname)529            if not drone.usable_by(username):530                continue531            drone_allowed = (drone_hostnames_allowed is None532                             or drone.hostname in drone_hostnames_allowed)533            if not drone_allowed:534                logging.debug('Drone %s not allowed: ', drone.hostname)535                continue536            if require_ssp and not drone.support_ssp:537                logging.debug('Drone %s does not support server-side '538                              'packaging.', drone.hostname)539                no_ssp_drones.append(drone)540                continue541            usable_drones.append(drone)542            if drone.active_processes + num_processes <= drone.max_processes:543                drone_to_use = drone544                break545            logging.info('Drone %s has %d active + %s requested > %s max',546                         drone.hostname, drone.active_processes, num_processes,547                         drone.max_processes)548        if not drone_to_use and usable_drones:549            # Drones are all over loaded, pick the one with least load.550            drone_summary = ','.join('%s %s/%s' % (drone.hostname,551                                                   drone.active_processes,552                                                   drone.max_processes)553                                     for drone in usable_drones)554            logging.error('No drone has capacity to handle %d processes (%s) '555                          'for user %s', num_processes, drone_summary, username)556            drone_to_use = self._least_loaded_drone(usable_drones)557        elif not drone_to_use and require_ssp and no_ssp_drones:558            # No drone supports server-side packaging, choose the least loaded.559            drone_to_use = self._least_loaded_drone(no_ssp_drones)560        # refill _drone_queue561        for drone in checked_drones:562            self._enqueue_drone(drone)563        return drone_to_use564    def _substitute_working_directory_into_command(self, command,565                                                   working_directory):566        for i, item in enumerate(command):567            if item is WORKING_DIRECTORY:568                command[i] = working_directory569    def execute_command(self, command, working_directory, pidfile_name,570                        num_processes, log_file=None, paired_with_pidfile=None,571                        username=None, drone_hostnames_allowed=None):572        """573        Execute the given command, taken as an argv list....

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.