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

How to use _wait_for_some_async_commands method in autotest

Best Python code snippet using autotest_python

drone_utility.py

Source:drone_utility.py

...272        for subproc in self._subcommands:273            if subproc.poll() is None:274                still_running.append(subproc)275        self._subcommands = still_running276    def _wait_for_some_async_commands(self):277        self._poll_async_commands()278        max_processes = scheduler_config.config.max_transfer_processes279        while len(self._subcommands) >= max_processes:280            time.sleep(1)281            self._poll_async_commands()282    def run_async_command(self, function, args):283        subproc = subcommand.subcommand(function, args)284        self._subcommands.append(subproc)285        subproc.fork_start()286    def _sync_get_file_from(self, hostname, source_path, destination_path):287        self._ensure_directory_exists(os.path.dirname(destination_path))288        host = create_host(hostname)289        host.get_file(source_path, destination_path, delete_dest=True)290    def get_file_from(self, hostname, source_path, destination_path):291        self.run_async_command(self._sync_get_file_from,292                               (hostname, source_path, destination_path))293    def sync_send_file_to(self, hostname, source_path, destination_path,294                          can_fail):295        host = create_host(hostname)296        try:297            host.run('mkdir -p ' + os.path.dirname(destination_path))298            host.send_file(source_path, destination_path, delete_dest=True)299        except error.AutoservError:300            if not can_fail:301                raise302            if os.path.isdir(source_path):303                failed_file = os.path.join(source_path, _TRANSFER_FAILED_FILE)304                file_object = open(failed_file, 'w')305                try:306                    file_object.write('%s:%s\n%s\n%s' %307                                      (hostname, destination_path,308                                       datetime.datetime.now(),309                                       traceback.format_exc()))310                finally:311                    file_object.close()312            else:313                copy_to = destination_path + _TRANSFER_FAILED_FILE314                self._ensure_directory_exists(os.path.dirname(copy_to))315                self.copy_file_or_directory(source_path, copy_to)316    def send_file_to(self, hostname, source_path, destination_path,317                     can_fail=False):318        self.run_async_command(self.sync_send_file_to,319                               (hostname, source_path, destination_path,320                                can_fail))321    def _report_long_execution(self, calls, duration):322        call_count = {}323        for call in calls:324            call_count.setdefault(call._method, 0)325            call_count[call._method] += 1326        call_summary = '\n'.join('%d %s' % (count, method)327                                 for method, count in call_count.iteritems())328        self._warn('Execution took %f sec\n%s' % (duration, call_summary))329    def execute_calls(self, calls):330        results = []331        start_time = time.time()332        max_processes = scheduler_config.config.max_transfer_processes333        for method_call in calls:334            results.append(method_call.execute_on(self))335            if len(self._subcommands) >= max_processes:336                self._wait_for_some_async_commands()337        self.wait_for_all_async_commands()338        duration = time.time() - start_time339        if duration > self._WARNING_DURATION:340            self._report_long_execution(calls, duration)341        warnings = self.warnings342        self.warnings = []343        return dict(results=results, warnings=warnings)344def create_host(hostname):345    username = settings.get_value('SCHEDULER', hostname + '_username',346                                  default=getpass.getuser())347    return hosts.SSHHost(hostname, user=username)348def parse_input():349    input_chunks = []350    chunk_of_input = sys.stdin.read()...

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.