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().

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

Lib/test/_test_multiprocessing.py

@@ -5341,10 +5341,9 @@ class TestSyncManagerTypes(unittest.TestCase):
             dt = time.monotonic() - start_time
             if dt >= 5.0:
                 test.support.environment_altered = True
-                print("Warning -- multiprocessing.Manager still has %s active "
+                support.print_warning(f"multiprocessing.Manager still has "
-                      "children after %s seconds"
+                                      f"{multiprocessing.active_children()} "
-                      % (multiprocessing.active_children(), dt),
+                                      f"active children after {dt} seconds")
-                      file=sys.stderr)
                 break
 
     def run_worker(self, worker, obj):
@@ -5544,15 +5543,13 @@ class BaseMixin(object):
         processes = set(multiprocessing.process._dangling) - set(cls.dangling[0])
         if processes:
             test.support.environment_altered = True
-            print('Warning -- Dangling processes: %s' % processes,
+            support.print_warning(f'Dangling processes: {processes}')
-                  file=sys.stderr)
         processes = None
 
         threads = set(threading._dangling) - set(cls.dangling[1])
         if threads:
             test.support.environment_altered = True
-            print('Warning -- Dangling threads: %s' % threads,
+            support.print_warning(f'Dangling threads: {threads}')
-                  file=sys.stderr)
         threads = None
 
 
@@ -5620,10 +5617,9 @@ class ManagerMixin(BaseMixin):
             dt = time.monotonic() - start_time
             if dt >= 5.0:
                 test.support.environment_altered = True
-                print("Warning -- multiprocessing.Manager still has %s active "
+                support.print_warning(f"multiprocessing.Manager still has "
-                      "children after %s seconds"
+                                      f"{multiprocessing.active_children()} "
-                      % (multiprocessing.active_children(), dt),
+                                      f"active children after {dt} seconds")
-                      file=sys.stderr)
                 break
 
         gc.collect()                       # do garbage collection
@@ -5632,9 +5628,9 @@ class ManagerMixin(BaseMixin):
             # 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 '
+            support.print_warning('Shared objects which still exist '
-                  'shutdown:')
+                                  'at manager shutdown:')
-            print(cls.manager._debug_info())
+            support.print_warning(cls.manager._debug_info())
         cls.manager.shutdown()
         cls.manager.join()
         cls.manager = None
@@ -5731,16 +5727,14 @@ def install_tests_in_module_dict(remote_globs, start_method):
         if processes:
             need_sleep = True
             test.support.environment_altered = True
-            print('Warning -- Dangling processes: %s' % processes,
+            support.print_warning(f'Dangling processes: {processes}')
-                  file=sys.stderr)
         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,
+            support.print_warning(f'Dangling threads: {threads}')
-                  file=sys.stderr)
         threads = None
 
         # Sleep 500 ms to give time to child processes to complete.

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:

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

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 "
+            print_warning(f"threading_cleanup() failed to cleanup "
-                  "%s threads (count: %s, dangling: %s)"
+                          f"{values[0] - original_values[0]} threads "
-                  % (values[0] - original_values[0],
+                          f"(count: {values[0]}, "
-                     values[0], len(dangling_threads)),
+                          f"dangling: {len(dangling_threads)})")
-                  file=sys.stderr)
             for thread in dangling_threads:
-                print(f"Dangling thread: {thread!r}", file=sys.stderr)
+                print_warning(f"Dangling thread: {thread!r}")
-            sys.stderr.flush()
 
             # 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"
+        print_warning(f"reap_children() reaped child process {pid}")
-              % pid, file=sys.stderr)
         environment_altered = True
 
 

Lib/test/test_support.py

@@ -430,8 +430,12 @@ class TestSupport(unittest.TestCase):
                 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 @@ class TestSupport(unittest.TestCase):
             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