gh-131178: Use support.captured_stdout() and support.captured_stderr()

Author: donbarbos

Lib/test/test_argparse.py

@@ -56,7 +56,7 @@ class StdStreamTest(unittest.TestCase):
     def test_skip_invalid_stderr(self):
         parser = argparse.ArgumentParser()
         with (
-            contextlib.redirect_stderr(None),
+            captured_stderr(),
             mock.patch('argparse._sys.exit')
         ):
             parser.exit(status=0, message='foo')

Lib/test/test_ast/test_ast.py

@@ -1,7 +1,6 @@
 import _ast_unparse
 import ast
 import builtins
-import contextlib
 import copy
 import dis
 import enum
@@ -14,7 +13,6 @@
 import types
 import unittest
 import weakref
-from io import StringIO
 from pathlib import Path
 from textwrap import dedent
 try:
@@ -3417,11 +3415,9 @@ def set_source(self, content):
         Path(self.filename).write_text(self.text_normalize(content))
 
     def invoke_ast(self, *flags):
-        stderr = StringIO()
-        stdout = StringIO()
         with (
-            contextlib.redirect_stdout(stdout),
-            contextlib.redirect_stderr(stderr),
+            support.captured_stdout() as stdout,
+            support.captured_stderr() as stderr,
         ):
             ast.main(args=[*flags, self.filename])
         self.assertEqual(stderr.getvalue(), '')
@@ -3462,9 +3458,8 @@ def f(x: int) -> int:
     def test_help_message(self):
         for flag in ('-h', '--help', '--unknown'):
             with self.subTest(flag=flag):
-                output = StringIO()
                 with self.assertRaises(SystemExit):
-                    with contextlib.redirect_stderr(output):
+                    with support.captured_stderr() as output:
                         ast.main(args=flag)
                 self.assertStartsWith(output.getvalue(), 'usage: ')
 

Lib/test/test_compile.py

@@ -1,6 +1,4 @@
-import contextlib
 import dis
-import io
 import itertools
 import math
 import opcode
@@ -967,8 +965,7 @@ class C:
             for mode in ["exec", "single"]:
                 with self.subTest(opt=opt, mode=mode):
                     code = compile(src, "<test>", mode, optimize=opt)
-                    output = io.StringIO()
-                    with contextlib.redirect_stdout(output):
+                    with support.captured_stdout() as output:
                         dis.dis(code)
                     self.assertNotIn('NOP', output.getvalue())
 

Lib/test/test_compileall.py

@@ -2,7 +2,6 @@
 import contextlib
 import filecmp
 import importlib.util
-import io
 import os
 import py_compile
 import shutil
@@ -89,7 +88,7 @@ def test_year_2038_mtime_compilation(self):
             os.utime(self.source_path, (2**32 - 1, 2**32 - 1))
         except (OverflowError, OSError):
             self.skipTest("filesystem doesn't support timestamps near 2**32")
-        with contextlib.redirect_stdout(io.StringIO()):
+        with support.captured_stdout():
             self.assertTrue(compileall.compile_file(self.source_path))
 
     def test_larger_than_32_bit_times(self):
@@ -99,7 +98,7 @@ def test_larger_than_32_bit_times(self):
             os.utime(self.source_path, (2**35, 2**35))
         except (OverflowError, OSError):
             self.skipTest("filesystem doesn't support large timestamps")
-        with contextlib.redirect_stdout(io.StringIO()):
+        with support.captured_stdout():
             self.assertTrue(compileall.compile_file(self.source_path))
 
     def recreation_check(self, metadata):
@@ -206,7 +205,7 @@ def test_no_pycache_in_non_package(self):
     def test_compile_file_encoding_fallback(self):
         # Bug 44666 reported that compile_file failed when sys.stdout.encoding is None
         self.add_bad_source_file()
-        with contextlib.redirect_stdout(io.StringIO()):
+        with support.captured_stdout():
             self.assertFalse(compileall.compile_file(self.bad_source_path))
 
 
