How to use __stop_job_logging method in avocado

Best Python code snippet using avocado_python

job.py

Source:job.py Github

copy

Full Screen

...191 logging.DEBUG,192 fmt="%(message)s")193 root_logger.addHandler(test_handler)194 self.__logging_handlers[test_handler] = ["avocado.test", ""]195 def __stop_job_logging(self):196 if self._stdout_stderr:197 sys.stdout, sys.stderr = self._stdout_stderr198 for handler, loggers in self.__logging_handlers.iteritems():199 for logger in loggers:200 logging.getLogger(logger).removeHandler(handler)201 def _update_latest_link(self):202 """203 Update the latest job result symbolic link [avocado-logs-dir]/latest.204 """205 def soft_abort(msg):206 """ Only log the problem """207 logging.getLogger("avocado.test").warning("Unable to update the "208 "latest link: %s" % msg)209 basedir = os.path.dirname(self.logdir)210 basename = os.path.basename(self.logdir)211 proc_latest = os.path.join(basedir, "latest.%s" % os.getpid())212 latest = os.path.join(basedir, "latest")213 if os.path.exists(latest) and not os.path.islink(latest):214 soft_abort('"%s" already exists and is not a symlink' % latest)215 return216 if os.path.exists(proc_latest):217 try:218 os.unlink(proc_latest)219 except OSError, details:220 soft_abort("Unable to remove %s: %s" % (proc_latest, details))221 return222 try:223 os.symlink(basename, proc_latest)224 os.rename(proc_latest, latest)225 except OSError, details:226 soft_abort("Unable to create create latest symlink: %s" % details)227 return228 finally:229 if os.path.exists(proc_latest):230 os.unlink(proc_latest)231 def _start_sysinfo(self):232 if hasattr(self.args, 'sysinfo'):233 if self.args.sysinfo == 'on':234 sysinfo_dir = path.init_dir(self.logdir, 'sysinfo')235 self.sysinfo = sysinfo.SysInfo(basedir=sysinfo_dir)236 def _make_test_runner(self):237 if hasattr(self.args, 'test_runner'):238 test_runner_class = self.args.test_runner239 else:240 test_runner_class = runner.TestRunner241 self.test_runner = test_runner_class(job=self,242 result=self.result)243 def _make_test_suite(self, references=None):244 """245 Prepares a test suite to be used for running tests246 :param references: String with tests references to be resolved, and then247 run, separated by whitespace. Optionally, a248 list of tests (each test a string).249 :returns: a test suite (a list of test factories)250 """251 loader.loader.load_plugins(self.args)252 try:253 suite = loader.loader.discover(references)254 except loader.LoaderUnhandledReferenceError as details:255 raise exceptions.OptionValidationError(details)256 except KeyboardInterrupt:257 raise exceptions.JobError('Command interrupted by user...')258 if not getattr(self.args, "dry_run", False):259 return suite260 for i in xrange(len(suite)):261 suite[i] = [test.DryRunTest, suite[i][1]]262 return suite263 def _log_job_id(self):264 job_log = _TEST_LOGGER265 job_log.info('Job ID: %s', self.unique_id)266 if self.replay_sourcejob is not None:267 job_log.info('Replay of Job ID: %s', self.replay_sourcejob)268 job_log.info('')269 @staticmethod270 def _log_cmdline():271 job_log = _TEST_LOGGER272 cmdline = " ".join(sys.argv)273 job_log.info("Command line: %s", cmdline)274 job_log.info('')275 @staticmethod276 def _log_avocado_version():277 job_log = _TEST_LOGGER278 job_log.info('CloudTest Version: %s', version.VERSION)279 if os.path.exists('.git') and os.path.exists('avocado.spec'):280 cmd = "git show --summary --pretty='%H' | head -1"281 status, top_commit = commands.getstatusoutput(cmd)282 cmd2 = "git rev-parse --abbrev-ref HEAD"283 status2, branch = commands.getstatusoutput(cmd2)284 # Let's display information only if git is installed285 # (commands succeed).286 if status == 0 and status2 == 0:287 job_log.info('Avocado git repo info')288 job_log.info("Top commit: %s", top_commit)289 job_log.info("Branch: %s", branch)290 job_log.info('')291 @staticmethod292 def _log_avocado_config():293 job_log = _TEST_LOGGER294 job_log.info('Config files read (in order):')295 for cfg_path in settings.config_paths:296 job_log.info(cfg_path)297 if settings.config_paths_failed:298 job_log.info('Config files failed to read (in order):')299 for cfg_path in settings.config_paths_failed:300 job_log.info(cfg_path)301 job_log.info('')302 job_log.info('Avocado config:')303 header = ('Section.Key', 'Value')304 config_matrix = []305 for section in settings.config.sections():306 for value in settings.config.items(section):307 config_key = ".".join((section, value[0]))308 config_matrix.append([config_key, value[1]])309 for line in astring.iter_tabular_output(config_matrix, header):310 job_log.info(line)311 job_log.info('')312 @staticmethod313 def _log_avocado_datadir():314 job_log = _TEST_LOGGER315 job_log.info('Avocado Data Directories:')316 job_log.info('')317 job_log.info("Avocado replaces config dirs that can't be accessed")318 job_log.info('with sensible defaults. Please edit your local config')319 job_log.info('file to customize values')320 job_log.info('')321 job_log.info('base ' + data_dir.get_base_dir())322 job_log.info('tests ' + data_dir.get_test_dir())323 job_log.info('data ' + data_dir.get_data_dir())324 job_log.info('logs ' + data_dir.get_logs_dir())325 job_log.info('')326 def _log_mux_tree(self, mux):327 job_log = _TEST_LOGGER328 tree_repr = tree.tree_view(mux.variants.root, verbose=True,329 use_utf8=False)330 if tree_repr:331 job_log.info('Multiplex tree representation:')332 for line in tree_repr.splitlines():333 job_log.info(line)334 job_log.info('')335 def _log_tmp_dir(self):336 job_log = _TEST_LOGGER337 job_log.info('Temporary dir: %s', data_dir.get_tmp_dir())338 job_log.info('')339 def _log_mux_variants(self, mux):340 job_log = _TEST_LOGGER341 for (index, tpl) in enumerate(mux.variants):342 paths = ', '.join([x.path for x in tpl])343 job_log.info('Variant %s: %s', index + 1, paths)344 if mux.variants:345 job_log.info('')346 def _log_job_debug_info(self, mux):347 """348 Log relevant debug information to the job log.349 """350 self._log_cmdline()351 self._log_avocado_version()352 self._log_avocado_config()353 self._log_avocado_datadir()354 self._log_mux_tree(mux)355 self._log_tmp_dir()356 self._log_mux_variants(mux)357 self._log_job_id()358 def create_test_suite(self):359 """360 Creates the test suite for this Job361 This is a public Job API as part of the documented Job phases362 """363 try:364 self.test_suite = self._make_test_suite(self.references)365 self.result.tests_total = len(self.test_suite)366 except loader.LoaderError as details:367 stacktrace.log_exc_info(sys.exc_info(), 'avocado.app.debug')368 raise exceptions.OptionValidationError(details)369 if not self.test_suite:370 if self.references:371 references = " ".join(self.references)372 e_msg = ("No tests found for given test references, try "373 "'avocado list -V %s' for details" % references)374 else:375 e_msg = ("No test references provided nor any other arguments "376 "resolved into tests. Please double check the executed"377 " command.")378 raise exceptions.OptionValidationError(e_msg)379 def pre_tests(self):380 """381 Run the pre tests execution hooks382 By default this runs the plugins that implement the383 :class:`avocado.core.plugin_interfaces.JobPre` interface.384 """385 self._job_pre_post_dispatcher = dispatcher.JobPrePostDispatcher()386 output.log_plugin_failures(self._job_pre_post_dispatcher.load_failures)387 self._job_pre_post_dispatcher.map_method('pre', self)388 self._result_events_dispatcher.map_method('pre_tests', self)389 def run_tests(self):390 mux = getattr(self.args, "mux", None)391 if mux is None:392 mux = multiplexer.Mux()393 if not mux.is_parsed(): # Mux not yet parsed, apply args394 try:395 mux.parse(self.args)396 except (IOError, ValueError) as details:397 raise exceptions.OptionValidationError("Unable to parse mux: "398 "%s" % details)399 self._make_test_runner()400 self._start_sysinfo()401 self._log_job_debug_info(mux)402 jobdata.record(self.args, self.logdir, mux, self.references, sys.argv)403 replay_map = getattr(self.args, 'replay_map', None)404 summary = self.test_runner.run_suite(self.test_suite,405 mux,406 self.timeout,407 replay_map)408 # If it's all good so far, set job status to 'PASS'409 if self.status == 'RUNNING':410 self.status = 'PASS'411 _TEST_LOGGER.info('Test results available in %s', self.logdir)412 if summary is None:413 self.exitcode |= exit_codes.AVOCADO_JOB_FAIL414 return self.exitcode415 if 'INTERRUPTED' in summary:416 self.exitcode |= exit_codes.AVOCADO_JOB_INTERRUPTED417 if 'FAIL' in summary:418 self.exitcode |= exit_codes.AVOCADO_TESTS_FAIL419 return self.exitcode420 def post_tests(self):421 """422 Run the post tests execution hooks423 By default this runs the plugins that implement the424 :class:`avocado.core.plugin_interfaces.JobPost` interface.425 """426 if self._job_pre_post_dispatcher is None:427 self._job_pre_post_dispatcher = dispatcher.JobPrePostDispatcher()428 output.log_plugin_failures(self._job_pre_post_dispatcher.load_failures)429 self._job_pre_post_dispatcher.map_method('post', self)430 def run(self):431 """432 Runs all job phases, returning the test execution results.433 This method is supposed to be the simplified interface for434 jobs, that is, they run all phases of a job.435 :return: Integer with overall job status. See436 :mod:`avocado.core.exit_codes` for more information.437 """438 runtime.CURRENT_JOB = self439 try:440 self.create_test_suite()441 self.pre_tests()442 return self.run_tests()443 except exceptions.JobBaseException as details:444 self.status = details.status445 fail_class = details.__class__.__name__446 self.log.error('\nAvocado job failed: %s: %s', fail_class, details)447 self.exitcode |= exit_codes.AVOCADO_JOB_FAIL448 return self.exitcode449 except exceptions.OptionValidationError as details:450 self.log.error('\n' + str(details))451 self.exitcode |= exit_codes.AVOCADO_JOB_FAIL452 return self.exitcode453 except Exception as details:454 self.status = "ERROR"455 exc_type, exc_value, exc_traceback = sys.exc_info()456 tb_info = traceback.format_exception(exc_type, exc_value,457 exc_traceback.tb_next)458 fail_class = details.__class__.__name__459 self.log.error('\nAvocado crashed: %s: %s', fail_class, details)460 for line in tb_info:461 self.log.debug(line)462 self.log.error("Please include the traceback info and command line"463 " used on your bug report")464 self.log.error('Report bugs visiting %s', _NEW_ISSUE_LINK)465 self.exitcode |= exit_codes.AVOCADO_FAIL466 return self.exitcode467 finally:468 self.post_tests()469 if not settings.get_value('runner.behavior', 'keep_tmp_files',470 key_type=bool, default=False):471 data_dir.clean_tmp_files()472 self.__stop_job_logging()473class TestProgram(object):474 """475 Convenience class to make avocado test modules executable.476 """477 def __init__(self):478 # Avoid fork loop/bomb when running a test via avocado.main() that479 # calls avocado.main() itself480 if os.environ.get('AVOCADO_STANDALONE_IN_MAIN', False):481 sys.stderr.write('AVOCADO_STANDALONE_IN_MAIN environment variable '482 'found. This means that this code is being '483 'called recursively. Exiting to avoid an infinite'484 ' fork loop.\n')485 sys.exit(exit_codes.AVOCADO_FAIL)486 os.environ['AVOCADO_STANDALONE_IN_MAIN'] = 'True'...

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run avocado 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?

Helpful

NotHelpful