[Python-checkins] bpo-39983: Add test.support.print_warning() (GH-19683)
Victor Stinner
webhook-mailer at python.org
Thu Apr 23 13:04:00 EDT 2020
https://github.com/python/cpython/commit/d663d34685e18588748569468c672763f4c73b3e
commit: d663d34685e18588748569468c672763f4c73b3e
branch: master
author: Victor Stinner <vstinner at python.org>
committer: GitHub <noreply at github.com>
date: 2020-04-23T19:03:52+02:00
summary:
bpo-39983: Add test.support.print_warning() (GH-19683)
Log "Warning -- ..." test warnings into sys.__stderr__ rather than
sys.stderr, to ensure to display them even if sys.stderr is captured.
test.libregrtest.utils.print_warning() now calls
test.support.print_warning().
files:
M Doc/library/test.rst
M Lib/test/_test_multiprocessing.py
M Lib/test/libregrtest/runtest.py
M Lib/test/libregrtest/utils.py
M Lib/test/support/__init__.py
M Lib/test/test_support.py
diff --git a/Doc/library/test.rst b/Doc/library/test.rst
index c33465d758d57..0573c275981c7 100644
--- a/Doc/library/test.rst
+++ b/Doc/library/test.rst
@@ -825,6 +825,15 @@ The :mod:`test.support` module defines the following functions:
target of the "as" clause, if there is one.
+.. function:: print_warning(msg)
+
+ Print a warning into :data:`sys.__stderr__`. Format the message as:
+ ``f"Warning -- {msg}"``. If *msg* is made of multiple lines, add
+ ``"Warning -- "`` prefix to each line.
+
+ .. versionadded:: 3.9
+
+
.. function:: wait_process(pid, *, exitcode, timeout=None)
Wait until process *pid* completes and check that the process exit code is
diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index d633e02d016fc..376f5e33466a4 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -5341,10 +5341,9 @@ def wait_proc_exit(self):
dt = time.monotonic() - start_time
if dt >= 5.0:
test.support.environment_altered = True
- print("Warning -- multiprocessing.Manager still has %s active "
- "children after %s seconds"
- % (multiprocessing.active_children(), dt),
- file=sys.stderr)
+ support.print_warning(f"multiprocessing.Manager still has "
+ f"{multiprocessing.active_children()} "
+ f"active children after {dt} seconds")
break
def run_worker(self, worker, obj):
@@ -5544,15 +5543,13 @@ def tearDownClass(cls):
processes = set(multiprocessing.process._dangling) - set(cls.dangling[0])
if processes:
test.support.environment_altered = True
- print('Warning -- Dangling processes: %s' % processes,
- file=sys.stderr)
+ support.print_warning(f'Dangling processes: {processes}')
processes = None
threads = set(threading._dangling) - set(cls.dangling[1])
if threads:
test.support.environment_altered = True
- print('Warning -- Dangling threads: %s' % threads,
- file=sys.stderr)
+ support.print_warning(f'Dangling threads: {threads}')
threads = None
@@ -5620,10 +5617,9 @@ def tearDownClass(cls):
dt = time.monotonic() - start_time
if dt >= 5.0:
test.support.environment_altered = True
- print("Warning -- multiprocessing.Manager still has %s active "
- "children after %s seconds"
- % (multiprocessing.active_children(), dt),
- file=sys.stderr)
+ support.print_warning(f"multiprocessing.Manager still has "
+ f"{multiprocessing.active_children()} "
+ f"active children after {dt} seconds")
break
gc.collect() # do garbage collection
@@ -5632,9 +5628,9 @@ def tearDownClass(cls):
# ensure that all processes which hold a reference to a
# managed object have been joined.
test.support.environment_altered = True
- print('Warning -- Shared objects which still exist at manager '
- 'shutdown:')
- print(cls.manager._debug_info())
+ support.print_warning('Shared objects which still exist '
+ 'at manager shutdown:')
+ support.print_warning(cls.manager._debug_info())
cls.manager.shutdown()
cls.manager.join()
cls.manager = None
@@ -5731,16 +5727,14 @@ def tearDownModule():
if processes:
need_sleep = True
test.support.environment_altered = True
- print('Warning -- Dangling processes: %s' % processes,
- file=sys.stderr)
+ support.print_warning(f'Dangling processes: {processes}')
processes = None
threads = set(threading._dangling) - set(dangling[1])
if threads:
need_sleep = True
test.support.environment_altered = True
- print('Warning -- Dangling threads: %s' % threads,
- file=sys.stderr)
+ support.print_warning(f'Dangling threads: {threads}')
threads = None
# Sleep 500 ms to give time to child processes to complete.
diff --git a/Lib/test/libregrtest/runtest.py b/Lib/test/libregrtest/runtest.py
index 558f2099c66f5..9338b28047954 100644
--- a/Lib/test/libregrtest/runtest.py
+++ b/Lib/test/libregrtest/runtest.py
@@ -327,7 +327,7 @@ def cleanup_test_droppings(test_name, verbose):
f"directory nor file")
if verbose:
- print_warning("%r left behind %s %r" % (test_name, kind, name))
+ print_warning(f"{test_name} left behind {kind} {name!r}")
support.environment_altered = True
try:
diff --git a/Lib/test/libregrtest/utils.py b/Lib/test/libregrtest/utils.py
index 40faed832c11b..0368694b2adcb 100644
--- a/Lib/test/libregrtest/utils.py
+++ b/Lib/test/libregrtest/utils.py
@@ -62,7 +62,7 @@ def printlist(x, width=70, indent=4, file=None):
def print_warning(msg):
- print(f"Warning -- {msg}", file=sys.stderr, flush=True)
+ support.print_warning(msg)
orig_unraisablehook = None
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index 4fe247aeb9d37..f3868c1041542 100644
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -2250,6 +2250,12 @@ def run_doctest(module, verbosity=None, optionflags=0):
#=======================================================================
# Support for saving and restoring the imported modules.
+def print_warning(msg):
+ # bpo-39983: Print into sys.__stderr__ to display the warning even
+ # when sys.stderr is captured temporarily by a test
+ for line in msg.splitlines():
+ print(f"Warning -- {line}", file=sys.__stderr__, flush=True)
+
def modules_setup():
return sys.modules.copy(),
@@ -2305,14 +2311,12 @@ def threading_cleanup(*original_values):
# Display a warning at the first iteration
environment_altered = True
dangling_threads = values[1]
- print("Warning -- threading_cleanup() failed to cleanup "
- "%s threads (count: %s, dangling: %s)"
- % (values[0] - original_values[0],
- values[0], len(dangling_threads)),
- file=sys.stderr)
+ print_warning(f"threading_cleanup() failed to cleanup "
+ f"{values[0] - original_values[0]} threads "
+ f"(count: {values[0]}, "
+ f"dangling: {len(dangling_threads)})")
for thread in dangling_threads:
- print(f"Dangling thread: {thread!r}", file=sys.stderr)
- sys.stderr.flush()
+ print_warning(f"Dangling thread: {thread!r}")
# Don't hold references to threads
dangling_threads = None
@@ -2409,8 +2413,7 @@ def reap_children():
if pid == 0:
break
- print("Warning -- reap_children() reaped child process %s"
- % pid, file=sys.stderr)
+ print_warning(f"reap_children() reaped child process {pid}")
environment_altered = True
diff --git a/Lib/test/test_support.py b/Lib/test/test_support.py
index 99a4cad2bb887..dee1db7d6d7c8 100644
--- a/Lib/test/test_support.py
+++ b/Lib/test/test_support.py
@@ -430,8 +430,12 @@ def test_reap_children(self):
if time.monotonic() > deadline:
self.fail("timeout")
- with contextlib.redirect_stderr(stderr):
+ old_stderr = sys.__stderr__
+ try:
+ sys.__stderr__ = stderr
support.reap_children()
+ finally:
+ sys.__stderr__ = old_stderr
# Use environment_altered to check if reap_children() found
# the child process
@@ -629,6 +633,24 @@ def test_fd_count(self):
os.close(fd)
self.assertEqual(more - start, 1)
+ def check_print_warning(self, msg, expected):
+ stderr = io.StringIO()
+
+ old_stderr = sys.__stderr__
+ try:
+ sys.__stderr__ = stderr
+ support.print_warning(msg)
+ finally:
+ sys.__stderr__ = old_stderr
+
+ self.assertEqual(stderr.getvalue(), expected)
+
+ def test_print_warning(self):
+ self.check_print_warning("msg",
+ "Warning -- msg\n")
+ self.check_print_warning("a\nb",
+ 'Warning -- a\nWarning -- b\n')
+
# XXX -follows a list of untested API
# make_legacy_pyc
# is_resource_enabled
More information about the Python-checkins
mailing list