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

How to use step_engine method in autotest

Best Python code snippet using autotest_python

job.py

Source:job.py

...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)871        if state:872            self.sysinfo.deserialize(state)873    def _save_sysinfo_state(self):874        state = self.sysinfo.serialize()875        self._state.set('client', 'sysinfo', state)876class disk_usage_monitor:877    def __init__(self, logging_func, device, max_mb_per_hour):878        self.func = logging_func879        self.device = device880        self.max_mb_per_hour = max_mb_per_hour881    def start(self):882        self.initial_space = utils.freespace(self.device)883        self.start_time = time.time()884    def stop(self):885        # if no maximum usage rate was set, we don't need to886        # generate any warnings887        if not self.max_mb_per_hour:888            return889        final_space = utils.freespace(self.device)890        used_space = self.initial_space - final_space891        stop_time = time.time()892        total_time = stop_time - self.start_time893        # round up the time to one minute, to keep extremely short894        # tests from generating false positives due to short, badly895        # timed bursts of activity896        total_time = max(total_time, 60.0)897        # determine the usage rate898        bytes_per_sec = used_space / total_time899        mb_per_sec = bytes_per_sec / 1024**2900        mb_per_hour = mb_per_sec * 60 * 60901        if mb_per_hour > self.max_mb_per_hour:902            msg = ("disk space on %s was consumed at a rate of %.2f MB/hour")903            msg %= (self.device, mb_per_hour)904            self.func(msg)905    @classmethod906    def watch(cls, *monitor_args, **monitor_dargs):907        """ Generic decorator to wrap a function call with the908        standard create-monitor -> start -> call -> stop idiom."""909        def decorator(func):910            def watched_func(*args, **dargs):911                monitor = cls(*monitor_args, **monitor_dargs)912                monitor.start()913                try:914                    func(*args, **dargs)915                finally:916                    monitor.stop()917            return watched_func918        return decorator919def runjob(control, drop_caches, options):920    """921    Run a job using the given control file.922    This is the main interface to this module.923    @see base_job.__init__ for parameter info.924    """925    control = os.path.abspath(control)926    state = control + '.state'927    # Ensure state file is cleaned up before the job starts to run if autotest928    # is not running with the --continue flag929    if not options.cont and os.path.isfile(state):930        logging.debug('Cleaning up previously found state file')931        os.remove(state)932    # instantiate the job object ready for the control file.933    myjob = None934    try:935        # Check that the control file is valid936        if not os.path.exists(control):937            raise error.JobError(control + ": control file not found")938        # When continuing, the job is complete when there is no939        # state file, ensure we don't try and continue.940        if options.cont and not os.path.exists(state):941            raise error.JobComplete("all done")942        myjob = job(control=control, drop_caches=drop_caches, options=options)943        # Load in the users control file, may do any one of:944        #  1) execute in toto945        #  2) define steps, and select the first via next_step()946        myjob.step_engine()947    except error.JobContinue:948        sys.exit(5)949    except error.JobComplete:950        sys.exit(1)951    except error.JobError, instance:952        logging.error("JOB ERROR: " + str(instance))953        if myjob:954            command = None955            if len(instance.args) > 1:956                command = instance.args[1]957                myjob.record('ABORT', None, command, str(instance))958            myjob.record('END ABORT', None, None, str(instance))959            assert myjob._record_indent == 0960            myjob.complete(1)...

game.py

Source:game.py

1from curses import wrapper2import locale3import game.story.welcome as welcome4import game.story.end as end5import game.story.map_factory as map_factory6import game.character.creator as character_creator7import game.engine.wait as wait8import game.character.character as character9import game.progression.ability as ability10import game.engine.step_engine as step_engine11def play_step(step, hero, stdscr):12    stdscr.refresh()13    if step and step["type"] != "empty":14        step_engine.forType(step.get("type"))(step, hero, stdscr)15def start():16    welcome.show()17    map = map_factory.generate()18    hero = character_creator.new()19    wrapper(loop, map, hero)20    end.show(win=False)21def loop(stdscr, map, hero):22    locale.setlocale(locale.LC_ALL, "")23    map_window = stdscr.derwin(15, 25, 1, 1)24    map_window.keypad(True)25    character_window = stdscr.derwin(15, 60, 1, 28)26    text_window = stdscr.derwin(15, 60, 16, 28)27    stdscr.vline(1, 27, "|", 15)28    map.render(map_window)29    while hero.is_alive():30        hero.print_character(character_window)31        text_window.refresh()32        character_window.refresh()33        direction = map_window.getch()34        text_window.clear()35        character_window.clear()36        map_window.clear()37        step = map.move_hero(direction)38        map.render(map_window)...

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.