Skip to content

[WIP] bpo-40275: Add loggingutils in test.support #19599

New issue

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

Closed
wants to merge 4 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
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
35 changes: 0 additions & 35 deletions Lib/test/support/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
import importlib
import importlib.util
import locale
import logging.handlers
import nntplib
import os
import platform
Expand Down Expand Up @@ -109,8 +108,6 @@
"bind_unix_socket",
# processes
'temp_umask', "reap_children",
# logging
"TestHandler",
# threads
"threading_setup", "threading_cleanup", "reap_threads", "start_threads",
# miscellaneous
Expand Down Expand Up @@ -2522,38 +2519,6 @@ def optim_args_from_interpreter_flags():
optimization settings in sys.flags."""
return subprocess._optim_args_from_interpreter_flags()

#============================================================
# Support for assertions about logging.
#============================================================

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

class Matcher(object):

_partial_matches = ('msg', 'message')
Expand Down
40 changes: 40 additions & 0 deletions Lib/test/support/loggingutils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
"""Utilities to support logging in the Python regression tests."""

import logging.handlers

__all__ = [
# logging
"TestHandler",
]

#============================================================
# Support for assertions about logging.
#============================================================

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
5 changes: 3 additions & 2 deletions Lib/test/test_logging.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import struct
import sys
import tempfile
from test.support.loggingutils import TestHandler
from test.support.script_helper import assert_python_ok, assert_python_failure
from test import support
import textwrap
Expand Down Expand Up @@ -3523,7 +3524,7 @@ def test_formatting(self):
@unittest.skipUnless(hasattr(logging.handlers, 'QueueListener'),
'logging.handlers.QueueListener required for this test')
def test_queue_listener(self):
handler = support.TestHandler(support.Matcher())
handler = TestHandler(support.Matcher())
listener = logging.handlers.QueueListener(self.queue, handler)
listener.start()
try:
Expand All @@ -3539,7 +3540,7 @@ def test_queue_listener(self):

# Now test with respect_handler_level set

handler = support.TestHandler(support.Matcher())
handler = TestHandler(support.Matcher())
handler.setLevel(logging.CRITICAL)
listener = logging.handlers.QueueListener(self.queue, handler,
respect_handler_level=True)
Expand Down