bpo-40275: Avoid importing logging in test.support (GH-19601)
Import logging lazily in assertLogs() in unittest.
Move TestHandler from test.support to logging_helper.
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index 15cf45d..f48decc 100644
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -15,7 +15,6 @@
import importlib
import importlib.util
import locale
-import logging.handlers
import os
import platform
import re
@@ -99,8 +98,6 @@
"open_urlresource",
# processes
'temp_umask', "reap_children",
- # logging
- "TestHandler",
# threads
"threading_setup", "threading_cleanup", "reap_threads", "start_threads",
# miscellaneous
@@ -2368,37 +2365,6 @@
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):
diff --git a/Lib/test/support/logging_helper.py b/Lib/test/support/logging_helper.py
new file mode 100644
index 0000000..12fcca4
--- /dev/null
+++ b/Lib/test/support/logging_helper.py
@@ -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
diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py
index 241ed2c..e1d0eb8 100644
--- a/Lib/test/test_logging.py
+++ b/Lib/test/test_logging.py
@@ -44,6 +44,7 @@
from test.support.script_helper import assert_python_ok, assert_python_failure
from test import support
from test.support import socket_helper
+from test.support.logging_helper import TestHandler
import textwrap
import threading
import time
@@ -3524,7 +3525,7 @@
@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:
@@ -3540,7 +3541,7 @@
# 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)