pythongh-138013: Remove test_io load_tests namespace manipulation (python#138366)

Reduce what happens in `load_tests` so that the next change,
moving the `Buffered*` tests to `test_bufferdio` is purely mechanical
movement and updating imports.

This adds two classes, one per I/O implementation, to act as dispatch to
the implementation-specific mocks as well as module members. Previously
the mappings CTestCase and PyTestCase provide were injected directly
during `load_tests`.

CTestCase and PyTestCase inherit from `unittest.TestCase` so when the
split happens default test discovery will work for the classes in
`test_bufferedio`. `test_general` keeps a manual test list for this
refactoring; some of the tests (ex. `ProtocolsTest`) aren't currently
run and fixing that + helpers to not be picked up is out of my current
scope.

CTestCase and PyTestCase have an `io` class member which points to the
implementation meaning that can be removed from individual test cases
which now inherit from them.

This code is picking up `MockRawIO` which is defined globally in the
module but these should use the mock specific to the I/O implementation
being tested.

Co-authored-by: Victor Stinner <[email protected]>

Author: 2 people

Lib/test/test_io/test_general.py

@@ -317,6 +317,45 @@ class PyMockNonBlockWriterIO(MockNonBlockWriterIO, pyio.RawIOBase):
     BlockingIOError = pyio.BlockingIOError
 
 
+# Build classes which point to all the right mocks per io implementation
+class CTestCase(unittest.TestCase):
+    io = io
+    is_C = True
+
+    MockRawIO = CMockRawIO
+    MisbehavedRawIO = CMisbehavedRawIO
+    MockFileIO = CMockFileIO
+    CloseFailureIO = CCloseFailureIO
+    MockNonBlockWriterIO = CMockNonBlockWriterIO
+    MockUnseekableIO = CMockUnseekableIO
+    MockRawIOWithoutRead = CMockRawIOWithoutRead
+    SlowFlushRawIO = CSlowFlushRawIO
+    MockCharPseudoDevFileIO = CMockCharPseudoDevFileIO
+
+    # Use the class as a proxy to the io module members.
+    def __getattr__(self, name):
+        return getattr(io, name)
+
+
+class PyTestCase(unittest.TestCase):
+    io = pyio
+    is_C = False
+
+    MockRawIO = PyMockRawIO
+    MisbehavedRawIO = PyMisbehavedRawIO
+    MockFileIO = PyMockFileIO
+    CloseFailureIO = PyCloseFailureIO
+    MockNonBlockWriterIO = PyMockNonBlockWriterIO
+    MockUnseekableIO = PyMockUnseekableIO
+    MockRawIOWithoutRead = PyMockRawIOWithoutRead
+    SlowFlushRawIO = PySlowFlushRawIO
+    MockCharPseudoDevFileIO = PyMockCharPseudoDevFileIO
+
+    # Use the class as a proxy to the _pyio module members.
+    def __getattr__(self, name):
+        return getattr(pyio, name)
+
+
 class IOTest(unittest.TestCase):
 
     def setUp(self):
@@ -1083,7 +1122,7 @@ def reader(file, barrier):
                          write_count * thread_count)
 
 
-class CIOTest(IOTest):
+class CIOTest(IOTest, CTestCase):
 
     def test_IOBase_finalize(self):
         # Issue #12149: segmentation fault on _PyIOBase_finalize when both a
@@ -1207,7 +1246,7 @@ def test_stringio_setstate(self):
         obj.__setstate__(('', '', 0, {}))
         self.assertEqual(obj.getvalue(), '')
 
-class PyIOTest(IOTest):
+class PyIOTest(IOTest, PyTestCase):
     pass
 
 
@@ -1438,7 +1477,7 @@ def test_buffer_freeing(self) :
         bufio.close()
         self.assertEqual(sys.getsizeof(bufio), size)
 
-class BufferedReaderTest(unittest.TestCase, CommonBufferedTests):
+class BufferedReaderTest(CommonBufferedTests):
     read_mode = "rb"
 
     def test_constructor(self):
@@ -1773,7 +1812,7 @@ def test_seek_character_device_file(self):
         self.assertEqual(buf.seek(0, io.SEEK_CUR), 0)
 
 
