filter_record_tuples() implementation

.travis.yml

@@ -7,7 +7,9 @@ python:
   - "3.4"
   - "3.5"
   - "pypy"
-  - "pypy3"
+# Note: Disabled pypy3 due to currently being broken on travis-ci.
+# See: https://github.com/travis-ci/travis-ci/issues/6277#issuecomment-265128578
+#  - "pypy3"
 matrix:
   include:
     - python: "2.7"

README.rst

@@ -147,6 +147,21 @@ given severity and message::
             ('root', logging.INFO, 'boo arg'),
         ]
 
+To aid in common test scenarios, caplog exposes logging levels to prevent the
+need to import the ``logging`` module in tests.
+
+Furthermore ``filter_records`` or ``filter_record_tuples`` can be used to easily filter
+log messages of a particular logger. This is especially useful for testing composite systems where
+several components have loggers::
+
+    def test_foo(caplog):
+        func_under_test()
+
+        assert not any([r.levelno >= caplog.ERROR for r in caplog.filter_records('components.a')])
+
+        assert caplog.filter_record_tuples('components.a', caplog.INFO, 'foo')
+        assert caplog.filter_record_tuples('components.b', caplog.INFO, re.compile(r'foo\s.+'))
+
 You can call ``caplog.clear()`` to reset the captured log records in a test::
 
     def test_something_with_clearing_records(caplog):

pytest_catchlog/fixture.py

@@ -17,6 +17,15 @@ def __init__(self, item):
         """Creates a new funcarg."""
         self._item = item
 
+    def __getattr__(self, name):
+        if name.isupper() and name.isalpha():
+            # lookup names matching level pattern in logging module
+            return getattr(logging, name)
+        raise AttributeError("'{0}' object has no attribute '{1}'".format(
+            type(self).__name__,
+            name
+        ))
+
     @property
     def handler(self):
         return self._item.catch_log_handler
@@ -31,6 +40,17 @@ def records(self):
         """Returns the list of log records."""
         return self.handler.records
 
+    def filter_records(self,  name=None, level=None, message=None):
+        """Returns a filtered list of records
+
+        Args:
+            name (str, optional): Exact match of the logger name
+            level (int, optional): The log level of the record
+            message (str|regex, optional): Message part that should be in the record text or
+                the regular expression the record text should match
+        """
+        return self._filter_records(name, level, message)
+
     @property
     def record_tuples(self):
         """Returns a list of a striped down version of log records intended
@@ -40,7 +60,18 @@ def record_tuples(self):
 
             (logger_name, log_level, message)
         """
-        return [(r.name, r.levelno, r.getMessage()) for r in self.records]
+        return self._make_tuples(self.records)
+
+    def filter_record_tuples(self, name=None, level=None, message=None):
+        """Returns a filtered list of record tuples for use in asserts
+
+        Args:
+            name (str, optional): Exact match of the logger name
+            level (int, optional): The log level of the record
+            message (str|regex, optional): Message part that should be in the record text or 
+                the regular expression the record text should match
+        """
+        return self._make_tuples(self._filter_records(name, level, message))
 
     def clear(self):
         """Reset the list of log records."""
@@ -68,6 +99,27 @@ def at_level(self, level, logger=None):
         obj = logger and logging.getLogger(logger) or self.handler
         return logging_at_level(level, obj)
 
+    def _filter_records(self, name, level, message):
+        """Filter records on given args"""
+        def _filter(r):
+            if name and not r.name == name:
+                return False
+            if level and not r.levelno == level:
+                return False
+            if message:
+                try:
+                    if not bool(message.search(r.getMessage())):
+                        return False
+                except AttributeError:
+                    if message not in r.getMessage():
+                        return False
+            return True
+        return list(filter(_filter, self.handler.records))
+
+    def _make_tuples(self, records):
+        """Convert records to tuples"""
+        return [(r.name, r.levelno, r.getMessage()) for r in records]
+
 
 class CallablePropertyMixin(object):
     """Backward compatibility for functions that became properties."""

tests/test_fixture.py

@@ -1,13 +1,33 @@
 # -*- coding: utf-8 -*-
+import re
 import sys
 import logging
 
+import pytest
 
 logger = logging.getLogger(__name__)
 sublogger = logging.getLogger(__name__+'.baz')
 
 u = (lambda x: x.decode('utf-8')) if sys.version_info < (3,) else (lambda x: x)
 
+filter_params = [
+    (__name__, logging.INFO, None, ['foo', 'bar']),
+    (__name__, logging.INFO, 'foo', ['foo']),
+    (__name__, logging.INFO, 'o ar', ['foo']),
+    ('other', logging.INFO, 'foo', []),
+    (__name__, logging.WARNING, 'foo', []),
+    (__name__, 45, 'foo arg', ['foo']),
+    (__name__, logging.INFO, re.compile('o\s'), ['foo']),
+    (__name__, logging.INFO, re.compile('foo$'), []),
+]
+
+
+def do_test_filter_record_logging():
+    """do logging for tests parametrized by filter_params"""
+    logger.info('foo %s', 'arg')
+    logger.info('bar %s', 'arg')
+    logger.log(45, 'foo %s', 'arg')
+
 
 def test_fixture_help(testdir):
     result = testdir.runpytest('--fixtures')
@@ -59,6 +79,49 @@ def test_record_tuples(caplog):
     ]
 
 
+@pytest.mark.parametrize('name, level, message, expected', filter_params)
+def test_filter_records(caplog, name, level, message, expected):
+    do_test_filter_record_logging()
+
+    filtered = caplog.filter_records(name, level, message)
+    filtered_named_args = caplog.filter_records(name=name, level=level, message=message)
+
+    assert filtered == filtered_named_args
+    assert [f.getMessage().split(' ')[0] for f in filtered] == expected
+
+
+@pytest.mark.parametrize('name, level, message, expected', filter_params)
+def test_filter_record_tuples(caplog, name, level, message, expected):
+    do_test_filter_record_logging()
+
+    filtered = caplog.filter_record_tuples(name, level, message)
+    filtered_named_args = caplog.filter_record_tuples(name=name, level=level, message=message)
+
+    assert filtered == filtered_named_args
+    assert [f[2].split(' ')[0] for f in filtered] == expected
+
+
+@pytest.mark.parametrize('level', [
+    'CRITICAL',
+    'ERROR',
+    'WARNING',
+    'INFO',
+    'DEBUG',
+    'NOTSET',
+])
+def test_levels(caplog, level):
+    assert getattr(caplog, level) == getattr(logging, level)
+
+
+@pytest.mark.parametrize('level', [
+    'UNKNOWNLEVEL',  # Matches log level pattern, looked up from logging module.
+    'something_else'  # Not looked up from logging. Not an attribute of LogCaptureFixture.
+])
+def test_levels_error(caplog, level):
+    with pytest.raises(AttributeError):
+        getattr(caplog, level)
+
+
 def test_unicode(caplog):
     logger.info(u('bū'))
     assert caplog.records[0].levelname == 'INFO'