How to use _least_loaded_drone method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...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 """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'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'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....

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?