How to use get_status_log_path method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...370 test_start_time = int(time.time())371 if self.resultdir:372 os.chdir(self.resultdir)373 # touch status.log so that the parser knows a job is running here374 open(self.get_status_log_path(), 'a').close()375 self.enable_external_logging()376 collect_crashinfo = True377 temp_control_file_dir = None378 try:379 if install_before and machines:380 self._execute_code(INSTALL_CONTROL_FILE, namespace)381 if not only_collect_crashinfo:382 # determine the dir to write the control files to383 cfd_specified = (control_file_dir384 and control_file_dir is not self.USE_TEMP_DIR)385 if cfd_specified:386 temp_control_file_dir = None387 else:388 temp_control_file_dir = tempfile.mkdtemp(389 suffix='temp_control_file_dir')390 control_file_dir = temp_control_file_dir391 server_control_file = os.path.join(control_file_dir,392 SERVER_CONTROL_FILENAME)393 client_control_file = os.path.join(control_file_dir,394 CLIENT_CONTROL_FILENAME)395 if self.client:396 namespace['control'] = control397 utils.open_write_close(client_control_file, control)398 shutil.copyfile(CLIENT_WRAPPER_CONTROL_FILE,399 server_control_file)400 else:401 utils.open_write_close(server_control_file, control)402"Processing control file")403 self._execute_code(server_control_file, namespace)404"Finished processing control file")405 # no error occured, so we don't need to collect crashinfo406 collect_crashinfo = False407 finally:408 if temp_control_file_dir:409 # Clean up temp directory used for copies of the control files410 try:411 shutil.rmtree(temp_control_file_dir)412 except Exception, e:413 logging.warn('Could not remove temp directory %s: %s',414 temp_control_file_dir, e)415 if machines and (collect_crashdumps or collect_crashinfo):416 namespace['test_start_time'] = test_start_time417 if collect_crashinfo:418 # includes crashdumps419 self._execute_code(CRASHINFO_CONTROL_FILE, namespace)420 else:421 self._execute_code(CRASHDUMPS_CONTROL_FILE, namespace)422 if self.uncollected_log_file:423 os.remove(self.uncollected_log_file)424 self.disable_external_logging()425 if cleanup and machines:426 self._execute_code(CLEANUP_CONTROL_FILE, namespace)427 if install_after and machines:428 self._execute_code(INSTALL_CONTROL_FILE, namespace)429 def set_test_tag(self, tag=''):430 """Set tag to be added to test name of all following run_test steps."""431 self.test_tag = tag432 def run_test(self, url, *args, **dargs):433 """434 Summon a test object and run it.435 tag436 tag to add to testname437 url438 url of the test to run439 """440 (group, testname) = self.pkgmgr.get_package_name(url, 'test')441 tag = dargs.pop('tag', None)442 if tag is None:443 tag = self.test_tag444 if tag:445 testname += '.' + str(tag)446 subdir = testname447 outputdir = os.path.join(self.resultdir, subdir)448 if os.path.exists(outputdir):449 msg = ("%s already exists, test <%s> may have"450 " already run with tag <%s>" % (outputdir, testname, tag))451 raise error.TestError(msg)452 os.mkdir(outputdir)453 def group_func():454 try:455 test.runtest(self, url, tag, args, dargs)456 except error.TestBaseException, e:457 self.record(e.exit_status, subdir, testname, str(e))458 raise459 except Exception, e:460 info = str(e) + "\n" + traceback.format_exc()461 self.record('FAIL', subdir, testname, info)462 raise463 else:464 self.record('GOOD', subdir, testname, 'completed successfully')465 result, exc_info = self._run_group(testname, subdir, group_func)466 if exc_info and isinstance(exc_info[1], error.TestBaseException):467 return False468 elif exc_info:469 raise exc_info[0], exc_info[1], exc_info[2]470 else:471 return True472 def _run_group(self, name, subdir, function, *args, **dargs):473 """\474 Underlying method for running something inside of a group.475 """476 result, exc_info = None, None477 old_record_prefix = self.record_prefix478 try:479 self.record('START', subdir, name)480 self.record_prefix += '\t'481 try:482 result = function(*args, **dargs)483 finally:484 self.record_prefix = old_record_prefix485 except error.TestBaseException, e:486 self.record("END %s" % e.exit_status, subdir, name)487 exc_info = sys.exc_info()488 except Exception, e:489 err_msg = str(e) + '\n'490 err_msg += traceback.format_exc()491 self.record('END ABORT', subdir, name, err_msg)492 raise error.JobError(name + ' failed\n' + traceback.format_exc())493 else:494 self.record('END GOOD', subdir, name)495 return result, exc_info496 def run_group(self, function, *args, **dargs):497 """\498 function:499 subroutine to run500 *args:501 arguments for the function502 """503 name = function.__name__504 # Allow the tag for the group to be specified.505 tag = dargs.pop('tag', None)506 if tag:507 name = tag508 return self._run_group(name, None, function, *args, **dargs)[0]509 def run_reboot(self, reboot_func, get_kernel_func):510 """\511 A specialization of run_group meant specifically for handling512 a reboot. Includes support for capturing the kernel version513 after the reboot.514 reboot_func: a function that carries out the reboot515 get_kernel_func: a function that returns a string516 representing the kernel version.517 """518 old_record_prefix = self.record_prefix519 try:520 self.record('START', None, 'reboot')521 self.record_prefix += '\t'522 reboot_func()523 except Exception, e:524 self.record_prefix = old_record_prefix525 err_msg = str(e) + '\n' + traceback.format_exc()526 self.record('END FAIL', None, 'reboot', err_msg)527 raise528 else:529 kernel = get_kernel_func()530 self.record_prefix = old_record_prefix531 self.record('END GOOD', None, 'reboot',532 optional_fields={"kernel": kernel})533 def run_control(self, path):534 """Execute a control file found at path (relative to the autotest535 path). Intended for executing a control file within a control file,536 not for running the top-level job control file."""537 path = os.path.join(self.autodir, path)538 control_file = self._load_control_file(path)539, control_file_dir=self.USE_TEMP_DIR)540 def add_sysinfo_command(self, command, logfile=None, on_every_test=False):541 self._add_sysinfo_loggable(sysinfo.command(command, logf=logfile),542 on_every_test)543 def add_sysinfo_logfile(self, file, on_every_test=False):544 self._add_sysinfo_loggable(sysinfo.logfile(file), on_every_test)545 def _add_sysinfo_loggable(self, loggable, on_every_test):546 if on_every_test:547 self.sysinfo.test_loggables.add(loggable)548 else:549 self.sysinfo.boot_loggables.add(loggable)550 def record(self, status_code, subdir, operation, status='',551 optional_fields=None):552 """553 Record job-level status554 The intent is to make this file both machine parseable and555 human readable. That involves a little more complexity, but556 really isn't all that bad ;-)557 Format is <status code>\t<subdir>\t<operation>\t<status>558 status code: see common_lib.log.is_valid_status()559 for valid status definition560 subdir: MUST be a relevant subdirectory in the results,561 or None, which will be represented as '----'562 operation: description of what you ran (e.g. "dbench", or563 "mkfs -t foobar /dev/sda9")564 status: error message or "completed sucessfully"565 ------------------------------------------------------------566 Initial tabs indicate indent levels for grouping, and is567 governed by self.record_prefix568 multiline messages have secondary lines prefaced by a double569 space (' ')570 Executing this method will trigger the logging of all new571 warnings to date from the various console loggers.572 """573 # poll all our warning loggers for new warnings574 warnings = self._read_warnings()575 old_record_prefix = self.record_prefix576 try:577 if status_code.startswith("END "):578 self.record_prefix += "\t"579 for timestamp, msg in warnings:580 self._record("WARN", None, None, msg, timestamp)581 finally:582 self.record_prefix = old_record_prefix583 # write out the actual status log line584 self._record(status_code, subdir, operation, status,585 optional_fields=optional_fields)586 def _read_warnings(self):587 """Poll all the warning loggers and extract any new warnings that have588 been logged. If the warnings belong to a category that is currently589 disabled, this method will discard them and they will no longer be590 retrievable.591 Returns a list of (timestamp, message) tuples, where timestamp is an592 integer epoch timestamp."""593 warnings = []594 while True:595 # pull in a line of output from every logger that has596 # output ready to be read597 loggers, _, _ =, [], [], 0)598 closed_loggers = set()599 for logger in loggers:600 line = logger.readline()601 # record any broken pipes (aka line == empty)602 if len(line) == 0:603 closed_loggers.add(logger)604 continue605 # parse out the warning606 timestamp, msgtype, msg = line.split('\t', 2)607 timestamp = int(timestamp)608 # if the warning is valid, add it to the results609 if self.warning_manager.is_valid(timestamp, msgtype):610 warnings.append((timestamp, msg.strip()))611 # stop listening to loggers that are closed612 self.warning_loggers -= closed_loggers613 # stop if none of the loggers have any output left614 if not loggers:615 break616 # sort into timestamp order617 warnings.sort()618 return warnings619 def disable_warnings(self, warning_type):620 self.warning_manager.disable_warnings(warning_type)621 self.record("INFO", None, None,622 "disabling %s warnings" % warning_type,623 {"warnings.disable": warning_type})624 def enable_warnings(self, warning_type):625 self.warning_manager.enable_warnings(warning_type)626 self.record("INFO", None, None,627 "enabling %s warnings" % warning_type,628 {"warnings.enable": warning_type})629 def get_status_log_path(self, subdir=None):630 """Return the path to the job status log.631 @param subdir - Optional paramter indicating that you want the path632 to a subdirectory status log.633 @returns The path where the status log should be.634 """635 if self.resultdir:636 if subdir:637 return os.path.join(self.resultdir, subdir, "status.log")638 else:639 return os.path.join(self.resultdir, "status.log")640 else:641 return None642 def _update_uncollected_logs_list(self, update_func):643 """Updates the uncollected logs list in a multi-process safe manner.644 @param update_func - a function that updates the list of uncollected645 logs. Should take one parameter, the list to be updated.646 """647 if self.uncollected_log_file:648 log_file = open(self.uncollected_log_file, "r+")649 fcntl.flock(log_file, fcntl.LOCK_EX)650 try:651 uncollected_logs = pickle.load(log_file)652 update_func(uncollected_logs)653 log_file.truncate()655 pickle.dump(uncollected_logs, log_file)656 log_file.flush()657 finally:658 fcntl.flock(log_file, fcntl.LOCK_UN)659 log_file.close()660 def add_client_log(self, hostname, remote_path, local_path):661 """Adds a new set of client logs to the list of uncollected logs,662 to allow for future log recovery.663 @param host - the hostname of the machine holding the logs664 @param remote_path - the directory on the remote machine holding logs665 @param local_path - the local directory to copy the logs into666 """667 def update_func(logs_list):668 logs_list.append((hostname, remote_path, local_path))669 self._update_uncollected_logs_list(update_func)670 def remove_client_log(self, hostname, remote_path, local_path):671 """Removes a set of client logs from the list of uncollected logs,672 to allow for future log recovery.673 @param host - the hostname of the machine holding the logs674 @param remote_path - the directory on the remote machine holding logs675 @param local_path - the local directory to copy the logs into676 """677 def update_func(logs_list):678 logs_list.remove((hostname, remote_path, local_path))679 self._update_uncollected_logs_list(update_func)680 def _render_record(self, status_code, subdir, operation, status='',681 epoch_time=None, record_prefix=None,682 optional_fields=None):683 """684 Internal Function to generate a record to be written into a685 status log. For use by server_job.* classes only.686 """687 if subdir:688 if re.match(r'[\n\t]', subdir):689 raise ValueError('Invalid character in subdir string')690 substr = subdir691 else:692 substr = '----'693 if not log.is_valid_status(status_code):694 raise ValueError('Invalid status code supplied: %s' % status_code)695 if not operation:696 operation = '----'697 if re.match(r'[\n\t]', operation):698 raise ValueError('Invalid character in operation string')699 operation = operation.rstrip()700 status = status.rstrip()701 status = re.sub(r"\t", " ", status)702 # Ensure any continuation lines are marked so we can703 # detect them in the status file to ensure it is parsable.704 status = re.sub(r"\n", "\n" + self.record_prefix + " ", status)705 if not optional_fields:706 optional_fields = {}707 # Generate timestamps for inclusion in the logs708 if epoch_time is None:709 epoch_time = int(time.time())710 local_time = time.localtime(epoch_time)711 optional_fields["timestamp"] = str(epoch_time)712 optional_fields["localtime"] = time.strftime("%b %d %H:%M:%S",713 local_time)714 fields = [status_code, substr, operation]715 fields += ["%s=%s" % x for x in optional_fields.iteritems()]716 fields.append(status)717 if record_prefix is None:718 record_prefix = self.record_prefix719 msg = '\t'.join(str(x) for x in fields)720 return record_prefix + msg + '\n'721 def _record_prerendered(self, msg):722 """723 Record a pre-rendered msg into the status logs. The only724 change this makes to the message is to add on the local725 indentation. Should not be called outside of server_job.*726 classes. Unlike _record, this does not write the message727 to standard output.728 """729 lines = []730 status_file = self.get_status_log_path()731 status_log = open(status_file, 'a')732 for line in msg.splitlines():733 line = self.record_prefix + line + '\n'734 lines.append(line)735 status_log.write(line)736 status_log.close()737 self.__parse_status(lines)738 def _fill_server_control_namespace(self, namespace, protect=True):739 """740 Prepare a namespace to be used when executing server control files.741 This sets up the control file API by importing modules and making them742 available under the appropriate names within namespace.743 For use by _execute_code().744 Args:745 namespace: The namespace dictionary to fill in.746 protect: Boolean. If True (the default) any operation that would747 clobber an existing entry in namespace will cause an error.748 Raises:749 error.AutoservError: When a name would be clobbered by import.750 """751 def _import_names(module_name, names=()):752 """753 Import a module and assign named attributes into namespace.754 Args:755 module_name: The string module name.756 names: A limiting list of names to import from module_name. If757 empty (the default), all names are imported from the module758 similar to a "from import *" statement.759 Raises:760 error.AutoservError: When a name being imported would clobber761 a name already in namespace.762 """763 module = __import__(module_name, {}, {}, names)764 # No names supplied? Import * from the lowest level module.765 # (Ugh, why do I have to implement this part myself?)766 if not names:767 for submodule_name in module_name.split('.')[1:]:768 module = getattr(module, submodule_name)769 if hasattr(module, '__all__'):770 names = getattr(module, '__all__')771 else:772 names = dir(module)773 # Install each name into namespace, checking to make sure it774 # doesn't override anything that already exists.775 for name in names:776 # Check for conflicts to help prevent future problems.777 if name in namespace and protect:778 if namespace[name] is not getattr(module, name):779 raise error.AutoservError('importing name '780 '%s from %s %r would override %r' %781 (name, module_name, getattr(module, name),782 namespace[name]))783 else:784 # Encourage cleanliness and the use of __all__ for a785 # more concrete API with less surprises on '*' imports.786 warnings.warn('%s (%r) being imported from %s for use '787 'in server control files is not the '788 'first occurrance of that import.' %789 (name, namespace[name], module_name))790 namespace[name] = getattr(module, name)791 # This is the equivalent of prepending a bunch of import statements to792 # the front of the control script.793 namespace.update(os=os, sys=sys, logging=logging)794 _import_names('autotest_lib.server',795 ('hosts', 'autotest', 'kvm', 'git', 'standalone_profiler',796 'source_kernel', 'rpm_kernel', 'deb_kernel', 'git_kernel'))797 _import_names('autotest_lib.server.subcommand',798 ('parallel', 'parallel_simple', 'subcommand'))799 _import_names('autotest_lib.server.utils',800 ('run', 'get_tmp_dir', 'sh_escape', 'parse_machine'))801 _import_names('autotest_lib.client.common_lib.error')802 _import_names('autotest_lib.client.common_lib.barrier', ('barrier',))803 # Inject ourself as the job object into other classes within the API.804 # (Yuck, this injection is a gross thing be part of a public API. -gps)805 #806 # XXX Base & SiteAutotest do not appear to use .job. Who does?807 namespace['autotest'].Autotest.job = self808 # server.hosts.base_classes.Host uses .job.809 namespace['hosts'].Host.job = self810 def _execute_code(self, code_file, namespace, protect=True):811 """812 Execute code using a copy of namespace as a server control script.813 Unless protect_namespace is explicitly set to False, the dict will not814 be modified.815 Args:816 code_file: The filename of the control file to execute.817 namespace: A dict containing names to make available during execution.818 protect: Boolean. If True (the default) a copy of the namespace dict819 is used during execution to prevent the code from modifying its820 contents outside of this function. If False the raw dict is821 passed in and modifications will be allowed.822 """823 if protect:824 namespace = namespace.copy()825 self._fill_server_control_namespace(namespace, protect=protect)826 # TODO: Simplify and get rid of the special cases for only 1 machine.827 if len(self.machines) > 1:828 machines_text = '\n'.join(self.machines) + '\n'829 # Only rewrite the file if it does not match our machine list.830 try:831 machines_f = open(MACHINES_FILENAME, 'r')832 existing_machines_text = machines_f.close()834 except EnvironmentError:835 existing_machines_text = None836 if machines_text != existing_machines_text:837 utils.open_write_close(MACHINES_FILENAME, machines_text)838 execfile(code_file, namespace, namespace)839 def _record(self, status_code, subdir, operation, status='',840 epoch_time=None, optional_fields=None):841 """842 Actual function for recording a single line into the status843 logs. Should never be called directly, only by job.record as844 this would bypass the console monitor logging.845 """846 msg = self._render_record(status_code, subdir, operation, status,847 epoch_time, optional_fields=optional_fields)848 status_file = self.get_status_log_path()849 sys.stdout.write(msg)850 if status_file:851 open(status_file, "a").write(msg)852 if subdir:853 sub_status_file = self.get_status_log_path(subdir)854 open(sub_status_file, "a").write(msg)855 self.__parse_status(msg.splitlines())856 def __parse_status(self, new_lines):857 if not self.using_parser:858 return859 new_tests = self.parser.process_lines(new_lines)860 for test in new_tests:861 self.__insert_test(test)862 def __insert_test(self, test):863 """864 An internal method to insert a new test result into the865 database. This method will not raise an exception, even if an866 error occurs during the insert, to avoid failing a test867 simply because of unexpected database issues."""...

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?