Merge pull request #7224 from bluetech/logging-simplifications

logging: some simplifications/cleanups

changelog/7224.breaking.rst

@@ -0,0 +1,4 @@
+The `item.catch_log_handler` and `item.catch_log_handlers` attributes, set by the
+logging plugin and never meant to be public , are no longer available.
+
+The deprecated ``--no-print-logs`` option is removed. Use ``--show-capture`` instead.

doc/en/deprecations.rst

@@ -37,10 +37,10 @@ a public API and may break in the future.
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 .. deprecated:: 5.4
+.. versionremoved:: 6.0
 
 
-Option ``--no-print-logs`` is deprecated and meant to be removed in a future release. If you use ``--no-print-logs``, please try out ``--show-capture`` and
+Option ``--no-print-logs`` is removed. If you used ``--no-print-logs``, please use ``--show-capture`` instead.
-provide feedback.
 
 ``--show-capture`` command-line option was added in ``pytest 3.5.0`` and allows to specify how to
 display captured output when tests fail: ``no``, ``stdout``, ``stderr``, ``log`` or ``all`` (the default).

src/_pytest/deprecated.py

@@ -54,11 +54,6 @@ JUNIT_XML_DEFAULT_FAMILY = PytestDeprecationWarning(
     "for more information."
 )
 
-NO_PRINT_LOGS = PytestDeprecationWarning(
-    "--no-print-logs is deprecated and scheduled for removal in pytest 6.0.\n"
-    "Please use --show-capture instead."
-)
-
 COLLECT_DIRECTORY_HOOK = PytestDeprecationWarning(
     "The pytest_collect_directory hook is not working.\n"
     "Please use collect_ignore in conftests or pytest_collection_modifyitems."

src/_pytest/logging.py

@@ -1,6 +1,8 @@
 """ Access and control log capturing. """
 import logging
+import os
 import re
+import sys
 from contextlib import contextmanager
 from io import StringIO
 from typing import AbstractSet
@@ -9,6 +11,7 @@ from typing import Generator
 from typing import List
 from typing import Mapping
 from typing import Optional
+from typing import Union
 
 import pytest
 from _pytest import nodes
@@ -17,10 +20,14 @@ from _pytest.config import _strtobool
 from _pytest.config import Config
 from _pytest.config import create_terminal_writer
 from _pytest.pathlib import Path
+from _pytest.store import StoreKey
+
 
 DEFAULT_LOG_FORMAT = "%(levelname)-8s %(name)s:%(filename)s:%(lineno)d %(message)s"
 DEFAULT_LOG_DATE_FORMAT = "%H:%M:%S"
 _ANSI_ESCAPE_SEQ = re.compile(r"\x1b\[[\d;]+m")
+catch_log_handler_key = StoreKey["LogCaptureHandler"]()
+catch_log_handlers_key = StoreKey[Dict[str, "LogCaptureHandler"]]()
 
 
 def _remove_ansi_escape_sequences(text):
@@ -183,15 +190,6 @@ def pytest_addoption(parser):
         )
         group.addoption(option, dest=dest, **kwargs)
 
-    add_option_ini(
-        "--no-print-logs",
-        dest="log_print",
-        action="store_const",
-        const=False,
-        default=True,
-        type="bool",
-        help="disable printing caught logs on failed tests.",
-    )
     add_option_ini(
         "--log-level",
         dest="log_level",
@@ -268,46 +266,47 @@ def pytest_addoption(parser):
     )
 
 
-@contextmanager
+# Not using @contextmanager for performance reasons.
-def catching_logs(handler, formatter=None, level=None):
+class catching_logs:
     """Context manager that prepares the whole logging machinery properly."""
+
+    __slots__ = ("handler", "level", "orig_level")
+
+    def __init__(self, handler, level=None):
+        self.handler = handler
+        self.level = level
+
+    def __enter__(self):
         root_logger = logging.getLogger()
+        if self.level is not None:
+            self.handler.setLevel(self.level)
+        root_logger.addHandler(self.handler)
+        if self.level is not None:
+            self.orig_level = root_logger.level
+            root_logger.setLevel(min(self.orig_level, self.level))
+        return self.handler
 
