#138 Add default log check(assert) requirements for LogCapture

testfixtures/logcapture.py

@@ -42,15 +42,21 @@ class LogCapture(logging.Handler):
 
       If ``True``, log messages will be compared recursively by
       :meth:`LogCapture.check`.
+
+    :param ensure_checks_above: The log level above which checks must be made for logged events.
+
     """
 
     instances = set()
     atexit_setup = False
     installed = False
+    default_ensure_checks_above = logging.NOTSET
 
     def __init__(self, names=None, install=True, level=1, propagate=None,
                  attributes=('name', 'levelname', 'getMessage'),
-                 recursive_check=False):
+                 recursive_check=False,
+                 ensure_checks_above=None
+                 ):
         logging.Handler.__init__(self)
         if not isinstance(names, tuple):
             names = (names, )
@@ -60,7 +66,12 @@ def __init__(self, names=None, install=True, level=1, propagate=None,
         self.attributes = attributes
         self.recursive_check = recursive_check
         self.old = defaultdict(dict)
-        self.clear()
+        #: The log level above which checks must be made for logged events.
+        if ensure_checks_above is None:
+            self.ensure_checks_above = self.default_ensure_checks_above
+        else:
+            self.ensure_checks_above = ensure_checks_above
+        self.clear()  # declares self.records: List[LogRecord]
         if install:
             self.install()
 
@@ -73,11 +84,55 @@ def atexit(cls):
                 '%s' % ('\n'.join((str(i.names) for i in cls.instances)))
                 )
 
+    def __len__(self):
+        return len(self.records)
+
+    def __getitem__(self, index):
+        return self._actual_row(self.records[index])
+
+    def __contains__(self, what):
+        for i, item in enumerate(self):
+            if what == item:
+                self.records[i].checked = True
+                return True
+
     def clear(self):
         """Clear any entries that have been captured."""
         self.records = []
 
+    def mark_all_checked(self):
+        """
+        Mark all captured events as checked.
+        This should be called if you have made assertions about logging
+        other than through :class:`LogCapture` methods.
+        """
+        for record in self.records:
+            record.checked = True
+
+    def ensure_checked(self, level=None):
+        """
+        Ensure every entry logged above the specified `level` has been checked.
+        Raises an :class:`AssertionError` if this is not the case.
+
+        :param level: the logging level, defaults to :attr:`ensure_checks_above`.
+        :type level: Optional[int]
+        """
+        if level is None:
+            level = self.ensure_checks_above
+        if level == logging.NOTSET:
+            return
+        un_checked = []
+        for record in self.records:
+            if record.levelno >= level and not record.checked:
+                un_checked.append(self._actual_row(record))
+        if un_checked:
+            raise AssertionError((
+                    'Not asserted ERROR log(s): %s'
+                ) % (pformat(un_checked)))
+
     def emit(self, record):
+        # record: logging.LogRecord
+        record.checked = False
         self.records.append(record)
 
     def install(self):
@@ -130,11 +185,22 @@ def uninstall_all(cls):
             i.uninstall()
 
     def _actual_row(self, record):
-        for a in self.attributes:
-            value = getattr(record, a, None)
-            if callable(value):
-                value = value()
-            yield value
+        # Convert a log record to a Tuple or attribute value according the attributes member.
+        # record: logging.LogRecord
+
+        if callable(self.attributes):
+            return self.attributes(record)
+        else:
+            values = []
+            for a in self.attributes:
+                value = getattr(record, a, None)
+                if callable(value):
+                    value = value()
+                values.append(value)
+            if len(values) == 1:
+                return values[0]
+            else:
+                return tuple(values)
 
     def actual(self):
         """
@@ -145,17 +211,13 @@ def actual(self):
         This can be useful for making more complex assertions about logged
         records. The actual records logged can also be inspected by using the
         :attr:`records` attribute.
+
+        :rtype: List
+
         """
         actual = []
         for r in self.records:
-            if callable(self.attributes):
-                actual.append(self.attributes(r))
-            else:
-                result = tuple(self._actual_row(r))
-                if len(result) == 1:
-                    actual.append(result[0])
-                else:
-                    actual.append(result)
+            actual.append(self._actual_row(r))
         return actual
 
     def __str__(self):
@@ -174,11 +236,13 @@ def check(self, *expected):
           A sequence of entries of the structure specified by the ``attributes``
           passed to the constructor.
         """