-class CBufferedReaderTest(BufferedReaderTest, SizeofTest):
+class CBufferedReaderTest(BufferedReaderTest, SizeofTest, CTestCase):
     tp = io.BufferedReader
 
     def test_initialization(self):
@@ -1832,11 +1871,11 @@ def test_bad_readinto_type(self):
         self.assertIsInstance(cm.exception.__cause__, TypeError)
 
 
-class PyBufferedReaderTest(BufferedReaderTest):
+class PyBufferedReaderTest(BufferedReaderTest, PyTestCase):
     tp = pyio.BufferedReader
 
 
-class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
+class BufferedWriterTest(CommonBufferedTests):
     write_mode = "wb"
 
     def test_constructor(self):
@@ -2131,8 +2170,7 @@ def test_slow_close_from_thread(self):
         t.join()
 
 
-
-class CBufferedWriterTest(BufferedWriterTest, SizeofTest):
+class CBufferedWriterTest(BufferedWriterTest, SizeofTest, CTestCase):
     tp = io.BufferedWriter
 
     def test_initialization(self):
@@ -2172,10 +2210,10 @@ def test_args_error(self):
             self.tp(self.BytesIO(), 1024, 1024, 1024)
 
 
-class PyBufferedWriterTest(BufferedWriterTest):
+class PyBufferedWriterTest(BufferedWriterTest, PyTestCase):
     tp = pyio.BufferedWriter
 
-class BufferedRWPairTest(unittest.TestCase):
+class BufferedRWPairTest:
 
     def test_constructor(self):
         pair = self.tp(self.MockRawIO(), self.MockRawIO())
@@ -2204,14 +2242,14 @@ def test_constructor_max_buffer_size_removal(self):
             self.tp(self.MockRawIO(), self.MockRawIO(), 8, 12)
 
     def test_constructor_with_not_readable(self):
-        class NotReadable(MockRawIO):
+        class NotReadable(self.MockRawIO):
             def readable(self):
                 return False
 
         self.assertRaises(OSError, self.tp, NotReadable(), self.MockRawIO())
 
     def test_constructor_with_not_writeable(self):
-        class NotWriteable(MockRawIO):
+        class NotWriteable(self.MockRawIO):
             def writable(self):
                 return False
 
@@ -2357,9 +2395,9 @@ def writer_close():
         writer.close = lambda: None
 
     def test_isatty(self):
-        class SelectableIsAtty(MockRawIO):
+        class SelectableIsAtty(self.MockRawIO):
             def __init__(self, isatty):
-                MockRawIO.__init__(self)
+                super().__init__()
                 self._isatty = isatty
 
             def isatty(self):
@@ -2383,10 +2421,10 @@ def test_weakref_clearing(self):
         brw = None
         ref = None # Shouldn't segfault.
 
-class CBufferedRWPairTest(BufferedRWPairTest):
+class CBufferedRWPairTest(BufferedRWPairTest, CTestCase):
     tp = io.BufferedRWPair
 
-class PyBufferedRWPairTest(BufferedRWPairTest):
+class PyBufferedRWPairTest(BufferedRWPairTest, PyTestCase):
     tp = pyio.BufferedRWPair
 
 
@@ -2645,7 +2683,7 @@ def test_interleaved_readline_write(self):
     test_truncate_on_read_only = None
 
 
-class CBufferedRandomTest(BufferedRandomTest, SizeofTest):
+class CBufferedRandomTest(BufferedRandomTest, SizeofTest, CTestCase):
     tp = io.BufferedRandom
 
     def test_garbage_collection(self):
@@ -2658,7 +2696,7 @@ def test_args_error(self):
             self.tp(self.BytesIO(), 1024, 1024, 1024)
 
 
-class PyBufferedRandomTest(BufferedRandomTest):
+class PyBufferedRandomTest(BufferedRandomTest, PyTestCase):
     tp = pyio.BufferedRandom
 
 
@@ -4058,8 +4096,7 @@ def _to_memoryview(buf):
     return memoryview(arr)
 
 
