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

How to use _execute_in_background method in autotest

Best Python code snippet using autotest_python

autotest.py

Source:autotest.py

...466    def log_unexpected_abort(self, stderr_redirector):467        stderr_redirector.flush_all_buffers()468        msg = "Autotest client terminated unexpectedly"469        self.host.job.record("END ABORT", None, None, msg)470    def _execute_in_background(self, section, timeout):471        full_cmd = self.get_background_cmd(section)472        devnull = open(os.devnull, "w")473        self.copy_client_config_file(self.get_client_log())474        self.host.job.push_execution_context(self.results_dir)475        try:476            result = self.host.run(full_cmd, ignore_status=True,477                                   timeout=timeout,478                                   stdout_tee=devnull,479                                   stderr_tee=devnull)480        finally:481            self.host.job.pop_execution_context()482        return result483    @staticmethod484    def _strip_stderr_prologue(stderr):485        """Strips the 'standard' prologue that get pre-pended to every486        remote command and returns the text that was actually written to487        stderr by the remote command."""488        stderr_lines = stderr.split("\n")[1:]489        if not stderr_lines:490            return ""491        elif stderr_lines[0].startswith("NOTE: autotestd_monitor"):492            del stderr_lines[0]493        return "\n".join(stderr_lines)494    def _execute_daemon(self, section, timeout, stderr_redirector,495                        client_disconnect_timeout):496        monitor_dir = self.host.get_tmp_dir()497        daemon_cmd = self.get_daemon_cmd(section, monitor_dir)498        # grab the location for the server-side client log file499        client_log_prefix = self.get_client_log()500        client_log_path = os.path.join(self.results_dir, 'debug',501                                       client_log_prefix + '.log')502        client_log = open(client_log_path, 'w', 0)503        self.copy_client_config_file(client_log_prefix)504        stdout_read = stderr_read = 0505        self.host.job.push_execution_context(self.results_dir)506        try:507            self.host.run(daemon_cmd, ignore_status=True, timeout=timeout)508            disconnect_warnings = []509            while True:510                monitor_cmd = self.get_monitor_cmd(monitor_dir, stdout_read,511                                                   stderr_read)512                try:513                    result = self.host.run(monitor_cmd, ignore_status=True,514                                           timeout=timeout,515                                           stdout_tee=client_log,516                                           stderr_tee=stderr_redirector)517                except error.AutoservRunError, e:518                    result = e.result_obj519                    result.exit_status = None520                    disconnect_warnings.append(e.description)521                    stderr_redirector.log_warning(522                        "Autotest client was disconnected: %s" % e.description,523                        "NETWORK")524                except error.AutoservSSHTimeout:525                    result = utils.CmdResult(monitor_cmd, "", "", None, 0)526                    stderr_redirector.log_warning(527                        "Attempt to connect to Autotest client timed out",528                        "NETWORK")529                stdout_read += len(result.stdout)530                stderr_read += len(self._strip_stderr_prologue(result.stderr))531                if result.exit_status is not None:532                    return result533                elif not self.host.wait_up(client_disconnect_timeout):534                    raise error.AutoservSSHTimeout(535                        "client was disconnected, reconnect timed out")536        finally:537            client_log.close()538            self.host.job.pop_execution_context()539    def execute_section(self, section, timeout, stderr_redirector,540                        client_disconnect_timeout):541        logging.info("Executing %s/bin/autotest %s/control phase %d",542                     self.autodir, self.autodir, section)543        if self.background:544            result = self._execute_in_background(section, timeout)545        else:546            result = self._execute_daemon(section, timeout, stderr_redirector,547                                          client_disconnect_timeout)548        last_line = stderr_redirector.last_line549        # check if we failed hard enough to warrant an exception550        if result.exit_status == 1:551            err = error.AutotestRunError("client job was aborted")552        elif not self.background and not result.stderr:553            err = error.AutotestRunError(554                "execute_section %s failed to return anything\n"555                "stdout:%s\n" % (section, result.stdout))556        else:557            err = None558        # log something if the client failed AND never finished logging...

mvn_repair.py

Source:mvn_repair.py

...25            self.startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW26    @handle_error27    def execute(self, command, silently=False, **kwargs):28        if silently:29            return self._execute_in_background(command)30        else:31            return self._execute_in_foreground(command)32    def _execute_in_foreground(self, command):33        os.system(command)34    def _execute_in_background(self, command):35        command_arguments = command.split(" ")36        result  = subprocess.run(command_arguments, stdout=subprocess.PIPE, stderr=subprocess.PIPE)37        if result.stderr:38            raise Exception(result.stderr)39        return result.stdout.decode()40class AtlasMavenAPI:41    def __init__(self, command_executer):42        self.command_executer = command_executer43    def clean(self, **kwargs):44        self._execute_command("atlas-clean", **kwargs)45    def package(self, **kwargs):46        self._execute_command("atlas-package", **kwargs)47    def list_packages(self, **kwargs):48        self._execute_command("atlas-mvn dependency:build-classpath -e -DincludeArtifactIds=lombok", **kwargs)...

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.