How to use step_engine method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...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)...

Full Screen

Full Screen Github


Full Screen

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 map = map_factory.generate()18 hero = wrapper(loop, map, hero)20 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)...

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?