@@ -510,8 +509,7 @@ def tearDown(self):
         shutil.rmtree(self.directory)
 
     def test_error(self):
-        buffer = io.TextIOWrapper(io.BytesIO(), encoding='ascii')
-        with contextlib.redirect_stdout(buffer):
+        with support.captured_stdout() as buffer:
             compiled = compileall.compile_dir(self.directory)
         self.assertFalse(compiled)  # should not be successful
         buffer.seek(0)

Lib/test/test_concurrent_futures/test_interpreter_pool.py

@@ -1,7 +1,6 @@
 import _thread
 import asyncio
 import contextlib
-import io
 import os
 import subprocess
 import sys
@@ -199,14 +198,14 @@ def init2():
             nonlocal count
             count += 1
 
-        with contextlib.redirect_stderr(io.StringIO()) as stderr:
+        with support.captured_stderr() as stderr:
             with self.executor_type(initializer=init1) as executor:
                 fut = executor.submit(lambda: None)
         self.assertIn('NotShareableError', stderr.getvalue())
         with self.assertRaises(BrokenInterpreterPool):
             fut.result()
 
-        with contextlib.redirect_stderr(io.StringIO()) as stderr:
+        with support.captured_stderr() as stderr:
             with self.executor_type(initializer=init2) as executor:
                 fut = executor.submit(lambda: None)
         self.assertIn('NotShareableError', stderr.getvalue())
@@ -219,7 +218,7 @@ def initializer(self):
                 raise NotImplementedError
         spam = Spam()
 
-        with contextlib.redirect_stderr(io.StringIO()) as stderr:
+        with support.captured_stderr() as stderr:
             with self.executor_type(initializer=spam.initializer) as executor:
                 fut = executor.submit(lambda: None)
         self.assertIn('NotShareableError', stderr.getvalue())
@@ -230,7 +229,7 @@ def initializer(self):
     def test_init_exception_in_script(self):
         executor = self.executor_type(initializer='raise Exception("spam")')
         with executor:
-            with contextlib.redirect_stderr(io.StringIO()) as stderr:
+            with support.captured_stderr() as stderr:
                 fut = executor.submit('pass')
                 with self.assertRaises(BrokenInterpreterPool):
                     fut.result()
@@ -244,7 +243,7 @@ def test_init_exception_in_func(self):
         executor = self.executor_type(initializer=fail,
                                       initargs=(Exception, 'spam'))
         with executor:
-            with contextlib.redirect_stderr(io.StringIO()) as stderr:
+            with support.captured_stderr() as stderr:
                 fut = executor.submit(noop)
                 with self.assertRaises(BrokenInterpreterPool):
                     fut.result()

Lib/test/test_descr.py

@@ -1295,7 +1295,7 @@ def __init__(self):
             def __del__(self_):
                 self.assertEqual(self_.a, 1)
                 self.assertEqual(self_.b, 2)
-        with support.captured_output('stderr') as s:
+        with support.captured_stderr() as s:
             h = H()
             del h
         self.assertEqual(s.getvalue(), '')

Lib/test/test_dis.py

@@ -1,7 +1,6 @@
 # Minimal tests for dis module
 
 import ast
-import contextlib
 import dis
 import functools
 import io
@@ -13,9 +12,9 @@
 import textwrap
 import types
 import unittest
-from test.support import (captured_stdout, requires_debug_ranges,
-                          requires_specialization, cpython_only,
-                          os_helper, import_helper, reset_code)
+from test.support import (captured_stderr, captured_stdout,
+                          requires_debug_ranges, requires_specialization,
+                          cpython_only, os_helper, import_helper, reset_code)
 from test.support.bytecode_helper import BytecodeTestCase
 
 
@@ -971,8 +970,7 @@ class DisTests(DisTestBase):
 
     def get_disassembly(self, func, lasti=-1, wrapper=True, **kwargs):
         # We want to test the default printing behaviour, not the file arg
