WIP: DO NOT MERGE

README.rst

@@ -30,11 +30,11 @@ Running without options::
 
 Shows failed tests like so::
 
-    -------------------------- Captured log ---------------------------
+    ----------------------- Captured stdlog call ----------------------
     test_pytest_catchlog.py    26 INFO     text going to logger
-    ------------------------- Captured stdout -------------------------
+    ----------------------- Captured stdout call ----------------------
     text going to stdout
-    ------------------------- Captured stderr -------------------------
+    ----------------------- Captured stderr call ----------------------
     text going to stderr
     ==================== 2 failed in 0.02 seconds =====================
 
@@ -50,11 +50,11 @@ Running pytest specifying formatting options::
 
 Shows failed tests like so::
 
-    -------------------------- Captured log ---------------------------
+    ----------------------- Captured stdlog call ----------------------
     2010-04-10 14:48:44 INFO text going to logger
-    ------------------------- Captured stdout -------------------------
+    ----------------------- Captured stdout call ----------------------
     text going to stdout
-    ------------------------- Captured stderr -------------------------
+    ----------------------- Captured stderr call ----------------------
     text going to stderr
     ==================== 2 failed in 0.02 seconds =====================
 
@@ -77,39 +77,39 @@ completely with::
 
 Shows failed tests in the normal manner as no logs were captured::
 
-    ------------------------- Captured stdout -------------------------
+    ----------------------- Captured stdout call ----------------------
     text going to stdout
-    ------------------------- Captured stderr -------------------------
+    ----------------------- Captured stderr call ----------------------
     text going to stderr
     ==================== 2 failed in 0.02 seconds =====================
 
 Inside tests it is possible to change the log level for the captured
 log messages.  This is supported by the ``caplog`` funcarg::
 
     def test_foo(caplog):
-        caplog.setLevel(logging.INFO)
+        caplog.set_level(logging.INFO)
         pass
 
 By default the level is set on the handler used to catch the log
 messages, however as a convenience it is also possible to set the log
 level of any logger::
 
     def test_foo(caplog):
-        caplog.setLevel(logging.CRITICAL, logger='root.baz')
+        caplog.set_level(logging.CRITICAL, logger='root.baz')
         pass
 
 It is also possible to use a context manager to temporarily change the
 log level::
 
     def test_bar(caplog):
-        with caplog.atLevel(logging.INFO):
+        with caplog.at_level(logging.INFO):
             pass
 
 Again, by default the level of the handler is affected but the level
 of any logger can be changed instead with::
 
     def test_bar(caplog):
-        with caplog.atLevel(logging.CRITICAL, logger='root.baz'):
+        with caplog.at_level(logging.CRITICAL, logger='root.baz'):
             pass
 
 Lastly all the logs sent to the logger during the test run are made

pytest_catchlog.py

@@ -2,10 +2,15 @@
 from __future__ import absolute_import, division, print_function
 
 import logging
+import logging.handlers
+import operator
+import re
+from fnmatch import fnmatch
 from contextlib import closing, contextmanager
 
-import pytest
 import py
+import pytest
+from _pytest import pytester
 
 
 __version__ = '1.1'
@@ -37,11 +42,42 @@ def logging_using_handler(handler, logger=None):
     """Context manager that safely register a given handler."""
     logger = get_logger_obj(logger)
 
-    logger.addHandler(handler)
-    try:
-        yield logger
-    finally:
-        logger.removeHandler(handler)
+    if handler in logger.handlers:  # reentrancy
+        # Adding the same handler twice would confuse logging system.
+        # Just don't do that.
+        yield
+    else:
+        logger.addHandler(handler)
+        try:
+            yield
+        finally:
+            logger.removeHandler(handler)
+
+@contextmanager
+def catching_logs(handler, filter=None, formatter=None,
+                  level=logging.NOTSET, logger=None):
+    """Context manager that prepares the whole logging machinery properly."""
+    logger = get_logger_obj(logger)
+
+    if filter is not None:
+        handler.addFilter(filter)
+    if formatter is not None:
+        handler.setFormatter(formatter)
+    handler.setLevel(level)
+
+    with closing(handler), \
+            logging_using_handler(handler, logger), \
+            logging_at_level(min(handler.level, logger.level), logger):
+
+        yield handler
+
+
+def _to_lines(self, lines):
+    if isinstance(lines, py.io._basestring):
+        lines = py.code.Source(lines)
+    if isinstance(lines, py.code.Source):
+        lines = lines.strip().lines
+    return lines
 
 
 def add_option_ini(parser, option, dest, default=None, help=None):