-        return compare(
+        result = compare(
             expected,
             actual=self.actual(),
             recursive=self.recursive_check
             )
+        self.mark_all_checked()
+        return result
 
     def check_present(self, *expected, **kw):
         """
@@ -213,6 +277,7 @@ def check_present(self, *expected, **kw):
                         matched.pop()
                     break
                 else:
+                    self.records[index].checked = True
                     matched_indices.append(index+1)
                     matched.append(entry)
             else:
@@ -225,12 +290,13 @@ def check_present(self, *expected, **kw):
             expected = list(expected)
             matched = []
             unmatched = []
-            for entry in actual:
+            for i, entry in enumerate(actual):
                 try:
                     index = expected.index(entry)
                 except ValueError:
                     unmatched.append(entry)
                 else:
+                    self.records[i].checked = True
                     matched.append(expected.pop(index))
                 if not expected:
                     break
@@ -247,6 +313,7 @@ def __enter__(self):
 
     def __exit__(self, type, value, traceback):
         self.uninstall()
+        self.ensure_checked()
 
 
 class LogCaptureForDecorator(LogCapture):

testfixtures/tests/test_log_capture.py

@@ -1,6 +1,6 @@
 from __future__ import absolute_import
 
-from logging import getLogger
+from logging import getLogger, ERROR
 from unittest import TestCase
 
 from testfixtures.shouldraise import ShouldAssert
@@ -49,6 +49,11 @@ def test_logging(self, l1, l2, l3):
             C('logging.LogRecord'),
             ])
 
+    @log_capture(ensure_checks_above=ERROR)
+    def test_simple_strict(self, l):
+        root.error('during')
+        l.check(("root", "ERROR", "during"))
+
     def test_uninstall_properly(self):
         root = getLogger()
         child = getLogger('child')

testfixtures/tests/test_logcapture.py

@@ -1,5 +1,5 @@
 from __future__ import print_function
-from logging import getLogger
+from logging import getLogger, ERROR
 from textwrap import dedent
 from unittest import TestCase
 from warnings import catch_warnings
@@ -25,6 +25,57 @@ def test_simple(self):
         root.info('after')
         assert str(l) == "root INFO\n  during"
 
+    def test_simple_strict(self):
+        log_capture = LogCapture(ensure_checks_above=ERROR)
+        root.error('during')
+        log_capture.uninstall()
+        with ShouldAssert("Not asserted ERROR log(s): [('root', 'ERROR', 'during')]"):
+            log_capture.ensure_checked()
+
+    def test_simple_strict_re_defaulted(self):
+        old = LogCapture.default_ensure_checks_above
+        try:
+            LogCapture.default_ensure_checks_above = ERROR
+            log_capture = LogCapture()
+            root.error('during')
+            log_capture.uninstall()
+            with ShouldAssert("Not asserted ERROR log(s): [('root', 'ERROR', 'during')]"):
+                log_capture.ensure_checked()
+        finally:
+            LogCapture.default_ensure_checks_above = old
+
+    def test_simple_strict_asserted(self):
+        log_capture = LogCapture(ensure_checks_above=ERROR)
+        root.error('during')
+        log_capture.uninstall()
+        log_capture.check(("root", "ERROR", "during"))
+        log_capture.ensure_checked()
+
+    def test_simple_strict_asserted_2(self):
+        log_capture = LogCapture(ensure_checks_above=ERROR)
+        root.error('during')
+        log_capture.uninstall()
+        assert ("root", "ERROR", "during") in log_capture
+        assert ("root", "INFO", "during") not in log_capture
+        log_capture.ensure_checked()
+
+    def test_simple_strict_asserted_3(self):
+        log_capture = LogCapture(ensure_checks_above=ERROR)
+        root.error('during')
+        log_capture.uninstall()
+        log_capture.mark_all_checked()
+        log_capture.ensure_checked()
+
+    def test_simple_strict_ctx(self):
+        with ShouldAssert("Not asserted ERROR log(s): [('root', 'ERROR', 'during')]"):
+            with LogCapture(ensure_checks_above=ERROR) as log_capture:
+                root.error('during')
+
+    def test_simple_strict_asserted_ctx(self):
+        with LogCapture(ensure_checks_above=ERROR) as log_capture:
+            root.error('during')
+            log_capture.check(("root", "ERROR", "during"))
+
     def test_specific_logger(self):
         l = LogCapture('one')
         root.info('1')