# Source code for nose2.plugins.logcapture

    """
    Capture log messages during test execution, appending them to the
    error reports of failed tests.

    This plugin implements :func:`startTestRun`, :func:`startTest`,
    :func:`stopTest`, :func:`setTestOutcome`, and :func:`outcomeDetail` to
    set up a logging configuration that captures log messages during test
    execution, and appends them to error reports for tests that fail or
    raise exceptions.

    """

    import logging
    import threading
    from logging.handlers import BufferingHandler

    from nose2.events import Plugin
    from nose2.util import ln, parse_log_level

    log = logging.getLogger(__name__)
    __unittest = True

    class LogCapture(Plugin):
        """Capture log messages during test execution"""

        configSection = "log-capture"
        commandLineSwitch = (None, "log-capture", "Enable log capture")
        logformat = "%(name)s: %(levelname)s: %(message)s"
        logdatefmt = None
        clear = False
        filters = ["-nose"]

        def __init__(self) -> None:
            self.logformat = self.config.as_str("format", self.logformat)
            self.logdatefmt = self.config.as_str("date-format", self.logdatefmt)
            self.filters = self.config.as_list("filter", self.filters)
            self.clear = self.config.as_bool("clear-handlers", self.clear)
            self.loglevel = parse_log_level(self.config.as_str("log-level", "NOTSET"))
            self.handler = MyMemoryHandler(
                1000, self.logformat, self.logdatefmt, self.filters
            )

        def registerInSubprocess(self, event):
            event.pluginClasses.append(self.__class__)

        def startSubprocess(self, event):
            self._setupLoghandler()

        def startTestRun(self, event):
            """Set up logging handler"""
            self._setupLoghandler()

        def startTest(self, event):
            """Set up handler for new test"""
            self._setupLoghandler()

        def setTestOutcome(self, event):
            """Store captured log messages in ``event.metadata``"""
            self._addCapturedLogs(event)

        def stopTest(self, event):
            """Clear captured messages, ready for next test"""
            self.handler.truncate()

        def outcomeDetail(self, event):
            """Append captured log messages to ``event.extraDetail``"""
            logs = event.outcomeEvent.metadata.get("logs", None)
            if logs:
                event.extraDetail.append(ln(">> begin captured logging <<"))
                event.extraDetail.extend(logs)
                event.extraDetail.append(ln(">> end captured logging <<"))

        def _setupLoghandler(self):
            # setup our handler with root logger
            root_logger = logging.getLogger()
            if self.clear:
                if hasattr(root_logger, "handlers"):
                    for handler in root_logger.handlers:
                        root_logger.removeHandler(handler)
                for logger in logging.Logger.manager.loggerDict.values():
                    if hasattr(logger, "handlers"):
                        for handler in logger.handlers:
                            logger.removeHandler(handler)
            # make sure there isn't one already
            # you can't simply use "if self.handler not in root_logger.handlers"
            # since at least in unit tests this doesn't work --
            # LogCapture() is instantiated for each test case while root_logger
            # is module global
            # so we always add new MyMemoryHandler instance
            for handler in root_logger.handlers[:]:
                if isinstance(handler, MyMemoryHandler):
                    root_logger.handlers.remove(handler)
            root_logger.addHandler(self.handler)
            root_logger.setLevel(self.loglevel)

        def _addCapturedLogs(self, event):
            format = self.handler.format
            records = [format(r) for r in self.handler.buffer]
            if "logs" in event.metadata:
                event.metadata["logs"].extend(records)
            else:
                event.metadata["logs"] = records

    class FilterSet:
        def __init__(self, filter_components) -> None:
            self.inclusive, self.exclusive = self._partition(filter_components)

        @staticmethod
        def _partition(components):
            inclusive, exclusive = [], []
            for component in components:
                if component.startswith("-"):
                    exclusive.append(component[1:])
                else:
                    inclusive.append(component)
            return inclusive, exclusive

        def allow(self, record):
            """returns whether this record should be printed"""
            if not self:
                # nothing to filter
                return True
            return self._allow(record) and not self._deny(record)

        @staticmethod
        def _any_match(matchers, record):
            """return the bool of whether `record` starts with
            any item in `matchers`"""

            def record_matches_key(key):
                return record == key or record.startswith(key + ".")

            return any(map(record_matches_key, matchers))

        def _allow(self, record):
            if not self.inclusive:
                return True
            return self._any_match(self.inclusive, record)

        def _deny(self, record):
            if not self.exclusive:
                return False
            return self._any_match(self.exclusive, record)

    class MyMemoryHandler(BufferingHandler):
        def __init__(self, capacity, logformat, logdatefmt, filters) -> None:
            BufferingHandler.__init__(self, capacity)
            fmt = logging.Formatter(logformat, logdatefmt)
            self.setFormatter(fmt)
            self.filterset = FilterSet(filters)

        def flush(self):
            pass  # do nothing

        def truncate(self):
            self.buffer = []

        def filter(self, record):
            return self.filterset.allow(record.name)

        def emit(self, record):
            # take a snapshot of the potentially mutable arguments
            record.msg = record.getMessage()
            record.args = {}
            BufferingHandler.emit(self, record)

        def __getstate__(self):
            state = self.__dict__.copy()
            del state["lock"]
            return state

        def __setstate__(self, state):
            self.__dict__.update(state)
            self.lock = threading.RLock()