How to use _load_control_file method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...353 def _get_status_logger(self):354 """Return a reference to the status logger."""355 return self._logger356 @staticmethod357 def _load_control_file(path):358 f = open(path)359 try:360 control_file = finally:362 f.close()363 return re.sub('\r', '', control_file)364 def _register_subcommand_hooks(self):365 """366 Register some hooks into the subcommand modules that allow us367 to properly clean up self.hosts created in forked subprocesses.368 """369 def on_fork(cmd):370 self._existing_hosts_on_fork = set(self.hosts)371 def on_join(cmd):372 new_hosts = self.hosts - self._existing_hosts_on_fork373 for host in new_hosts:374 host.close()375 subcommand.subcommand.register_fork_hook(on_fork)376 subcommand.subcommand.register_join_hook(on_join)377 # TODO add a kwargs parameter.378 def _make_namespace(self):379 """Create a namespace dictionary to be passed along to control file.380 Creates a namespace argument populated with standard values:381 machines, job, ssh_user, ssh_port, ssh_pass, ssh_verbosity_flag,382 and ssh_options.383 """384 namespace = {'machines' : self.machine_dict_list,385 'job' : self,386 'ssh_user' : self._ssh_user,387 'ssh_port' : self._ssh_port,388 'ssh_pass' : self._ssh_pass,389 'ssh_verbosity_flag' : self._ssh_verbosity_flag,390 'ssh_options' : self._ssh_options}391 return namespace392 def cleanup(self, labels):393 """Cleanup machines.394 @param labels: Comma separated job labels, will be used to395 determine special task actions.396 """397 if not self.machines:398 raise error.AutoservError('No machines specified to cleanup')399 if self.resultdir:400 os.chdir(self.resultdir)401 namespace = self._make_namespace()402 namespace.update({'job_labels': labels, 'args': ''})403 self._execute_code(CLEANUP_CONTROL_FILE, namespace, protect=False)404 def verify(self, labels):405 """Verify machines are all ssh-able.406 @param labels: Comma separated job labels, will be used to407 determine special task actions.408 """409 if not self.machines:410 raise error.AutoservError('No machines specified to verify')411 if self.resultdir:412 os.chdir(self.resultdir)413 namespace = self._make_namespace()414 namespace.update({'job_labels': labels, 'args': ''})415 self._execute_code(VERIFY_CONTROL_FILE, namespace, protect=False)416 def reset(self, labels):417 """Reset machines by first cleanup then verify each machine.418 @param labels: Comma separated job labels, will be used to419 determine special task actions.420 """421 if not self.machines:422 raise error.AutoservError('No machines specified to reset.')423 if self.resultdir:424 os.chdir(self.resultdir)425 namespace = self._make_namespace()426 namespace.update({'job_labels': labels, 'args': ''})427 self._execute_code(RESET_CONTROL_FILE, namespace, protect=False)428 def repair(self, labels):429 """Repair machines.430 @param labels: Comma separated job labels, will be used to431 determine special task actions.432 """433 if not self.machines:434 raise error.AutoservError('No machines specified to repair')435 if self.resultdir:436 os.chdir(self.resultdir)437 namespace = self._make_namespace()438 namespace.update({'job_labels': labels, 'args': ''})439 self._execute_code(REPAIR_CONTROL_FILE, namespace, protect=False)440 def provision(self, labels):441 """442 Provision all hosts to match |labels|.443 @param labels: A comma seperated string of labels to provision the444 host to.445 """446 control = self._load_control_file(PROVISION_CONTROL_FILE)447, job_labels=labels)448 def precheck(self):449 """450 perform any additional checks in derived classes.451 """452 pass453 def enable_external_logging(self):454 """455 Start or restart external logging mechanism.456 """457 pass458 def disable_external_logging(self):459 """460 Pause or stop external logging mechanism.461 """462 pass463 def use_external_logging(self):464 """465 Return True if external logging should be used.466 """467 return False468 def _make_parallel_wrapper(self, function, machines, log):469 """Wrap function as appropriate for calling by parallel_simple."""470 # machines could be a list of dictionaries, e.g.,471 # [{'host_attributes': {}, 'hostname': ''}]472 # The dictionary is generated in server_job.__init__, refer to473 # variable machine_dict_list, then passed in with namespace, see method474 # server_job._make_namespace.475 # To compare the machinese to self.machines, which is a list of machine476 # hostname, we need to convert machines back to a list of hostnames.477 if (machines and isinstance(machines, list)478 and isinstance(machines[0], dict)):479 machines = [m['hostname'] for m in machines]480 if len(machines) > 1 and log:481 def wrapper(machine):482 hostname = server_utils.get_hostname_from_machine(machine)483 self.push_execution_context(hostname)484 os.chdir(self.resultdir)485 machine_data = {'hostname' : hostname,486 'status_version' : str(self._STATUS_VERSION)}487 utils.write_keyval(self.resultdir, machine_data)488 result = function(machine)489 return result490 else:491 wrapper = function492 return wrapper493 def parallel_simple(self, function, machines, log=True, timeout=None,494 return_results=False):495 """496 Run 'function' using parallel_simple, with an extra wrapper to handle497 the necessary setup for continuous parsing, if possible. If continuous498 parsing is already properly initialized then this should just work.499 @param function: A callable to run in parallel given each machine.500 @param machines: A list of machine names to be passed one per subcommand501 invocation of function.502 @param log: If True, output will be written to output in a subdirectory503 named after each machine.504 @param timeout: Seconds after which the function call should timeout.505 @param return_results: If True instead of an AutoServError being raised506 on any error a list of the results|exceptions from the function507 called on each arg is returned. [default: False]508 @raises error.AutotestError: If any of the functions failed.509 """510 wrapper = self._make_parallel_wrapper(function, machines, log)511 return subcommand.parallel_simple(512 wrapper, machines,513 subdir_name_constructor=server_utils.get_hostname_from_machine,514 log=log, timeout=timeout, return_results=return_results)515 def parallel_on_machines(self, function, machines, timeout=None):516 """517 @param function: Called in parallel with one machine as its argument.518 @param machines: A list of machines to call function(machine) on.519 @param timeout: Seconds after which the function call should timeout.520 @returns A list of machines on which function(machine) returned521 without raising an exception.522 """523 results = self.parallel_simple(function, machines, timeout=timeout,524 return_results=True)525 success_machines = []526 for result, machine in itertools.izip(results, machines):527 if not isinstance(result, Exception):528 success_machines.append(machine)529 return success_machines530 def record_skipped_test(self, skipped_test, message=None):531 """Insert a failure record into status.log for this test."""532 msg = message533 if msg is None:534 msg = 'No valid machines found for test %s.' % skipped_test535 self.record('START', None, skipped_test.test_name)537 self.record('INFO', None, skipped_test.test_name, msg)538 self.record('END TEST_NA', None, skipped_test.test_name, msg)539 def _has_failed_tests(self):540 """Parse status log for failed tests.541 This checks the current working directory and is intended only for use542 by the run() method.543 @return boolean544 """545 path = os.getcwd()546 # TODO(ayatane): Copied from tko/ Needs extensive refactor to547 # make code reuse plausible.548 job_keyval = tko_models.job.read_keyval(path)549 status_version = job_keyval.get("status_version", 0)550 # parse out the job551 parser = parser_lib.parser(status_version)552 job = parser.make_job(path)553 status_log = os.path.join(path, "status.log")554 if not os.path.exists(status_log):555 status_log = os.path.join(path, "status")556 if not os.path.exists(status_log):557 logging.warning("! Unable to parse job, no status file")558 return True559 # parse the status logs560 status_lines = open(status_log).readlines()561 parser.start(job)562 tests = parser.end(status_lines)563 # parser.end can return the same object multiple times, so filter out564 # dups565 job.tests = []566 already_added = set()567 for test in tests:568 if test not in already_added:569 already_added.add(test)570 job.tests.append(test)571 failed = False572 for test in job.tests:573 # The current job is still running and shouldn't count as failed.574 # The parser will fail to parse the exit status of the job since it575 # hasn't exited yet (this running right now is the job).576 failed = failed or (test.status != 'GOOD'577 and not _is_current_server_job(test))578 return failed579 def _collect_crashes(self, namespace, collect_crashinfo):580 """Collect crashes.581 @param namespace: namespace dict.582 @param collect_crashinfo: whether to collect crashinfo in addition to583 dumps584 """585 if collect_crashinfo:586 # includes crashdumps587 crash_control_file = CRASHINFO_CONTROL_FILE588 else:589 crash_control_file = CRASHDUMPS_CONTROL_FILE590 self._execute_code(crash_control_file, namespace)591 _USE_TEMP_DIR = object()592 def run(self, collect_crashdumps=True, namespace={}, control=None,593 control_file_dir=None, verify_job_repo_url=False,594 only_collect_crashinfo=False, skip_crash_collection=False,595 job_labels='', use_packaging=True):596 # for a normal job, make sure the uncollected logs file exists597 # for a crashinfo-only run it should already exist, bail out otherwise598 created_uncollected_logs = False599"I am PID %s", os.getpid())600 if self.resultdir and not os.path.exists(self._uncollected_log_file):601 if only_collect_crashinfo:602 # if this is a crashinfo-only run, and there were no existing603 # uncollected logs, just bail out early604"No existing uncollected logs, "605 "skipping crashinfo collection")606 return607 else:608 log_file = open(self._uncollected_log_file, "w")609 pickle.dump([], log_file)610 log_file.close()611 created_uncollected_logs = True612 # use a copy so changes don't affect the original dictionary613 namespace = namespace.copy()614 machines = self.machines615 if control is None:616 if self.control is None:617 control = ''618 elif self._use_client_trampoline:619 control = self._load_control_file(620 CLIENT_TRAMPOLINE_CONTROL_FILE)621 # repr of a string is safe for eval.622 control = (('trampoline_testname = %r\n' % str(self.control))623 + control)624 else:625 control = self._load_control_file(self.control)626 if control_file_dir is None:627 control_file_dir = self.resultdir628 self.aborted = False629 namespace.update(self._make_namespace())630 namespace.update({631 'args': self.args,632 'job_labels': job_labels,633 'gtest_runner': site_gtest_runner.gtest_runner(),634 })635 test_start_time = int(time.time())636 if self.resultdir:637 os.chdir(self.resultdir)638 # touch status.log so that the parser knows a job is running here639 open(self.get_status_log_path(), 'a').close()640 self.enable_external_logging()641 collect_crashinfo = True642 temp_control_file_dir = None643 try:644 try:645 if not with metrics.SecondsTimer(647 'chromeos/autotest/job/get_network_stats',648 fields = {'stage': 'start'}):649 namespace['network_stats_label'] = 'at-start'650 self._execute_code(GET_NETWORK_STATS_CONTROL_FILE,651 namespace)652 if only_collect_crashinfo:653 return654 # If the verify_job_repo_url option is set but we're unable655 # to actually verify that the job_repo_url contains the autotest656 # package, this job will fail.657 if verify_job_repo_url:658 self._execute_code(VERIFY_JOB_REPO_URL_CONTROL_FILE,659 namespace)660 else:661 logging.warning('Not checking if job_repo_url contains '662 'autotest packages on %s', machines)663 # determine the dir to write the control files to664 cfd_specified = (control_file_dir665 and control_file_dir is not self._USE_TEMP_DIR)666 if cfd_specified:667 temp_control_file_dir = None668 else:669 temp_control_file_dir = tempfile.mkdtemp(670 suffix='temp_control_file_dir')671 control_file_dir = temp_control_file_dir672 server_control_file = os.path.join(control_file_dir,673 self._control_filename)674 client_control_file = os.path.join(control_file_dir,675 CLIENT_CONTROL_FILENAME)676 if self._client:677 namespace['control'] = control678 utils.open_write_close(client_control_file, control)679 shutil.copyfile(CLIENT_WRAPPER_CONTROL_FILE,680 server_control_file)681 else:682 utils.open_write_close(server_control_file, control)683"Processing control file")684 namespace['use_packaging'] = use_packaging685 self._execute_code(server_control_file, namespace)686"Finished processing control file")687 # If no device error occured, no need to collect crashinfo.688 collect_crashinfo = self.failed_with_device_error689 except Exception as e:690 try:691 logging.exception(692 'Exception escaped control file, job aborting:')693 reason = re.sub(base_job.status_log_entry.BAD_CHAR_REGEX,694 ' ', str(e))695 self.record('INFO', None, None, str(e),696 {'job_abort_reason': reason})697 except:698 pass # don't let logging exceptions here interfere699 raise700 finally:701 if temp_control_file_dir:702 # Clean up temp directory used for copies of the control files703 try:704 shutil.rmtree(temp_control_file_dir)705 except Exception as e:706 logging.warning('Could not remove temp directory %s: %s',707 temp_control_file_dir, e)708 if machines and (collect_crashdumps or collect_crashinfo):709 if skip_crash_collection or'Skipping crash dump/info collection '711 'as requested.')712 else:713 with metrics.SecondsTimer(714 'chromeos/autotest/job/collect_crashinfo'):715 namespace['test_start_time'] = test_start_time716 # Remove crash files for passing tests.717 # TODO(ayatane): Tests that create crash files should be718 # reported.719 namespace['has_failed_tests'] = self._has_failed_tests()720 self._collect_crashes(namespace, collect_crashinfo)721 self.disable_external_logging()722 if self._uncollected_log_file and created_uncollected_logs:723 os.remove(self._uncollected_log_file)724 if not with metrics.SecondsTimer(726 'chromeos/autotest/job/get_network_stats',727 fields = {'stage': 'end'}):728 namespace['network_stats_label'] = 'at-end'729 self._execute_code(GET_NETWORK_STATS_CONTROL_FILE,730 namespace)731 def run_test(self, url, *args, **dargs):732 """733 Summon a test object and run it.734 tag735 tag to add to testname736 url737 url of the test to run738 """739 if self._disable_sysinfo:740 dargs['disable_sysinfo'] = True741 group, testname = self.pkgmgr.get_package_name(url, 'test')742 testname, subdir, tag = self._build_tagged_test_name(testname, dargs)743 outputdir = self._make_test_outputdir(subdir)744 def group_func():745 try:746 test.runtest(self, url, tag, args, dargs)747 except error.TestBaseException as e:748 self.record(e.exit_status, subdir, testname, str(e))749 raise750 except Exception as e:751 info = str(e) + "\n" + traceback.format_exc()752 self.record('FAIL', subdir, testname, info)753 raise754 else:755 self.record('GOOD', subdir, testname, 'completed successfully')756 try:757 result = self._run_group(testname, subdir, group_func)758 except error.TestBaseException as e:759 return False760 else:761 return True762 def _run_group(self, name, subdir, function, *args, **dargs):763 """Underlying method for running something inside of a group."""764 result, exc_info = None, None765 try:766 self.record('START', subdir, name)767 result = function(*args, **dargs)768 except error.TestBaseException as e:769 self.record("END %s" % e.exit_status, subdir, name)770 raise771 except Exception as e:772 err_msg = str(e) + '\n'773 err_msg += traceback.format_exc()774 self.record('END ABORT', subdir, name, err_msg)775 raise error.JobError(name + ' failed\n' + traceback.format_exc())776 else:777 self.record('END GOOD', subdir, name)778 finally:779 for hook in self._post_run_hooks:780 hook()781 return result782 def run_group(self, function, *args, **dargs):783 """\784 @param function: subroutine to run785 @returns: (result, exc_info). When the call succeeds, result contains786 the return value of |function| and exc_info is None. If787 |function| raises an exception, exc_info contains the tuple788 returned by sys.exc_info(), and result is None.789 """790 name = function.__name__791 # Allow the tag for the group to be specified.792 tag = dargs.pop('tag', None)793 if tag:794 name = tag795 try:796 result = self._run_group(name, None, function, *args, **dargs)[0]797 except error.TestBaseException:798 return None, sys.exc_info()799 return result, None800 def run_op(self, op, op_func, get_kernel_func):801 """\802 A specialization of run_group meant specifically for handling803 management operation. Includes support for capturing the kernel version804 after the operation.805 Args:806 op: name of the operation.807 op_func: a function that carries out the operation (reboot, suspend)808 get_kernel_func: a function that returns a string809 representing the kernel version.810 """811 try:812 self.record('START', None, op)813 op_func()814 except Exception as e:815 err_msg = str(e) + '\n' + traceback.format_exc()816 self.record('END FAIL', None, op, err_msg)817 raise818 else:819 kernel = get_kernel_func()820 self.record('END GOOD', None, op,821 optional_fields={"kernel": kernel})822 def run_control(self, path):823 """Execute a control file found at path (relative to the autotest824 path). Intended for executing a control file within a control file,825 not for running the top-level job control file."""826 path = os.path.join(self.autodir, path)827 control_file = self._load_control_file(path)828, control_file_dir=self._USE_TEMP_DIR)829 def add_sysinfo_command(self, command, logfile=None, on_every_test=False):830 self._add_sysinfo_loggable(sysinfo.command(command, logf=logfile),831 on_every_test)832 def add_sysinfo_logfile(self, file, on_every_test=False):833 self._add_sysinfo_loggable(sysinfo.logfile(file), on_every_test)834 def _add_sysinfo_loggable(self, loggable, on_every_test):835 if on_every_test:836 self.sysinfo.test_loggables.add(loggable)837 else:838 self.sysinfo.boot_loggables.add(loggable)839 def _read_warnings(self):840 """Poll all the warning loggers and extract any new warnings that have841 been logged. If the warnings belong to a category that is currently...

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?