-        output = io.StringIO()
-        with contextlib.redirect_stdout(output):
+        with captured_stdout() as output:
             if wrapper:
                 dis.dis(func, **kwargs)
             else:
@@ -988,8 +986,7 @@ def do_disassembly_test(self, func, expected, **kwargs):
         self.do_disassembly_compare(got, expected)
         # Add checks for dis.disco
         if hasattr(func, '__code__'):
-            got_disco = io.StringIO()
-            with contextlib.redirect_stdout(got_disco):
+            with captured_stdout() as got_disco:
                 dis.disco(func.__code__, **kwargs)
             self.do_disassembly_compare(got_disco.getvalue(), expected)
 
@@ -1709,8 +1706,7 @@ def _stringify_instruction(instr):
     return base + "),"
 
 def _prepare_test_cases():
-    ignore = io.StringIO()
-    with contextlib.redirect_stdout(ignore):
+    with captured_stdout():
         f = outer()
         inner = f()
     _instructions_outer = dis.get_instructions(outer, first_line=expected_outer_line)
@@ -2428,8 +2424,7 @@ def setUp(self) -> None:
         return super().setUp()
 
     def get_disassembly(self, tb):
-        output = io.StringIO()
-        with contextlib.redirect_stdout(output):
+        with captured_stdout() as output:
             dis.distb(tb)
         return output.getvalue()
 
@@ -2455,8 +2450,7 @@ def test_distb_explicit_arg(self):
 class TestDisTracebackWithFile(TestDisTraceback):
     # Run the `distb` tests again, using the file arg instead of print
     def get_disassembly(self, tb):
-        output = io.StringIO()
-        with contextlib.redirect_stdout(output):
+        with captured_stdout() as output:
             dis.distb(tb, file=output)
         return output.getvalue()
 
@@ -2506,8 +2500,7 @@ def set_source(self, content):
             fp.write(self.text_normalize(content))
 
     def invoke_dis(self, *flags):
-        output = io.StringIO()
-        with contextlib.redirect_stdout(output):
+        with captured_stdout() as output:
             dis.main(args=[*flags, self.filename])
         return self.text_normalize(output.getvalue())
 
@@ -2541,7 +2534,7 @@ def f():
 
         with self.assertRaises(SystemExit):
             # suppress argparse error message
-            with contextlib.redirect_stderr(io.StringIO()):
+            with captured_stderr():
                 _ = self.invoke_dis('--unknown')
 
     def test_show_cache(self):

Lib/test/test_future_stmt/test_future_multiple_features.py

@@ -12,7 +12,7 @@ def test_unicode_literals(self):
         self.assertIsInstance("", str)
 
     def test_print_function(self):
-        with support.captured_output("stderr") as s:
+        with support.captured_stderr() as s:
             print("foo", file=sys.stderr)
         self.assertEqual(s.getvalue(), "foo\n")
 

Lib/test/test_itertools.py

@@ -879,7 +879,7 @@ def __next__(self):
         def run(r1, r2):
             result = []
             for i, j in zip_longest(r1, r2, fillvalue=0):
-                with support.captured_output('stdout'):
+                with support.captured_stdout():
                     print((i, j))
                 result.append((i, j))
             return result

Lib/test/test_pdb.py

@@ -17,7 +17,7 @@
 import zipfile
 
 from asyncio.events import _set_event_loop_policy
-from contextlib import ExitStack, redirect_stdout
+from contextlib import ExitStack
 from io import StringIO
 from test import support
 from test.support import has_socket_support, os_helper
@@ -4571,7 +4571,7 @@ def test_checkline_is_not_executable(self):
         with open(os_helper.TESTFN, "w") as f:
             f.write(s)
         num_lines = len(s.splitlines()) + 2  # Test for EOF
-        with redirect_stdout(StringIO()):
+        with support.captured_stdout():
             db = pdb.Pdb()
             for lineno in range(num_lines):
                 self.assertFalse(db.checkline(os_helper.TESTFN, lineno))