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

How to use _run_step_fn method in autotest

Best Python code snippet using autotest_python

job.py

Source:job.py

...754        steps = self._state.get('client', 'steps')755        steps.insert(0, self.__create_step_tuple(fn, args, dargs))756        self._next_step_index += 1757        self._state.set('client', 'steps', steps)758    def _run_step_fn(self, local_vars, fn, args, dargs):759        """Run a (step) function within the given context"""760        local_vars['__args'] = args761        local_vars['__dargs'] = dargs762        try:763            exec('__ret = %s(*__args, **__dargs)' % fn, local_vars, local_vars)764            return local_vars['__ret']765        except SystemExit:766            raise  # Send error.JobContinue and JobComplete on up to runjob.767        except error.TestNAError, detail:768            self.record(detail.exit_status, None, fn, str(detail))769        except Exception, detail:770            raise error.UnhandledJobError(detail)771    def _create_frame(self, global_vars, ancestry, fn_name):772        """Set up the environment like it would have been when this773        function was first defined.774        Child step engine 'implementations' must have 'return locals()'775        at end end of their steps.  Because of this, we can call the776        parent function and get back all child functions (i.e. those777        defined within it).778        Unfortunately, the call stack of the function calling779        job.next_step might have been deeper than the function it780        added.  In order to make sure that the environment is what it781        should be, we need to then pop off the frames we built until782        we find the frame where the function was first defined."""783        # The copies ensure that the parent frames are not modified784        # while building child frames.  This matters if we then785        # pop some frames in the next part of this function.786        current_frame = copy.copy(global_vars)787        frames = [current_frame]788        for steps_fn_name in ancestry:789            ret = self._run_step_fn(current_frame, steps_fn_name, [], {})790            current_frame = copy.copy(ret)791            frames.append(current_frame)792        # Walk up the stack frames until we find the place fn_name was defined.793        while len(frames) > 2:794            if fn_name not in frames[-2]:795                break796            if frames[-2][fn_name] != frames[-1][fn_name]:797                break798            frames.pop()799            ancestry.pop()800        return (frames[-1], ancestry)801    def _add_step_init(self, local_vars, current_function):802        """If the function returned a dictionary that includes a803        function named 'step_init', prepend it to our list of steps.804        This will only get run the first time a function with a nested805        use of the step engine is run."""806        if (isinstance(local_vars, dict) and807            'step_init' in local_vars and808            callable(local_vars['step_init'])):809            # The init step is a child of the function810            # we were just running.811            self._current_step_ancestry.append(current_function)812            self.next_step_prepend('step_init')813    def step_engine(self):814        """The multi-run engine used when the control file defines step_init.815        Does the next step.816        """817        # Set up the environment and then interpret the control file.818        # Some control files will have code outside of functions,819        # which means we need to have our state engine initialized820        # before reading in the file.821        global_control_vars = {'job': self,822                               'args': self.args}823        exec(JOB_PREAMBLE, global_control_vars, global_control_vars)824        try:825            execfile(self.control, global_control_vars, global_control_vars)826        except error.TestNAError, detail:827            self.record(detail.exit_status, None, self.control, str(detail))828        except SystemExit:829            raise  # Send error.JobContinue and JobComplete on up to runjob.830        except Exception, detail:831            # Syntax errors or other general Python exceptions coming out of832            # the top level of the control file itself go through here.833            raise error.UnhandledJobError(detail)834        # If we loaded in a mid-job state file, then we presumably835        # know what steps we have yet to run.836        if not self._is_continuation:837            if 'step_init' in global_control_vars:838                self.next_step(global_control_vars['step_init'])839        else:840            # if last job failed due to unexpected reboot, record it as fail841            # so harness gets called842            last_job = self._state.get('client', 'unexpected_reboot', None)843            if last_job:844                subdir, testname = last_job845                self.record('FAIL', subdir, testname, 'unexpected reboot')846                self.record('END FAIL', subdir, testname)847        # Iterate through the steps.  If we reboot, we'll simply848        # continue iterating on the next step.849        while len(self._state.get('client', 'steps')) > 0:850            steps = self._state.get('client', 'steps')851            (ancestry, fn_name, args, dargs) = steps.pop(0)852            self._state.set('client', 'steps', steps)853            self._next_step_index = 0854            ret = self._create_frame(global_control_vars, ancestry, fn_name)855            local_vars, self._current_step_ancestry = ret856            local_vars = self._run_step_fn(local_vars, fn_name, args, dargs)857            self._add_step_init(local_vars, fn_name)858    def add_sysinfo_command(self, command, logfile=None, on_every_test=False):859        self._add_sysinfo_loggable(sysinfo.command(command, logf=logfile),860                                   on_every_test)861    def add_sysinfo_logfile(self, file, on_every_test=False):862        self._add_sysinfo_loggable(sysinfo.logfile(file), on_every_test)863    def _add_sysinfo_loggable(self, loggable, on_every_test):864        if on_every_test:865            self.sysinfo.test_loggables.add(loggable)866        else:867            self.sysinfo.boot_loggables.add(loggable)868        self._save_sysinfo_state()869    def _load_sysinfo_state(self):870        state = self._state.get('client', 'sysinfo', None)...

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.