bpo-45410: Enhance libregrtest -W/--verbose3 option

Lib/test/libregrtest/runtest.py

@@ -1,10 +1,12 @@
+import contextlib
 import faulthandler
 import functools
 import gc
 import importlib
 import io
 import os
 import sys
+import tempfile
 import time
 import traceback
 import unittest
@@ -173,6 +175,63 @@ def get_abs_module(ns: Namespace, test_name: str) -> str:
         return 'test.' + test_name
 
 
+@contextlib.contextmanager
+def override_fd(fd, fd2):
+    fd2_copy = os.dup(fd2)
+    try:
+        os.dup2(fd, fd2)
+        yield
+    finally:
+        os.dup2(fd2_copy, fd2)
+        os.close(fd2_copy)
+
+
+def get_stream_fd(stream):
+    if stream is None:
+        return None
+    try:
+        return stream.fileno()
+    except io.UnsupportedOperation:
+        return None
+
+
+@contextlib.contextmanager
+def capture_std_streams():
+    """
+    Redirect all standard streams to a temporary file:
+
+    * stdout and stderr file descriptors (fd 1 and fd 2)
+    * sys.stdout, sys.__stdout__
+    * sys.stderr, sys.__stderr__
+    """
+    try:
+        stderr_fd = sys.stderr.fileno()
+    except io.UnsupportedOperation:
+        stderr_fd = None
+
+    # Use a temporary file to support fileno() operation
+    tmp_file = tempfile.TemporaryFile(mode='w+',
+                                      # line buffering
+                                      buffering=1,
+                                      encoding=sys.stderr.encoding,
+                                      errors=sys.stderr.errors)
+    with contextlib.ExitStack() as stack:
+        stack.enter_context(tmp_file)
+
+        # Override stdout and stderr file descriptors
+        tmp_fd = tmp_file.fileno()
+        for stream in (sys.stdout, sys.stderr):
+            fd = get_stream_fd(stream)
+            if fd is not None:
+                stack.enter_context(override_fd(tmp_fd, fd))
+
+        # Override sys attributes
+        for name in ('stdout', 'stderr', '__stdout__', '__stderr__'):
+            stack.enter_context(support.swap_attr(sys, name, tmp_file))
+
+        yield tmp_file
+
+
 def _runtest(ns: Namespace, test_name: str) -> TestResult:
     # Handle faulthandler timeout, capture stdout+stderr, XML serialization
     # and measure time.
@@ -193,21 +252,17 @@ def _runtest(ns: Namespace, test_name: str) -> TestResult:
         if output_on_failure:
             support.verbose = True
 
-            stream = io.StringIO()
-            orig_stdout = sys.stdout
-            orig_stderr = sys.stderr
-            try:
-                sys.stdout = stream
-                sys.stderr = stream
+            output = None
+            with capture_std_streams() as stream:
                 result = _runtest_inner(ns, test_name,
                                         display_failure=False)
                 if not isinstance(result, Passed):
-                    output = stream.getvalue()
-                    orig_stderr.write(output)
-                    orig_stderr.flush()
-            finally:
-                sys.stdout = orig_stdout
-                sys.stderr = orig_stderr
+                    stream.seek(0)
+                    output = stream.read()
+
+            if output is not None:
+                sys.stderr.write(output)
+                sys.stderr.flush()
         else:
             # Tell tests to be moderately quiet
             support.verbose = ns.verbose

Lib/test/libregrtest/utils.py

@@ -71,7 +71,7 @@ def print_warning(msg):
 def regrtest_unraisable_hook(unraisable):
     global orig_unraisablehook
     support.environment_altered = True
-    print_warning("Unraisable exception")
+    support.print_warning("Unraisable exception")
     old_stderr = sys.stderr
     try:
         support.flush_std_streams()
@@ -94,7 +94,7 @@ def setup_unraisable_hook():
 def regrtest_threading_excepthook(args):
     global orig_threading_excepthook
     support.environment_altered = True
-    print_warning(f"Uncaught thread exception: {args.exc_type.__name__}")
+    support.print_warning(f"Uncaught thread exception: {args.exc_type.__name__}")
     old_stderr = sys.stderr
     try:
         support.flush_std_streams()

Lib/test/support/__init__.py

@@ -1179,8 +1179,10 @@ def print_warning(msg):
     flush_std_streams()
     # bpo-39983: Print into sys.__stderr__ to display the warning even
     # when sys.stderr is captured temporarily by a test
+    stream = sys.__stderr__
     for line in msg.splitlines():
-        print(f"Warning -- {line}", file=sys.__stderr__, flush=True)
+        print(f"Warning -- {line}", file=stream)
+    stream.flush()
 
 
 # Flag used by saved_test_environment of test.libregrtest.save_env,