| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| import logging.handlers | ||
|
|
||
| class TestHandler(logging.handlers.BufferingHandler): | ||
| def __init__(self, matcher): | ||
| # BufferingHandler takes a "capacity" argument | ||
| # so as to know when to flush. As we're overriding | ||
| # shouldFlush anyway, we can set a capacity of zero. | ||
| # You can call flush() manually to clear out the | ||
| # buffer. | ||
| logging.handlers.BufferingHandler.__init__(self, 0) | ||
| self.matcher = matcher | ||
|
|
||
| def shouldFlush(self): | ||
| return False | ||
|
|
||
| def emit(self, record): | ||
| self.format(record) | ||
| self.buffer.append(record.__dict__) | ||
|
|
||
| def matches(self, **kwargs): | ||
| """ | ||
| Look for a saved dict whose keys/values match the supplied arguments. | ||
| """ | ||
| result = False | ||
| for d in self.buffer: | ||
| if self.matcher.matches(d, **kwargs): | ||
| result = True | ||
| break | ||
| return result | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| 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)) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,2 @@ | ||
| The :mod:`logging` package is now imported lazily in :mod:`unittest` only | ||
| when the :meth:`~unittest.TestCase.assertLogs` assertion is used. |
| Back | FazBrowse Home | New Git URL |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityIt's only used in test_logging. Why not simply moving this class into test_logging?
We don't provide any backward compatibility warranty for test.support. It should not be used outside CPython code base.
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityIt is what I did originally. But then move it into a separate submodule on the request of @vsajip (logging maintainer, who added this class 10 years ago).
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.