Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use __stop_job_logging method in avocado

Best Python code snippet using avocado_python
job.py
Source:job.py
...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'...
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.