@@ -96,31 +132,20 @@ def __init__(self, config):
                 get_option_ini(config, 'log_format'),
                 get_option_ini(config, 'log_date_format'))
 
-    @contextmanager
-    def catching_logs(self, handler=None):
-        """Context manager that prepares the logging machinery properly."""
-        if handler is None:
-            handler = CatchLogHandler()
-
-        with closing(handler), \
-                logging_using_handler(handler), \
-                logging_at_level(logging.NOTSET):
-
-            handler.setFormatter(self.formatter)
-            yield handler
-
     def _make_runtest_for(when):
         """Create a hook method for a given context (setup/call/teardown)."""
 
         @pytest.mark.hookwrapper
         def runtest_func(self, item):
             """Implements pytest_runtest_xxx() hook."""
-            with self.catching_logs() as log_handler:
+            with closing(py.io.TextIO()) as stream, \
+                    catching_logs(logging.StreamHandler(stream),
+                                  formatter=self.formatter):
                 yield  # run test
 
                 if self.print_logs:
                     # Add a captured log section to the report.
-                    log = log_handler.stream.getvalue().strip()
+                    log = stream.getvalue().strip()
                     item.add_report_section(when, 'log', log)
 
         runtest_func.__name__ = 'pytest_runtest_' + when  # just in case
@@ -133,56 +158,30 @@ def runtest_func(self, item):
     del _make_runtest_for
 
 
-class CatchLogHandler(logging.StreamHandler):
-    """A logging handler that captures the log into a log stream."""
-
-    def __init__(self):
-        """Creates a new log handler."""
-        super(CatchLogHandler, self).__init__(stream=py.io.TextIO())
-
-    def close(self):
-        """Close this log handler and its underlying stream."""
-        try:
-            try:
-                self.flush()
-            finally:
-                self.stream.close()
-        finally:
-            super(CatchLogHandler, self).close()
-
-
-class RecordingCatchLogHandler(CatchLogHandler):
+class RecordingHandler(logging.handlers.BufferingHandler):
     """A logging handler that stores log records and the log text."""
 
     def __init__(self):
-        """Creates a new log handler."""
-        super(RecordingCatchLogHandler, self).__init__()
-        self.records = []
-
-    def emit(self, record):
-        """Keep the log records in a list in addition to the log text."""
-        self.records.append(record)
-        super(RecordingCatchLogHandler, self).emit(record)
+        super(RecordingHandler, self).__init__(capacity=0)  # ignored
 
+    def shouldFlush(self, record):
+        return False
 
-class CatchLogFuncArg(object):
-    """Provides access and control of log capturing."""
 
-    def __init__(self, handler):
-        """Creates a new funcarg."""
+class CatchLogRec(pytester.LineMatcher):
 
-        self.handler = handler
+    _records_to_text = staticmethod(logging.BufferingFormatter().format)
 
+    @property
     def text(self):
         """Returns the log text."""
+        return self._records_to_text(self.buffer)
 
-        return self.handler.stream.getvalue()
-
+    @property
     def records(self):
-        """Returns the list of log records."""
-
-        return self.handler.records
+        return list(self.buffer)
 
+    @property
     def record_tuples(self):
         """Returns a list of a striped down version of log records intended
         for use in assertion comparison.
@@ -191,13 +190,26 @@ def record_tuples(self):
 
             (logger_name, log_level, message)
         """
-        return [(r.name, r.levelno, r.getMessage()) for r in self.records()]
+        return [(r.name, r.levelno, r.getMessage()) for r in self.buffer]
 
-    # Methods controlling a level of the internal log recorder of the funcarg.
+    @property
+    def buffer(self):
+        return self._handler.buffer
 
-    def set_level(self, level):
-        """Sets the level for recording of logs."""
-        self.handler.setLevel(level)
+    @property
+    def lines(self):
+        return [r.getMessage() for r in self.buffer]
+
+    def __init__(self, handler):
+        # Do not call LineMatcher.__init__(),
+        # it attempts to assign to self.lines, which is a property.
+        self._handler = handler
+
+    @classmethod
+    @contextmanager
+    def recording(cls, **kwargs):
+        with catching_logs(RecordingHandler(), **kwargs) as handler:
+            yield cls(handler)
 
     def at_level(self, level):
         """Context manager that sets the level for recording of logs.
@@ -206,12 +218,44 @@ def at_level(self, level):
         captured logs that are reported for failed tests use the
         ``logging_at_level()`` context manager.
         """