-    if formatter is not None:
+    def __exit__(self, type, value, traceback):
-        handler.setFormatter(formatter)
+        root_logger = logging.getLogger()
-    if level is not None:
+        if self.level is not None:
-        handler.setLevel(level)
+            root_logger.setLevel(self.orig_level)
-
+        root_logger.removeHandler(self.handler)
-    # Adding the same handler twice would confuse logging system.
-    # Just don't do that.
-    add_new_handler = handler not in root_logger.handlers
-
-    if add_new_handler:
-        root_logger.addHandler(handler)
-    if level is not None:
-        orig_level = root_logger.level
-        root_logger.setLevel(min(orig_level, level))
-    try:
-        yield handler
-    finally:
-        if level is not None:
-            root_logger.setLevel(orig_level)
-        if add_new_handler:
-            root_logger.removeHandler(handler)
 
 
 class LogCaptureHandler(logging.StreamHandler):
     """A logging handler that stores log records and the log text."""
 
+    stream = None  # type: StringIO
+
     def __init__(self) -> None:
         """Creates a new log handler."""
-        logging.StreamHandler.__init__(self, StringIO())
+        super().__init__(StringIO())
         self.records = []  # type: List[logging.LogRecord]
 
     def emit(self, record: logging.LogRecord) -> None:
         """Keep the log records in a list in addition to the log text."""
         self.records.append(record)
-        logging.StreamHandler.emit(self, record)
+        super().emit(record)
 
     def reset(self) -> None:
         self.records = []
@@ -317,7 +316,7 @@ class LogCaptureHandler(logging.StreamHandler):
 class LogCaptureFixture:
     """Provides access and control of log capturing."""
 
-    def __init__(self, item) -> None:
+    def __init__(self, item: nodes.Node) -> None:
         """Creates a new funcarg."""
         self._item = item
         # dict of log name -> log level
@@ -338,7 +337,7 @@ class LogCaptureFixture:
         """
         :rtype: LogCaptureHandler
         """
-        return self._item.catch_log_handler  # type: ignore[no-any-return]
+        return self._item._store[catch_log_handler_key]
 
     def get_records(self, when: str) -> List[logging.LogRecord]:
         """
@@ -352,9 +351,9 @@ class LogCaptureFixture:
 
         .. versionadded:: 3.4
         """
-        handler = self._item.catch_log_handlers.get(when)
+        handler = self._item._store[catch_log_handlers_key].get(when)
         if handler:
-            return handler.records  # type: ignore[no-any-return]
+            return handler.records
         else:
             return []
 
