Suspend stdout/stderr capturing when emitting live logging messages

Author: nicoddemus

_pytest/logging.py

@@ -340,7 +340,8 @@ def _setup_cli_logging(self):
         """
         terminal_reporter = self._config.pluginmanager.get_plugin('terminalreporter')
         if self._config.getini('log_cli') and terminal_reporter is not None:
-            log_cli_handler = _LiveLoggingStreamHandler(terminal_reporter._tw)
+            capture_manager = self._config.pluginmanager.get_plugin('capturemanager')
+            log_cli_handler = _LiveLoggingStreamHandler(terminal_reporter, capture_manager)
             log_cli_format = get_option_ini(self._config, 'log_cli_format', 'log_format')
             log_cli_date_format = get_option_ini(self._config, 'log_cli_date_format', 'log_date_format')
             log_cli_formatter = logging.Formatter(log_cli_format, datefmt=log_cli_date_format)
@@ -356,12 +357,30 @@ class _LiveLoggingStreamHandler(logging.StreamHandler):
     """
     Custom StreamHandler used by the live logging feature: it will write a newline before the first log message
     in each test.
+
+    During live logging we must also explicitly disable stdout/stderr capturing otherwise it will get captured
+    and won't appear in the terminal.
     """
 
+    def __init__(self, terminal_reporter, capture_manager):
+        """
+        :param _pytest.terminal.TerminalReporter terminal_reporter:
+        :param _pytest.capture.CaptureManager capture_manager:
+        """
+        logging.StreamHandler.__init__(self, stream=terminal_reporter)
+        self.capture_manager = capture_manager
+        self._first_record_emitted = False
+
     def emit(self, record):
-        if not getattr(self, '_first_record_emitted', False):
-            self.stream.write('\n')
-            # we might consider adding a header at this point using self.stream.sep('-', 'live log') or something
-            # similar when we improve live logging output
-            self._first_record_emitted = True
-        logging.StreamHandler.emit(self, record)
+        if self.capture_manager is not None:
+            self.capture_manager.suspend_global_capture()
+        try:
+            if not self._first_record_emitted:
+                self.stream.write('\n')
+                # we might consider adding a header at this point using self.stream.section('live log', sep='-')
+                # or something similar when we improve live logging output
+                self._first_record_emitted = True
+            logging.StreamHandler.emit(self, record)
+        finally:
+            if self.capture_manager is not None:
+                self.capture_manager.resume_global_capture()

testing/logging/test_reporting.py

@@ -1,5 +1,8 @@
 # -*- coding: utf-8 -*-
 import os
+
+import six
+
 import pytest
 
 
@@ -410,3 +413,55 @@ def test_log_file(request):
         contents = rfh.read()
         assert "This log message will be shown" in contents
         assert "This log message won't be shown" not in contents
+
+
+@pytest.mark.parametrize('has_capture_manager', [True, False])
+def test_live_logging_suspends_capture(has_capture_manager, request):
+    """Test that capture manager is suspended when we emitting messages for live logging.
+
+    This tests the implementation calls instead of behavior because it is difficult/impossible to do it using
+    ``testdir`` facilities because they do their own capturing.
+
+    We parametrize the test to also make sure _LiveLoggingStreamHandler works correctly if no capture manager plugin
+    is installed.
+    """
+    import logging
+    from functools import partial
+    from _pytest.capture import CaptureManager
+    from _pytest.logging import _LiveLoggingStreamHandler
+
+    if six.PY2:
+        # need to use the 'generic' StringIO instead of io.StringIO because we might receive both bytes
+        # and unicode objects; io.StringIO only accepts unicode
+        from StringIO import StringIO
+    else:
+        from io import StringIO
+
+    class MockCaptureManager:
+        calls = []
+
+        def suspend_global_capture(self):
+            self.calls.append('suspend_global_capture')
+
+        def resume_global_capture(self):
+            self.calls.append('resume_global_capture')
+
+    # sanity check
+    assert CaptureManager.suspend_capture_item
+    assert CaptureManager.resume_global_capture
+
+    capture_manager = MockCaptureManager() if has_capture_manager else None
+    out_file = StringIO()
+
+    handler = _LiveLoggingStreamHandler(out_file, capture_manager)
+
+    logger = logging.getLogger(__file__ + '.test_live_logging_suspends_capture')
+    logger.addHandler(handler)
+    request.addfinalizer(partial(logger.removeHandler, handler))
+
+    logger.critical('some message')
+    if has_capture_manager:
+        assert MockCaptureManager.calls == ['suspend_global_capture', 'resume_global_capture']
+    else:
+        assert MockCaptureManager.calls == []
+    assert out_file.getvalue() == '\nsome message\n'