Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Move _AssertLogsContext in a private unittest._log submodule.
  • Loading branch information
serhiy-storchaka committed Apr 20, 2020
commit 58ae64560918ad08c9e6d2faaaf3479fea65d7ed
71 changes: 71 additions & 0 deletions Lib/unittest/_log.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
import logging
import collections

from .case import _BaseTestCaseContext


_LoggingWatcher = collections.namedtuple("_LoggingWatcher",
["records", "output"])

class _CapturingHandler(logging.Handler):
"""
A logging handler capturing all (raw and formatted) logging output.
"""

def __init__(self):
logging.Handler.__init__(self)
self.watcher = _LoggingWatcher([], [])

def flush(self):
pass

def emit(self, record):
self.watcher.records.append(record)
msg = self.format(record)
self.watcher.output.append(msg)


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)
self.logger_name = logger_name
if level:
self.level = logging._nameToLevel.get(level, level)
else:
self.level = logging.INFO
self.msg = None

def __enter__(self):
if isinstance(self.logger_name, logging.Logger):
logger = self.logger = self.logger_name
else:
logger = self.logger = logging.getLogger(self.logger_name)
formatter = logging.Formatter(self.LOGGING_FORMAT)
handler = _CapturingHandler()
handler.setFormatter(formatter)
self.watcher = handler.watcher
self.old_handlers = logger.handlers[:]
self.old_level = logger.level
self.old_propagate = logger.propagate
logger.handlers = [handler]
logger.setLevel(self.level)
logger.propagate = False
return handler.watcher

def __exit__(self, exc_type, exc_value, tb):
self.logger.handlers = self.old_handlers
self.logger.propagate = self.old_propagate
self.logger.setLevel(self.old_level)
if exc_type is not None:
# let unexpected exceptions pass through
return False
if len(self.watcher.records) == 0:
self._raiseFailure(
"no logs of level {} or higher triggered on {}"
.format(logging.getLevelName(self.level), self.logger.name))

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please remove empty lines at the end.


81 changes: 2 additions & 79 deletions Lib/unittest/case.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,6 @@
from .util import (strclass, safe_repr, _count_diff_all_purpose,
_count_diff_hashable, _common_shorten_repr)

logging = None # imported lazily

__unittest = True

_subtest_msg_sentinel = object()
Expand Down Expand Up @@ -298,83 +296,6 @@ def __exit__(self, exc_type, exc_value, tb):



_LoggingWatcher = collections.namedtuple("_LoggingWatcher",
["records", "output"])

# Import logging lazily for assertLogs().
# assertLogs() is one of the least used assertions, and most tests do
# not need to import logging.

def _lazy_logging_import():
global logging
global _CapturingHandler
if logging:
return

import logging
class _CapturingHandler(logging.Handler):
"""
A logging handler capturing all (raw and formatted) logging output.
"""

def __init__(self):
logging.Handler.__init__(self)
self.watcher = _LoggingWatcher([], [])

def flush(self):
pass

def emit(self, record):
self.watcher.records.append(record)
msg = self.format(record)
self.watcher.output.append(msg)


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)
self.logger_name = logger_name
_lazy_logging_import()
if level:
self.level = logging._nameToLevel.get(level, level)
else:
self.level = logging.INFO
self.msg = None

def __enter__(self):
if isinstance(self.logger_name, logging.Logger):
logger = self.logger = self.logger_name
else:
logger = self.logger = logging.getLogger(self.logger_name)
formatter = logging.Formatter(self.LOGGING_FORMAT)
handler = _CapturingHandler()
handler.setFormatter(formatter)
self.watcher = handler.watcher
self.old_handlers = logger.handlers[:]
self.old_level = logger.level
self.old_propagate = logger.propagate
logger.handlers = [handler]
logger.setLevel(self.level)
logger.propagate = False
return handler.watcher

def __exit__(self, exc_type, exc_value, tb):
self.logger.handlers = self.old_handlers
self.logger.propagate = self.old_propagate
self.logger.setLevel(self.old_level)
if exc_type is not None:
# let unexpected exceptions pass through
return False
if len(self.watcher.records) == 0:
self._raiseFailure(
"no logs of level {} or higher triggered on {}"
.format(logging.getLevelName(self.level), self.logger.name))


class _OrderedChainMap(collections.ChainMap):
def __iter__(self):
seen = set()
Expand Down Expand Up @@ -865,6 +786,8 @@ def assertLogs(self, logger=None, level=None):
self.assertEqual(cm.output, ['INFO:foo:first message',
'ERROR:foo.bar:second message'])
"""
# Lazy import to avoid importing logging if it is not needed.
from ._log import _AssertLogsContext
return _AssertLogsContext(self, logger, level)

def _getAssertEqualityFunc(self, first, second):
Expand Down