-        return logging_at_level(level, self.handler)  # duck typing: quack!
+        return logging_at_level(level, self._handler)  # duck typing: quack!
+
+    def set_level(self, level):
+        """Alias for the ``RecordingHandler.setLevel()`` method."""
+        self._handler.setLevel(level)
+
+
+class CatchLogFuncArg(CatchLogRec):
+    """Provides access and control of log capturing."""
+
+    @staticmethod
+    @contextmanager
+    def emits(level, lines, logger=None):
+        __tracebackhide__ = True
+
+        with CatchLogRec.recording(level=level, logger=logger) as rec:
+            yield
+            rec.fnmatch_lines(lines)
+
+    def _make_emits(level):
+        @classmethod
+        def emits_at_level(cls, **kwargs):
+            __tracebackhide__ = True
+            return cls.emits(level, **kwargs)
+        return emits_at_level
+
+    emits_debug    = _make_emits(logging.DEBUG)
+    emits_info     = _make_emits(logging.INFO)
+    emits_warning  = _make_emits(logging.WARNING)
+    emits_error    = _make_emits(logging.ERROR)
+    emits_critical = _make_emits(logging.CRITICAL)
+
+    del _make_emits
 
     # Helper methods controlling a level of the global logging.
 
     @staticmethod
-    def logging_set_level(level, logger=None):
+    def set_logging_level(level, logger=None):
         """Context manager that sets the level for the logging subsystem.
 
         By default, the level is set on the root logger aggregating all
@@ -234,10 +278,8 @@ def logging_at_level(level, logger=None):
 @pytest.yield_fixture
 def caplog(request):
     """Returns a funcarg to access and control log capturing."""
-    plugin = request.config.pluginmanager.getplugin('_catch_log')
-
-    with plugin.catching_logs(RecordingCatchLogHandler()) as log_handler:
-        yield CatchLogFuncArg(log_handler)
+    with CatchLogFuncArg.recording() as funcarg:
+        yield funcarg
 
 
 capturelog = caplog

test_pytest_catchlog.py

@@ -56,12 +56,12 @@ def test_change_level(testdir):
         pytest_plugins = 'catchlog'
 
         def test_foo(caplog):
-            caplog.set_level(logging.INFO)
+            caplog.set_logging_level(logging.INFO)
             log = logging.getLogger()
             log.debug('handler DEBUG level')
             log.info('handler INFO level')
 
-            caplog.set_level(logging.CRITICAL, logger='root.baz')
+            caplog.set_logging_level(logging.CRITICAL, logger='root.baz')
             log = logging.getLogger('root.baz')
             log.warning('logger WARNING level')
             log.critical('logger CRITICAL level')
@@ -89,13 +89,12 @@ def test_with_statement(testdir):
         pytest_plugins = 'catchlog'
 
         def test_foo(caplog):
-            with caplog.at_level(logging.INFO):
-                log = logging.getLogger()
+            with caplog.logging_at_level(logging.INFO) as log:
                 log.debug('handler DEBUG level')
                 log.info('handler INFO level')
 
-                with caplog.at_level(logging.CRITICAL, logger='root.baz'):
-                    log = logging.getLogger('root.baz')
+                with caplog.logging_at_level(logging.CRITICAL,
+                                             logger='root.baz') as log:
                     log.warning('logger WARNING level')
                     log.critical('logger CRITICAL level')
 
@@ -121,9 +120,9 @@ def test_log_access(testdir):
 
         def test_foo(caplog):
             logging.getLogger().info('boo %s', 'arg')
-            assert caplog.records()[0].levelname == 'INFO'
-            assert caplog.records()[0].msg == 'boo %s'
-            assert 'boo arg' in caplog.text()
+            assert caplog.records[0].levelname == 'INFO'
+            assert caplog.records[0].msg == 'boo %s'
+            assert 'boo arg' in caplog.text
         ''')
     result = testdir.runpytest()
     assert result.ret == 0
@@ -144,7 +143,7 @@ def test_record_tuples(testdir):
         def test_foo(caplog):
             logging.getLogger().info('boo %s', 'arg')
 
-            assert caplog.record_tuples() == [
+            assert caplog.record_tuples == [
                 ('root', logging.INFO, 'boo arg'),
             ]
         ''')