@@ -491,13 +490,7 @@ class LoggingPlugin:
         """
         self._config = config
 
-        self.print_logs = get_option_ini(config, "log_print")
+        # Report logging.
-        if not self.print_logs:
-            from _pytest.warnings import _issue_warning_captured
-            from _pytest.deprecated import NO_PRINT_LOGS
-
-            _issue_warning_captured(NO_PRINT_LOGS, self._config.hook, stacklevel=2)
-
         self.formatter = self._create_formatter(
             get_option_ini(config, "log_format"),
             get_option_ini(config, "log_date_format"),
@@ -505,33 +498,40 @@ class LoggingPlugin:
         )
         self.log_level = get_log_level_for_setting(config, "log_level")
 
+        # File logging.
         self.log_file_level = get_log_level_for_setting(config, "log_file_level")
-        self.log_file_format = get_option_ini(config, "log_file_format", "log_format")
+        log_file = get_option_ini(config, "log_file") or os.devnull
-        self.log_file_date_format = get_option_ini(
-            config, "log_file_date_format", "log_date_format"
-        )
-        self.log_file_formatter = logging.Formatter(
-            self.log_file_format, datefmt=self.log_file_date_format
-        )
-
-        log_file = get_option_ini(config, "log_file")
-        if log_file:
         self.log_file_handler = logging.FileHandler(
             log_file, mode="w", encoding="UTF-8"
-            )  # type: Optional[logging.FileHandler]
+        )
-            self.log_file_handler.setFormatter(self.log_file_formatter)
+        log_file_format = get_option_ini(config, "log_file_format", "log_format")
-        else:
+        log_file_date_format = get_option_ini(
-            self.log_file_handler = None
+            config, "log_file_date_format", "log_date_format"
-
+        )
-        self.log_cli_handler = None
+        log_file_formatter = logging.Formatter(
-
+            log_file_format, datefmt=log_file_date_format
-        self.live_logs_context = lambda: nullcontext()
+        )
-        # Note that the lambda for the live_logs_context is needed because
+        self.log_file_handler.setFormatter(log_file_formatter)
-        # live_logs_context can otherwise not be entered multiple times due
-        # to limitations of contextlib.contextmanager.
 
+        # CLI/live logging.
+        self.log_cli_level = get_log_level_for_setting(
+            config, "log_cli_level", "log_level"
+        )
         if self._log_cli_enabled():
-            self._setup_cli_logging()
+            terminal_reporter = config.pluginmanager.get_plugin("terminalreporter")
+            capture_manager = config.pluginmanager.get_plugin("capturemanager")
+            # if capturemanager plugin is disabled, live logging still works.
+            self.log_cli_handler = _LiveLoggingStreamHandler(
+                terminal_reporter, capture_manager
+            )  # type: Union[_LiveLoggingStreamHandler, _LiveLoggingNullHandler]
+        else:
+            self.log_cli_handler = _LiveLoggingNullHandler()
+        log_cli_formatter = self._create_formatter(
+            get_option_ini(config, "log_cli_format", "log_format"),
+            get_option_ini(config, "log_cli_date_format", "log_date_format"),
+            get_option_ini(config, "log_auto_indent"),
+        )
+        self.log_cli_handler.setFormatter(log_cli_formatter)
 
     def _create_formatter(self, log_format, log_date_format, auto_indent):
         # color option doesn't exist if terminal plugin is disabled
@@ -551,29 +551,6 @@ class LoggingPlugin:
 
         return formatter
 
-    def _setup_cli_logging(self):
-        config = self._config
-        terminal_reporter = config.pluginmanager.get_plugin("terminalreporter")
-        if terminal_reporter is None:
-            # terminal reporter is disabled e.g. by pytest-xdist.
-            return
-
-        capture_manager = config.pluginmanager.get_plugin("capturemanager")
-        # if capturemanager plugin is disabled, live logging still works.
-        log_cli_handler = _LiveLoggingStreamHandler(terminal_reporter, capture_manager)
-
-        log_cli_formatter = self._create_formatter(
-            get_option_ini(config, "log_cli_format", "log_format"),
-            get_option_ini(config, "log_cli_date_format", "log_date_format"),
-            get_option_ini(config, "log_auto_indent"),
-        )
-
-        log_cli_level = get_log_level_for_setting(config, "log_cli_level", "log_level")
-        self.log_cli_handler = log_cli_handler
-        self.live_logs_context = lambda: catching_logs(
-            log_cli_handler, formatter=log_cli_formatter, level=log_cli_level
-        )
-
     def set_log_path(self, fname):
         """Public method, which can set filename parameter for
         Logging.FileHandler(). Also creates parent directory if
@@ -590,136 +567,49 @@ class LoggingPlugin:
         if not fname.parent.exists():
             fname.parent.mkdir(exist_ok=True, parents=True)
 
-        self.log_file_handler = logging.FileHandler(
+        stream = fname.open(mode="w", encoding="UTF-8")
-            str(fname), mode="w", encoding="UTF-8"
+        if sys.version_info >= (3, 7):
-        )
+            old_stream = self.log_file_handler.setStream(stream)
-        self.log_file_handler.setFormatter(self.log_file_formatter)
+        else:
+            old_stream = self.log_file_handler.stream
+            self.log_file_handler.acquire()
+            try:
+                self.log_file_handler.flush()
+                self.log_file_handler.stream = stream
+            finally:
+                self.log_file_handler.release()
+        if old_stream:
+            old_stream.close()
 
     def _log_cli_enabled(self):
