cpython: 4f5815747f58 (original) (raw)

Mercurial > cpython

changeset 85701:4f5815747f58

Issue #18937: Add an assertLogs() context manager to unittest.TestCase to ensure that a block of code emits a message using the logging module. [#18937]

Antoine Pitrou solipsis@pitrou.net
date	Sat, 14 Sep 2013 19:45:47 +0200
parents	9354325d2ee4
children	1dc925ee441a
files	Doc/library/unittest.rst Lib/unittest/case.py Lib/unittest/test/test_case.py Misc/NEWS
diffstat	4 files changed, 242 insertions(+), 6 deletions(-)[+] [-] Doc/library/unittest.rst 41 Lib/unittest/case.py 109 Lib/unittest/test/test_case.py 96 Misc/NEWS 2

line wrap: on

line diff

--- a/Doc/library/unittest.rst +++ b/Doc/library/unittest.rst @@ -1031,6 +1031,47 @@ Test cases .. versionchanged:: 3.3 Added the msg keyword argument when used as a context manager.

.. method:: assertLogs(logger=None, level=None) +

 A context manager to test that at least one message is logged on[](#l1.9)

 the *logger* or one of its children, with at least the given[](#l1.10)

```
 *level*.[](#l1.11)
```

 If given, *logger* should be a :class:`logging.Logger` object or a[](#l1.13)

 :class:`str` giving the name of a logger.  The default is the root[](#l1.14)

 logger, which will catch all messages.[](#l1.15)

 If given, *level* should be either a numeric logging level or[](#l1.17)

 its string equivalent (for example either ``"ERROR"`` or[](#l1.18)

 :attr:`logging.ERROR`).  The default is :attr:`logging.INFO`.[](#l1.19)

 The test passes if at least one message emitted inside the ``with``[](#l1.21)

 block matches the *logger* and *level* conditions, otherwise it fails.[](#l1.22)

 The object returned by the context manager is a recording helper[](#l1.24)

 which keeps tracks of the matching log messages.  It has two[](#l1.25)

```
 attributes:[](#l1.26)
```

```
 .. attribute:: records[](#l1.28)
```

    A list of :class:`logging.LogRecord` objects of the matching[](#l1.30)

```
    log messages.[](#l1.31)
```

```
 .. attribute:: output[](#l1.33)
```

    A list of :class:`str` objects with the formatted output of[](#l1.35)

```
    matching messages.[](#l1.36)
```

```
 Example::[](#l1.38)
```

    with self.assertLogs('foo', level='INFO') as cm:[](#l1.40)

       logging.getLogger('foo').info('first message')[](#l1.41)

       logging.getLogger('foo.bar').error('second message')[](#l1.42)

    self.assertEqual(cm.output, ['INFO:foo:first message',[](#l1.43)

                                 'ERROR:foo.bar:second message'])[](#l1.44)

```
 .. versionadded:: 3.4[](#l1.46)
```

+ There are also other methods used to perform more specific checks, such as:

--- a/Lib/unittest/case.py +++ b/Lib/unittest/case.py @@ -3,6 +3,7 @@ import sys import functools import difflib +import logging import pprint import re import warnings @@ -115,10 +116,21 @@ def expectedFailure(test_item): return test_item -class _AssertRaisesBaseContext(object): +class _BaseTestCaseContext: +

def init(self, test_case):
```
   self.test_case = test_case[](#l2.19)
```

def _raiseFailure(self, standardMsg):

   msg = self.test_case._formatMessage(self.msg, standardMsg)[](#l2.22)

   raise self.test_case.failureException(msg)[](#l2.23)

+ + +class _AssertRaisesBaseContext(_BaseTestCaseContext): def init(self, expected, test_case, callable_obj=None, expected_regex=None):

   _BaseTestCaseContext.__init__(self, test_case)[](#l2.30)
   self.expected = expected[](#l2.31)
   self.test_case = test_case[](#l2.32)
   if callable_obj is not None:[](#l2.33)

@@ -133,10 +145,6 @@ class _AssertRaisesBaseContext(object): self.expected_regex = expected_regex self.msg = None

def _raiseFailure(self, standardMsg):

   msg = self.test_case._formatMessage(self.msg, standardMsg)[](#l2.39)

   raise self.test_case.failureException(msg)[](#l2.40)

- def handle(self, name, callable_obj, args, kwargs): """ If callable_obj is None, assertRaises/Warns is being used as a @@ -150,7 +158,6 @@ class _AssertRaisesBaseContext(object): callable_obj(*args, *kwargs) - class _AssertRaisesContext(_AssertRaisesBaseContext): """A context manager used to implement TestCase.assertRaises methods.""" @@ -232,6 +239,74 @@ class _AssertWarnsContext(_AssertRaisesB self._raiseFailure("{} not triggered".format(exc_name)) + +_LoggingWatcher = collections.namedtuple("_LoggingWatcher",

                                    ["records", "output"])[](#l2.59)

+ + +class _CapturingHandler(logging.Handler):

"""
A logging handler capturing all (raw and formatted) logging output.
"""

def init(self):

   logging.Handler.__init__(self)[](#l2.68)

   self.watcher = _LoggingWatcher([], [])[](#l2.69)

def flush(self):
```
   pass[](#l2.72)
```

def emit(self, record):

   self.watcher.records.append(record)[](#l2.75)

```
   msg = self.format(record)[](#l2.76)
```

   self.watcher.output.append(msg)[](#l2.77)

+ + + +class _AssertLogsContext(_BaseTestCaseContext):

"""A context manager used to implement TestCase.assertLogs()."""

LOGGING_FORMAT = "%(levelname)s:%(name)s:%(message)s"

def init(self, test_case, logger_name, level):

   _BaseTestCaseContext.__init__(self, test_case)[](#l2.87)

   self.logger_name = logger_name[](#l2.88)

```
   if level:[](#l2.89)
```

       self.level = logging._nameToLevel.get(level, level)[](#l2.90)

```
   else:[](#l2.91)
```

       self.level = logging.INFO[](#l2.92)

```
   self.msg = None[](#l2.93)
```

def enter(self):

   if isinstance(self.logger_name, logging.Logger):[](#l2.96)

       logger = self.logger = self.logger_name[](#l2.97)

```
   else:[](#l2.98)
```

       logger = self.logger = logging.getLogger(self.logger_name)[](#l2.99)

   formatter = logging.Formatter(self.LOGGING_FORMAT)[](#l2.100)

   handler = _CapturingHandler()[](#l2.101)

   handler.setFormatter(formatter)[](#l2.102)

   self.watcher = handler.watcher[](#l2.103)

   self.old_handlers = logger.handlers[:][](#l2.104)

   self.old_level = logger.level[](#l2.105)

   self.old_propagate = logger.propagate[](#l2.106)

   logger.handlers = [handler][](#l2.107)

   logger.setLevel(self.level)[](#l2.108)

```
   logger.propagate = False[](#l2.109)
```
```
   return handler.watcher[](#l2.110)
```

def exit(self, exc_type, exc_value, tb):

   self.logger.handlers = self.old_handlers[](#l2.113)

   self.logger.propagate = self.old_propagate[](#l2.114)

   self.logger.setLevel(self.old_level)[](#l2.115)

```
   if exc_type is not None:[](#l2.116)
```

       # let unexpected exceptions pass through[](#l2.117)

```
       return False[](#l2.118)
```

   if len(self.watcher.records) == 0:[](#l2.119)

```
       self._raiseFailure([](#l2.120)
```

           "no logs of level {} or higher triggered on {}"[](#l2.121)

           .format(logging.getLevelName(self.level), self.logger.name))[](#l2.122)

+ + class TestCase(object): """A class whose instances are single test cases. @@ -644,6 +719,28 @@ class TestCase(object): context = _AssertWarnsContext(expected_warning, self, callable_obj) return context.handle('assertWarns', callable_obj, args, kwargs)

def assertLogs(self, logger=None, level=None):

   """Fail unless a log message of level *level* or higher is emitted[](#l2.133)

   on *logger_name* or its children.  If omitted, *level* defaults to[](#l2.134)

   INFO and *logger* defaults to the root logger.[](#l2.135)

   This method must be used as a context manager, and will yield[](#l2.137)

   a recording object with two attributes: `output` and `records`.[](#l2.138)

   At the end of the context manager, the `output` attribute will[](#l2.139)

   be a list of the matching formatted log messages and the[](#l2.140)

   `records` attribute will be a list of the corresponding LogRecord[](#l2.141)

```
   objects.[](#l2.142)
```

```
   Example::[](#l2.144)
```

       with self.assertLogs('foo', level='INFO') as cm:[](#l2.146)

           logging.getLogger('foo').info('first message')[](#l2.147)

           logging.getLogger('foo.bar').error('second message')[](#l2.148)

       self.assertEqual(cm.output, ['INFO:foo:first message',[](#l2.149)

                                    'ERROR:foo.bar:second message'])[](#l2.150)

```
   """[](#l2.151)
```

   return _AssertLogsContext(self, logger, level)[](#l2.152)

+ def _getAssertEqualityFunc(self, first, second): """Get a detailed comparison function for the types of the two args.

--- a/Lib/unittest/test/test_case.py +++ b/Lib/unittest/test/test_case.py @@ -1,8 +1,10 @@ +import contextlib import difflib import pprint import pickle import re import sys +import logging import warnings import weakref import inspect @@ -16,6 +18,12 @@ from .support import ( TestEquality, TestHashing, LoggingResult, LegacyLoggingResult, ResultWithNoStartTestRunStopTestRun ) +from test.support import captured_stderr + + +log_foo = logging.getLogger('foo') +log_foobar = logging.getLogger('foo.bar') +log_quux = logging.getLogger('quux') class Test(object): @@ -1251,6 +1259,94 @@ test case with self.assertWarnsRegex(RuntimeWarning, "o+"): _runtime_warn("barz")

@contextlib.contextmanager
def assertNoStderr(self):

   with captured_stderr() as buf:[](#l3.33)

```
       yield[](#l3.34)
```

   self.assertEqual(buf.getvalue(), "")[](#l3.35)

def assertLogRecords(self, records, matches):

   self.assertEqual(len(records), len(matches))[](#l3.38)

   for rec, match in zip(records, matches):[](#l3.39)

       self.assertIsInstance(rec, logging.LogRecord)[](#l3.40)

       for k, v in match.items():[](#l3.41)

           self.assertEqual(getattr(rec, k), v)[](#l3.42)

def testAssertLogsDefaults(self):

   # defaults: root logger, level INFO[](#l3.45)

   with self.assertNoStderr():[](#l3.46)

       with self.assertLogs() as cm:[](#l3.47)

```
           log_foo.info("1")[](#l3.48)
```

           log_foobar.debug("2")[](#l3.49)

       self.assertEqual(cm.output, ["INFO:foo:1"])[](#l3.50)

       self.assertLogRecords(cm.records, [{'name': 'foo'}])[](#l3.51)

def testAssertLogsTwoMatchingMessages(self):

   # Same, but with two matching log messages[](#l3.54)

   with self.assertNoStderr():[](#l3.55)

       with self.assertLogs() as cm:[](#l3.56)

```
           log_foo.info("1")[](#l3.57)
```

           log_foobar.debug("2")[](#l3.58)

           log_quux.warning("3")[](#l3.59)

       self.assertEqual(cm.output, ["INFO:foo:1", "WARNING:quux:3"])[](#l3.60)

       self.assertLogRecords(cm.records,[](#l3.61)

                              [{'name': 'foo'}, {'name': 'quux'}])[](#l3.62)

def checkAssertLogsPerLevel(self, level):
```
   # Check level filtering[](#l3.65)
```

   with self.assertNoStderr():[](#l3.66)

       with self.assertLogs(level=level) as cm:[](#l3.67)

           log_foo.warning("1")[](#l3.68)

           log_foobar.error("2")[](#l3.69)

           log_quux.critical("3")[](#l3.70)

       self.assertEqual(cm.output, ["ERROR:foo.bar:2", "CRITICAL:quux:3"])[](#l3.71)

       self.assertLogRecords(cm.records,[](#l3.72)

                              [{'name': 'foo.bar'}, {'name': 'quux'}])[](#l3.73)

def testAssertLogsPerLevel(self):

   self.checkAssertLogsPerLevel(logging.ERROR)[](#l3.76)

   self.checkAssertLogsPerLevel('ERROR')[](#l3.77)

def checkAssertLogsPerLogger(self, logger):

   # Check per-logger fitering[](#l3.80)

   with self.assertNoStderr():[](#l3.81)

       with self.assertLogs(level='DEBUG') as outer_cm:[](#l3.82)

           with self.assertLogs(logger, level='DEBUG') as cm:[](#l3.83)

               log_foo.info("1")[](#l3.84)

               log_foobar.debug("2")[](#l3.85)

               log_quux.warning("3")[](#l3.86)

           self.assertEqual(cm.output, ["INFO:foo:1", "DEBUG:foo.bar:2"])[](#l3.87)

           self.assertLogRecords(cm.records,[](#l3.88)

                                  [{'name': 'foo'}, {'name': 'foo.bar'}])[](#l3.89)

       # The outer catchall caught the quux log[](#l3.90)

       self.assertEqual(outer_cm.output, ["WARNING:quux:3"])[](#l3.91)

def testAssertLogsPerLogger(self):

   self.checkAssertLogsPerLogger(logging.getLogger('foo'))[](#l3.94)

   self.checkAssertLogsPerLogger('foo')[](#l3.95)

def testAssertLogsFailureNoLogs(self):
```
   # Failure due to no logs[](#l3.98)
```

   with self.assertNoStderr():[](#l3.99)

       with self.assertRaises(self.failureException):[](#l3.100)

           with self.assertLogs():[](#l3.101)

```
               pass[](#l3.102)
```

def testAssertLogsFailureLevelTooHigh(self):

   # Failure due to level too high[](#l3.105)

   with self.assertNoStderr():[](#l3.106)

       with self.assertRaises(self.failureException):[](#l3.107)

           with self.assertLogs(level='WARNING'):[](#l3.108)

               log_foo.info("1")[](#l3.109)

def testAssertLogsFailureMismatchingLogger(self):

   # Failure due to mismatching logger (and the logged message is[](#l3.112)

```
   # passed through)[](#l3.113)
```

   with self.assertLogs('quux', level='ERROR'):[](#l3.114)

       with self.assertRaises(self.failureException):[](#l3.115)

           with self.assertLogs('foo'):[](#l3.116)

               log_quux.error("1")[](#l3.117)

+ def testDeprecatedMethodNames(self): """ Test that the deprecated methods raise a DeprecationWarning. See #9424.

--- a/Misc/NEWS +++ b/Misc/NEWS @@ -12,6 +12,8 @@ Core and Builtins Library ------- +- Issue #18937: Add an assertLogs() context manager to unittest.TestCase

to ensure that a block of code emits a message using the logging module.

Issue #17324: Fix http.server's request handling case on trailing '/'. Patch contributed by Vajrasky Kok.