Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _flush_all_streams method in autotest

Best Python code snippet using autotest_python

logging_manager.py

Source:logging_manager.py

...282    def _push_context(self, context):283        for stream_manager in self._streams:284            stream_manager.on_push_context(context)285        self._context_stack.append(context)286    def _flush_all_streams(self):287        for stream_manager in self._streams:288            stream_manager.flush()289    def _add_log_handlers(self, add_handlers_fn):290        """291        Modify the logging module's registered handlers and push a new context292        onto the stack.293        :param add_handlers_fn: function to modify the registered logging294        handlers. Accepts a context dictionary which may be modified.295        """296        self._flush_all_streams()297        context = self._get_context()298        add_handlers_fn(context)299        self._push_context(context)300    class _TaggingFormatter(logging.Formatter):301        """302        Delegates to a given formatter, but prefixes each line of output with a303        tag.304        """305        def __init__(self, base_formatter, tag):306            self.base_formatter = base_formatter307            prefix = tag + ' : '308            self._fmt = base_formatter._fmt.replace('%(message)s',309                                                    prefix + '%(message)s')310            self.datefmt = base_formatter.datefmt311    def _add_tagging_formatter(self, tag):312        for handler in _current_handlers():313            tagging_formatter = self._TaggingFormatter(handler.formatter, tag)314            handler.setFormatter(tagging_formatter)315    def _do_redirect(self, stream=None, filename=None, level=None,316                     clear_other_handlers=False):317        """318        :param clear_other_handlers - if true, clear out all other logging319        handlers.320        """321        assert bool(stream) != bool(filename)  # xor322        if not self._started:323            raise RuntimeError('You must call start_logging() before this')324        def add_handler(context):325            if clear_other_handlers:326                self._clear_all_handlers()327            if stream:328                handler = self.logging_config_object.add_stream_handler(stream)329            else:330                handler = self.logging_config_object.add_file_handler(filename)331            if level:332                handler.setLevel(level)333        self._add_log_handlers(add_handler)334    def redirect(self, filename):335        """Redirect output to the specified file"""336        self._do_redirect(filename=filename, clear_other_handlers=True)337    def redirect_to_stream(self, stream):338        """Redirect output to the given stream"""339        self._do_redirect(stream=stream, clear_other_handlers=True)340    def tee_redirect(self, filename, level=None):341        """Tee output to the specified file"""342        self._do_redirect(filename=filename, level=level)343    def tee_redirect_to_stream(self, stream):344        """Tee output to the given stream"""345        self._do_redirect(stream=stream)346    def tee_redirect_debug_dir(self, debug_dir, log_name=None, tag=None):347        """348        Tee output to a full new set of debug logs in the given directory.349        """350        def add_handlers(context):351            if tag:352                self._add_tagging_formatter(tag)353                context['tag_added'] = True354            self.logging_config_object.add_debug_file_handlers(355                debug_dir, log_name=log_name)356        self._add_log_handlers(add_handlers)357    def _restore_context(self, context):358        for stream_handler in self._streams:359            stream_handler.on_restore_context(context)360        # restore logging handlers361        old_handlers = context['old_handlers']362        for handler in _current_handlers() - old_handlers:363            handler.close()364        self._clear_all_handlers()365        for handler in old_handlers:366            logger.addHandler(handler)367        if 'tag_added' in context:368            for handler in _current_handlers():369                tagging_formatter = handler.formatter370                handler.setFormatter(tagging_formatter.base_formatter)371    def _pop_context(self):372        self._flush_all_streams()373        context = self._context_stack.pop()374        self._restore_context(context)375    def undo_redirect(self):376        """377        Undo the last redirection (that hasn't yet been undone).378        If any subprocesses have been launched since the redirection was379        performed, they must have ended by the time this is called.  Otherwise,380        this will hang waiting for the logging subprocess to end.381        """382        if not self._context_stack:383            raise RuntimeError('No redirects to undo')384        self._pop_context()385    def restore(self):386        """...

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.