-        """Return True if log_cli should be considered enabled, either explicitly
+        """Return whether live logging is enabled."""
-        or because --log-cli-level was given in the command-line.
+        enabled = self._config.getoption(
-        """
-        return self._config.getoption(
             "--log-cli-level"
         ) is not None or self._config.getini("log_cli")
+        if not enabled:
+            return False
 
-    @pytest.hookimpl(hookwrapper=True, tryfirst=True)
+        terminal_reporter = self._config.pluginmanager.get_plugin("terminalreporter")
-    def pytest_collection(self) -> Generator[None, None, None]:
+        if terminal_reporter is None:
-        with self.live_logs_context():
+            # terminal reporter is disabled e.g. by pytest-xdist.
-            if self.log_cli_handler:
+            return False
-                self.log_cli_handler.set_when("collection")
 
-            if self.log_file_handler is not None:
+        return True
-                with catching_logs(self.log_file_handler, level=self.log_file_level):
-                    yield
-            else:
-                # Add a dummy handler to ensure logging.root.handlers is not empty.
-                # If it were empty, then a `logging.warning()` call (and similar) during collection
-                # would trigger a `logging.basicConfig()` call, which would add a `StreamHandler`
-                # handler, which would cause all subsequent logs which reach the root to be also
-                # printed to stdout, which we don't want (issue #6240).
-                with catching_logs(logging.NullHandler()):
-                    yield
-
-    @contextmanager
-    def _runtest_for(self, item, when):
-        with self._runtest_for_main(item, when):
-            if self.log_file_handler is not None:
-                with catching_logs(self.log_file_handler, level=self.log_file_level):
-                    yield
-            else:
-                yield
-
-    @contextmanager
-    def _runtest_for_main(
-        self, item: nodes.Item, when: str
-    ) -> Generator[None, None, None]:
-        """Implements the internals of pytest_runtest_xxx() hook."""
-        with catching_logs(
-            LogCaptureHandler(), formatter=self.formatter, level=self.log_level
-        ) as log_handler:
-            if self.log_cli_handler:
-                self.log_cli_handler.set_when(when)
-
-            if item is None:
-                yield  # run the test
-                return
-
-            if not hasattr(item, "catch_log_handlers"):
-                item.catch_log_handlers = {}  # type: ignore[attr-defined]
-            item.catch_log_handlers[when] = log_handler  # type: ignore[attr-defined]
-            item.catch_log_handler = log_handler  # type: ignore[attr-defined]
-            try:
-                yield  # run test
-            finally:
-                if when == "teardown":
-                    del item.catch_log_handler  # type: ignore[attr-defined]
-                    del item.catch_log_handlers  # type: ignore[attr-defined]
-
-            if self.print_logs:
-                # Add a captured log section to the report.
-                log = log_handler.stream.getvalue().strip()
-                item.add_report_section(when, "log", log)
-
-    @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_setup(self, item):
-        with self._runtest_for(item, "setup"):
-            yield
-
-    @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_call(self, item):
-        with self._runtest_for(item, "call"):
-            yield
-
-    @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_teardown(self, item):
-        with self._runtest_for(item, "teardown"):
-            yield
-
-    @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_logstart(self):
-        if self.log_cli_handler:
-            self.log_cli_handler.reset()
-        with self._runtest_for(None, "start"):
-            yield
-
-    @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_logfinish(self):
-        with self._runtest_for(None, "finish"):
-            yield
-
-    @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_logreport(self):
-        with self._runtest_for(None, "logreport"):
-            yield
-
-    @pytest.hookimpl(hookwrapper=True, tryfirst=True)
-    def pytest_sessionfinish(self):
-        with self.live_logs_context():
-            if self.log_cli_handler:
-                self.log_cli_handler.set_when("sessionfinish")
-            if self.log_file_handler is not None:
-                try:
-                    with catching_logs(
-                        self.log_file_handler, level=self.log_file_level
-                    ):
-                        yield
-                finally:
-                    # Close the FileHandler explicitly.
-                    # (logging.shutdown might have lost the weakref?!)
-                    self.log_file_handler.close()
-            else:
-                yield
 
     @pytest.hookimpl(hookwrapper=True, tryfirst=True)
     def pytest_sessionstart(self):