-class CTextIOWrapperTest(TextIOWrapperTest):
-    io = io
+class CTextIOWrapperTest(TextIOWrapperTest, CTestCase):
     shutdown_error = "LookupError: unknown encoding: ascii"
 
     def test_initialization(self):
@@ -4159,8 +4196,7 @@ def write(self, data):
                          buf._write_stack)
 
 
-class PyTextIOWrapperTest(TextIOWrapperTest):
-    io = pyio
+class PyTextIOWrapperTest(TextIOWrapperTest, PyTestCase):
     shutdown_error = "LookupError: unknown encoding: ascii"
 
 
@@ -4282,6 +4318,8 @@ def test_translate(self):
         self.assertEqual(decoder.decode(b"\r\r\n"), "\r\r\n")
 
 class CIncrementalNewlineDecoderTest(IncrementalNewlineDecoderTest):
+    IncrementalNewlineDecoder = io.IncrementalNewlineDecoder
+
     @support.cpython_only
     def test_uninitialized(self):
         uninitialized = self.IncrementalNewlineDecoder.__new__(
@@ -4293,7 +4331,7 @@ def test_uninitialized(self):
 
 
 class PyIncrementalNewlineDecoderTest(IncrementalNewlineDecoderTest):
-    pass
+    IncrementalNewlineDecoder = pyio.IncrementalNewlineDecoder
 
 
 # XXX Tests for open()
@@ -4662,8 +4700,7 @@ def test_text_encoding(self):
         self.assertEqual(b"utf-8", proc.out.strip())
 
 
-class CMiscIOTest(MiscIOTest):
-    io = io
+class CMiscIOTest(MiscIOTest, CTestCase):
     name_of_module = "io", "_io"
     extra_exported = "BlockingIOError",
 
@@ -4728,8 +4765,7 @@ def test_daemon_threads_shutdown_stderr_deadlock(self):
         self.check_daemon_threads_shutdown_deadlock('stderr')
 
 
-class PyMiscIOTest(MiscIOTest):
-    io = pyio
+class PyMiscIOTest(MiscIOTest, PyTestCase):
     name_of_module = "_pyio", "io"
     extra_exported = "BlockingIOError", "open_code",
     not_exported = "valid_seek_flags",
@@ -4990,11 +5026,11 @@ def test_interrupted_write_retry_text(self):
         self.check_interrupted_write_retry("x", mode="w", encoding="latin1")
 
 
-class CSignalsTest(SignalsTest):
-    io = io
+class CSignalsTest(SignalsTest, CTestCase):
+    pass
 
-class PySignalsTest(SignalsTest):
-    io = pyio
+class PySignalsTest(SignalsTest, PyTestCase):
+    pass
 
     # Handling reentrancy issues would slow down _pyio even more, so the
     # tests are disabled.
@@ -5034,27 +5070,6 @@ def load_tests(loader, tests, pattern):
              ProtocolsTest,
              )
 
-    # Put the namespaces of the IO module we are testing and some useful mock
-    # classes in the __dict__ of each test.
-    mocks = (MockRawIO, MisbehavedRawIO, MockFileIO, CloseFailureIO,
-             MockNonBlockWriterIO, MockUnseekableIO, MockRawIOWithoutRead,
-             SlowFlushRawIO, MockCharPseudoDevFileIO)
-    all_members = io.__all__
-    c_io_ns = {name : getattr(io, name) for name in all_members}
-    py_io_ns = {name : getattr(pyio, name) for name in all_members}
-    globs = globals()
-    c_io_ns.update((x.__name__, globs["C" + x.__name__]) for x in mocks)
-    py_io_ns.update((x.__name__, globs["Py" + x.__name__]) for x in mocks)
-    for test in tests:
-        if test.__name__.startswith("C"):
-            for name, obj in c_io_ns.items():
-                setattr(test, name, obj)
-            test.is_C = True
-        elif test.__name__.startswith("Py"):
-            for name, obj in py_io_ns.items():
-                setattr(test, name, obj)
-            test.is_C = False
-
     suite = loader.suiteClass()
     for test in tests:
         suite.addTest(loader.loadTestsFromTestCase(test))