-        with self.live_logs_context():
-            if self.log_cli_handler:
         self.log_cli_handler.set_when("sessionstart")
-            if self.log_file_handler is not None:
+
+        with catching_logs(self.log_cli_handler, level=self.log_cli_level):
             with catching_logs(self.log_file_handler, level=self.log_file_level):
                 yield
-            else:
+
+    @pytest.hookimpl(hookwrapper=True, tryfirst=True)
+    def pytest_collection(self) -> Generator[None, None, None]:
+        self.log_cli_handler.set_when("collection")
+
+        with catching_logs(self.log_cli_handler, level=self.log_cli_level):
+            with catching_logs(self.log_file_handler, level=self.log_file_level):
                 yield
 
     @pytest.hookimpl(hookwrapper=True)
@@ -734,12 +624,71 @@ class LoggingPlugin:
             # setting verbose flag is needed to avoid messy test progress output
             self._config.option.verbose = 1
 
-        with self.live_logs_context():
+        with catching_logs(self.log_cli_handler, level=self.log_cli_level):
-            if self.log_file_handler is not None:
             with catching_logs(self.log_file_handler, level=self.log_file_level):
                 yield  # run all the tests
-            else:
+
-                yield  # run all the tests
+    @pytest.hookimpl
+    def pytest_runtest_logstart(self):
+        self.log_cli_handler.reset()
+        self.log_cli_handler.set_when("start")
+
+    @pytest.hookimpl
+    def pytest_runtest_logreport(self):
+        self.log_cli_handler.set_when("logreport")
+
+    def _runtest_for(self, item: nodes.Item, when: str) -> Generator[None, None, None]:
+        """Implements the internals of pytest_runtest_xxx() hook."""
+        log_handler = LogCaptureHandler()
+        log_handler.setFormatter(self.formatter)
+        with catching_logs(log_handler, level=self.log_level):
+            item._store[catch_log_handlers_key][when] = log_handler
+            item._store[catch_log_handler_key] = log_handler
+
+            yield
+
+            log = log_handler.stream.getvalue().strip()
+            item.add_report_section(when, "log", log)
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_setup(self, item):
+        self.log_cli_handler.set_when("setup")
+
+        empty = {}  # type: Dict[str, LogCaptureHandler]
+        item._store[catch_log_handlers_key] = empty
+        yield from self._runtest_for(item, "setup")
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_call(self, item):
+        self.log_cli_handler.set_when("call")
+
+        yield from self._runtest_for(item, "call")
+
+    @pytest.hookimpl(hookwrapper=True)
+    def pytest_runtest_teardown(self, item):
+        self.log_cli_handler.set_when("teardown")
+
+        yield from self._runtest_for(item, "teardown")
+        del item._store[catch_log_handlers_key]
+        del item._store[catch_log_handler_key]
+
+    @pytest.hookimpl
+    def pytest_runtest_logfinish(self):
+        self.log_cli_handler.set_when("finish")
+
+    @pytest.hookimpl(hookwrapper=True, tryfirst=True)
+    def pytest_sessionfinish(self):
+        self.log_cli_handler.set_when("sessionfinish")
+
+        with catching_logs(self.log_cli_handler, level=self.log_cli_level):
+            with catching_logs(self.log_file_handler, level=self.log_file_level):
+                yield
+
+    @pytest.hookimpl
+    def pytest_unconfigure(self):
+        # Close the FileHandler explicitly.
+        # (logging.shutdown might have lost the weakref?!)
+        self.log_file_handler.close()
 
 
 class _LiveLoggingStreamHandler(logging.StreamHandler):
@@ -790,4 +739,14 @@ class _LiveLoggingStreamHandler(logging.StreamHandler):
             if not self._section_name_shown and self._when:
                 self.stream.section("live log " + self._when, sep="-", bold=True)
                 self._section_name_shown = True
-            logging.StreamHandler.emit(self, record)
+            super().emit(record)
+
+
+class _LiveLoggingNullHandler(logging.NullHandler):
+    """A handler used when live logging is disabled."""
+
+    def reset(self):
+        pass
+
+    def set_when(self, when):
+        pass

testing/deprecated_test.py

@@ -117,47 +117,6 @@ def test_node_direct_ctor_warning():
     assert w[0].filename == __file__
 
 
-def assert_no_print_logs(testdir, args):
-    result = testdir.runpytest(*args)
-    result.stdout.fnmatch_lines(
-        [
-            "*--no-print-logs is deprecated and scheduled for removal in pytest 6.0*",
-            "*Please use --show-capture instead.*",
-        ]
-    )
-
-
-@pytest.mark.filterwarnings("default")
-def test_noprintlogs_is_deprecated_cmdline(testdir):
-    testdir.makepyfile(
-        """
-        def test_foo():
-            pass
-        """
-    )
-
-    assert_no_print_logs(testdir, ("--no-print-logs",))
-
-
-@pytest.mark.filterwarnings("default")
-def test_noprintlogs_is_deprecated_ini(testdir):
-    testdir.makeini(
-        """
-        [pytest]
-        log_print=False
-        """
-    )
-
-    testdir.makepyfile(
-        """
-        def test_foo():
-            pass
-        """
-    )
-
-    assert_no_print_logs(testdir, ())
-
-
 def test__fillfuncargs_is_deprecated() -> None:
     with pytest.warns(
         pytest.PytestDeprecationWarning,

testing/logging/test_fixture.py

@@ -1,6 +1,7 @@
 import logging
 
 import pytest
+from _pytest.logging import catch_log_handlers_key
 
 logger = logging.getLogger(__name__)
 sublogger = logging.getLogger(__name__ + ".baz")
@@ -136,4 +137,4 @@ def test_caplog_captures_for_all_stages(caplog, logging_during_setup_and_teardow
     assert [x.message for x in caplog.get_records("setup")] == ["a_setup_log"]
 
     # This reaches into private API, don't use this type of thing in real tests!
-    assert set(caplog._item.catch_log_handlers.keys()) == {"setup", "call"}
+    assert set(caplog._item._store[catch_log_handlers_key]) == {"setup", "call"}

testing/logging/test_reporting.py

@@ -166,60 +166,6 @@ def test_teardown_logging(testdir):
     )
 
 
-def test_disable_log_capturing(testdir):
-    testdir.makepyfile(
-        """
-        import sys
-        import logging
-
-        logger = logging.getLogger(__name__)
-
-        def test_foo():
-            sys.stdout.write('text going to stdout')
-            logger.warning('catch me if you can!')
-            sys.stderr.write('text going to stderr')
-            assert False
-        """
-    )
-    result = testdir.runpytest("--no-print-logs")
-    print(result.stdout)
-    assert result.ret == 1
-    result.stdout.fnmatch_lines(["*- Captured stdout call -*", "text going to stdout"])
-    result.stdout.fnmatch_lines(["*- Captured stderr call -*", "text going to stderr"])
-    with pytest.raises(pytest.fail.Exception):
-        result.stdout.fnmatch_lines(["*- Captured *log call -*"])
-
-
-def test_disable_log_capturing_ini(testdir):
-    testdir.makeini(
-        """
-        [pytest]
-        log_print=False
-        """
-    )
-    testdir.makepyfile(
-        """
-        import sys
-        import logging
-
-        logger = logging.getLogger(__name__)
-
-        def test_foo():
-            sys.stdout.write('text going to stdout')
-            logger.warning('catch me if you can!')
-            sys.stderr.write('text going to stderr')
-            assert False
-        """
-    )
-    result = testdir.runpytest()
-    print(result.stdout)
-    assert result.ret == 1
-    result.stdout.fnmatch_lines(["*- Captured stdout call -*", "text going to stdout"])
-    result.stdout.fnmatch_lines(["*- Captured stderr call -*", "text going to stderr"])
-    with pytest.raises(pytest.fail.Exception):
-        result.stdout.fnmatch_lines(["*- Captured *log call -*"])
-
-
 @pytest.mark.parametrize("enabled", [True, False])
 def test_log_cli_enabled_disabled(testdir, enabled):
     msg = "critical message logged by test"