Update test_io from v3.14.3

Lib/test/test_io.py

@@ -445,9 +445,25 @@ class IOTest(unittest.TestCase):
             self.assertRaises(exc, fp.seek, 1, self.SEEK_CUR)
             self.assertRaises(exc, fp.seek, -1, self.SEEK_END)
 
-    @unittest.skipIf(
-        support.is_emscripten, "fstat() of a pipe fd is not supported"
-    )
+    @support.cpython_only
+    def test_startup_optimization(self):
+        # gh-132952: Test that `io` is not imported at startup and that the
+        # __module__ of UnsupportedOperation is set to "io".
+        assert_python_ok("-S", "-c", textwrap.dedent(
+            """
+            import sys
+            assert "io" not in sys.modules
+            try:
+                sys.stdin.truncate()
+            except Exception as e:
+                typ = type(e)
+                assert typ.__module__ == "io", (typ, typ.__module__)
+                assert typ.__name__ == "UnsupportedOperation", (typ, typ.__name__)
+            else:
+                raise AssertionError("Expected UnsupportedOperation")
+            """
+        ))
+
     @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
     def test_optional_abilities(self):
         # Test for OSError when optional APIs are not supported
@@ -501,57 +517,65 @@ class IOTest(unittest.TestCase):
             (text_reader, "r"), (text_writer, "w"),
             (self.BytesIO, "rws"), (self.StringIO, "rws"),
         )
+
+        def do_test(test, obj, abilities):
+            readable = "r" in abilities
+            self.assertEqual(obj.readable(), readable)
+            writable = "w" in abilities
+            self.assertEqual(obj.writable(), writable)
+
+            if isinstance(obj, self.TextIOBase):
+                data = "3"
+            elif isinstance(obj, (self.BufferedIOBase, self.RawIOBase)):
+                data = b"3"
+            else:
+                self.fail("Unknown base class")
+
+            if "f" in abilities:
+                obj.fileno()
+            else:
+                self.assertRaises(OSError, obj.fileno)
+
+            if readable:
+                obj.read(1)
+                obj.read()
+            else:
+                self.assertRaises(OSError, obj.read, 1)
+                self.assertRaises(OSError, obj.read)
+
+            if writable:
+                obj.write(data)
+            else:
+                self.assertRaises(OSError, obj.write, data)
+
+            if sys.platform.startswith("win") and test in (
+                    pipe_reader, pipe_writer):
+                # Pipes seem to appear as seekable on Windows
+                return
+            seekable = "s" in abilities
+            self.assertEqual(obj.seekable(), seekable)
+
+            if seekable:
+                obj.tell()
+                obj.seek(0)
+            else:
+                self.assertRaises(OSError, obj.tell)
+                self.assertRaises(OSError, obj.seek, 0)
+
+            if writable and seekable:
+                obj.truncate()
+                obj.truncate(0)
+            else:
+                self.assertRaises(OSError, obj.truncate)
+                self.assertRaises(OSError, obj.truncate, 0)
+
         for [test, abilities] in tests:
-            with self.subTest(test), test() as obj:
-                readable = "r" in abilities
-                self.assertEqual(obj.readable(), readable)
-                writable = "w" in abilities
-                self.assertEqual(obj.writable(), writable)
+            with self.subTest(test):
+                if test == pipe_writer and not threading_helper.can_start_thread:
+                    self.skipTest("Need threads")
+                with test() as obj:
+                    do_test(test, obj, abilities)
 
-                if isinstance(obj, self.TextIOBase):
-                    data = "3"
-                elif isinstance(obj, (self.BufferedIOBase, self.RawIOBase)):
-                    data = b"3"
-                else:
-                    self.fail("Unknown base class")
-
-                if "f" in abilities:
-                    obj.fileno()
-                else:
-                    self.assertRaises(OSError, obj.fileno)
-
-                if readable:
-                    obj.read(1)
-                    obj.read()
-                else:
-                    self.assertRaises(OSError, obj.read, 1)
-                    self.assertRaises(OSError, obj.read)
-
-                if writable:
-                    obj.write(data)
-                else:
-                    self.assertRaises(OSError, obj.write, data)
-
-                if sys.platform.startswith("win") and test in (
-                        pipe_reader, pipe_writer):
-                    # Pipes seem to appear as seekable on Windows
-                    continue
-                seekable = "s" in abilities
-                self.assertEqual(obj.seekable(), seekable)
-
-                if seekable:
-                    obj.tell()
-                    obj.seek(0)
-                else:
-                    self.assertRaises(OSError, obj.tell)
-                    self.assertRaises(OSError, obj.seek, 0)
-
-                if writable and seekable:
-                    obj.truncate()
-                    obj.truncate(0)
-                else:
-                    self.assertRaises(OSError, obj.truncate)
-                    self.assertRaises(OSError, obj.truncate, 0)
 
     def test_open_handles_NUL_chars(self):
         fn_with_NUL = 'foo\0bar'
@@ -781,7 +805,7 @@ class IOTest(unittest.TestCase):
             self.assertEqual(file.buffer.raw.closefd, False)
 
     @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: filter ('', ResourceWarning) did not catch any warning
-    @unittest.skipIf(sys.platform == 'win32', 'TODO: RUSTPYTHON; cyclic GC not supported, causes file locking')
+    @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; cyclic GC not supported, causes file locking")
     def test_garbage_collection(self):
         # FileIO objects are collected, and collecting them flushes
         # all data to disk.
@@ -896,7 +920,7 @@ class IOTest(unittest.TestCase):
             self.BytesIO()
         )
         for obj in test:
-            self.assertTrue(hasattr(obj, "__dict__"))
+            self.assertHasAttr(obj, "__dict__")
 
     def test_opener(self):
         with self.open(os_helper.TESTFN, "w", encoding="utf-8") as f:
@@ -1090,7 +1114,7 @@ class IOTest(unittest.TestCase):
 
 class CIOTest(IOTest):
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON; cyclic gc
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; cyclic gc
     def test_IOBase_finalize(self):
         # Issue #12149: segmentation fault on _PyIOBase_finalize when both a
         # class which inherits IOBase and an object of this class are caught
@@ -1109,9 +1133,6 @@ class CIOTest(IOTest):
         support.gc_collect()
         self.assertIsNone(wr(), wr)
 
-    def test_destructor(self):
-        return super().test_destructor()
-
 @support.cpython_only
 class TestIOCTypes(unittest.TestCase):
     def setUp(self):
@@ -1146,7 +1167,7 @@ class TestIOCTypes(unittest.TestCase):
         def check_subs(types, base):
             for tp in types:
                 with self.subTest(tp=tp, base=base):
-                    self.assertTrue(issubclass(tp, base))
+                    self.assertIsSubclass(tp, base)
 
         def recursive_check(d):
             for k, v in d.items():
@@ -1803,7 +1824,7 @@ class CBufferedReaderTest(BufferedReaderTest, SizeofTest):
         self.assertRaises(OSError, bufio.read, 10)
 
     @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: filter ('', ResourceWarning) did not catch any warning
-    @unittest.skipIf(sys.platform == 'win32', 'TODO: RUSTPYTHON; cyclic GC not supported, causes file locking')
+    @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; cyclic GC not supported, causes file locking")
     def test_garbage_collection(self):
         # C BufferedReader objects are collected.
         # The Python version has __del__, so it ends into gc.garbage instead
@@ -1838,12 +1859,6 @@ class CBufferedReaderTest(BufferedReaderTest, SizeofTest):
             bufio.readline()
         self.assertIsInstance(cm.exception.__cause__, TypeError)
 
-    def test_pickling_subclass(self):
-        return super().test_pickling_subclass()
-
-    def test_error_through_destructor(self):
-        return super().test_error_through_destructor()
-
 
 class PyBufferedReaderTest(BufferedReaderTest):
     tp = pyio.BufferedReader
@@ -1907,7 +1922,7 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
         flushed = b"".join(writer._write_stack)
         # At least (total - 8) bytes were implicitly flushed, perhaps more
         # depending on the implementation.
-        self.assertTrue(flushed.startswith(contents[:-8]), flushed)
+        self.assertStartsWith(flushed, contents[:-8])
 
     def check_writes(self, intermediate_func):
         # Lots of writes, test the flushed output is as expected.
@@ -1977,7 +1992,7 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
         self.assertEqual(bufio.write(b"ABCDEFGHI"), 9)
         s = raw.pop_written()
         # Previously buffered bytes were flushed
-        self.assertTrue(s.startswith(b"01234567A"), s)
+        self.assertStartsWith(s, b"01234567A")
 
     def test_write_and_rewind(self):
         raw = self.BytesIO()
@@ -2159,7 +2174,7 @@ class CBufferedWriterTest(BufferedWriterTest, SizeofTest):
         self.assertRaises(ValueError, bufio.write, b"def")
 
     @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: filter ('', ResourceWarning) did not catch any warning
-    @unittest.skipIf(sys.platform == 'win32', 'TODO: RUSTPYTHON; cyclic GC not supported, causes file locking')
+    @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; cyclic GC not supported, causes file locking")
     def test_garbage_collection(self):
         # C BufferedWriter objects are collected, and collecting them flushes
         # all data to disk.
@@ -2182,11 +2197,6 @@ class CBufferedWriterTest(BufferedWriterTest, SizeofTest):
         with self.assertRaisesRegex(TypeError, "BufferedWriter"):
             self.tp(self.BytesIO(), 1024, 1024, 1024)
 
-    def test_pickling_subclass(self):
-        return super().test_pickling_subclass()
-
-    def test_error_through_destructor(self):
-        return super().test_error_through_destructor()
 
 class PyBufferedWriterTest(BufferedWriterTest):
     tp = pyio.BufferedWriter
@@ -2280,7 +2290,7 @@ class BufferedRWPairTest(unittest.TestCase):
     def test_peek(self):
         pair = self.tp(self.BytesIO(b"abcdef"), self.MockRawIO())
 
-        self.assertTrue(pair.peek(3).startswith(b"abc"))
+        self.assertStartsWith(pair.peek(3), b"abc")
         self.assertEqual(pair.read(3), b"abc")
 
     def test_readable(self):
@@ -2665,7 +2675,7 @@ class CBufferedRandomTest(BufferedRandomTest, SizeofTest):
     tp = io.BufferedRandom
 
     @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: filter ('', ResourceWarning) did not catch any warning
-    @unittest.skipIf(sys.platform == 'win32', 'TODO: RUSTPYTHON; cyclic GC not supported, causes file locking')
+    @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; cyclic GC not supported, causes file locking")
     def test_garbage_collection(self):
         CBufferedReaderTest.test_garbage_collection(self)
         CBufferedWriterTest.test_garbage_collection(self)
@@ -2675,12 +2685,6 @@ class CBufferedRandomTest(BufferedRandomTest, SizeofTest):
         with self.assertRaisesRegex(TypeError, "BufferedRandom"):
             self.tp(self.BytesIO(), 1024, 1024, 1024)
 
-    def test_pickling_subclass(self):
-        return super().test_pickling_subclass()
-
-    def test_error_through_destructor(self):
-        return super().test_error_through_destructor()
-
 
 class PyBufferedRandomTest(BufferedRandomTest):
     tp = pyio.BufferedRandom
@@ -2990,14 +2994,11 @@ class TextIOWrapperTest(unittest.TestCase):
 
     @unittest.skipIf(sys.flags.utf8_mode, "utf-8 mode is enabled")
     def test_default_encoding(self):
-        old_environ = dict(os.environ)
-        try:
+        with os_helper.EnvironmentVarGuard() as env:
             # try to get a user preferred encoding different than the current
             # locale encoding to check that TextIOWrapper() uses the current
             # locale encoding and not the user preferred encoding
-            for key in ('LC_ALL', 'LANG', 'LC_CTYPE'):
-                if key in os.environ:
-                    del os.environ[key]
+            env.unset('LC_ALL', 'LANG', 'LC_CTYPE')
 
             current_locale_encoding = locale.getencoding()
             b = self.BytesIO()
@@ -3005,9 +3006,6 @@ class TextIOWrapperTest(unittest.TestCase):
                 warnings.simplefilter("ignore", EncodingWarning)
                 t = self.TextIOWrapper(b)
             self.assertEqual(t.encoding, current_locale_encoding)
-        finally:
-            os.environ.clear()
-            os.environ.update(old_environ)
 
     def test_encoding(self):
         # Check the encoding attribute is always set, and valid
@@ -4073,6 +4071,22 @@ class TextIOWrapperTest(unittest.TestCase):
                 self.assertEqual(newtxt.tag, 'ham')
         del MyTextIO
 
+    @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
+    def test_read_non_blocking(self):
+        import os
+        r, w = os.pipe()
+        try:
+            os.set_blocking(r, False)
+            with self.io.open(r, 'rt') as textfile:
+                r = None
+                # Nothing has been written so a non-blocking read raises a BlockingIOError exception.
+                with self.assertRaises(BlockingIOError):
+                    textfile.read()
+        finally:
+            if r is not None:
+                os.close(r)
+            os.close(w)
+
 
 class MemviewBytesIO(io.BytesIO):
     '''A BytesIO object whose read method returns memoryviews
@@ -4108,7 +4122,7 @@ class CTextIOWrapperTest(TextIOWrapperTest):
         self.assertRaises(Exception, repr, t)
 
     @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: filter ('', ResourceWarning) did not catch any warning
-    @unittest.skipIf(sys.platform == 'win32', 'TODO: RUSTPYTHON; cyclic GC not supported, causes file locking')
+    @unittest.skipIf(sys.platform == "win32", "TODO: RUSTPYTHON; cyclic GC not supported, causes file locking")
     def test_garbage_collection(self):
         # C TextIOWrapper objects are collected, and collecting them flushes
         # all data to disk.
@@ -4194,66 +4208,32 @@ class CTextIOWrapperTest(TextIOWrapperTest):
         self.assertEqual([b"abcdef", b"middle", b"g"*chunk_size],
                          buf._write_stack)
 
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; AttributeError: 'NoneType' object has no attribute 'closed'
+    def test_issue142594(self):
+        wrapper = None
+        detached = False
+        class ReentrantRawIO(self.RawIOBase):
+            @property
+            def closed(self):
+                nonlocal detached
+                if wrapper is not None and not detached:
+                    detached = True
+                    wrapper.detach()
+                return False
+
+        raw = ReentrantRawIO()
+        wrapper = self.TextIOWrapper(raw)
+        wrapper.close()  # should not crash
+
     @unittest.expectedFailure  # TODO: RUSTPYTHON; LookupError: unknown encoding: euc_jis_2004
     def test_seek_with_encoder_state(self):
         return super().test_seek_with_encoder_state()
 
-    def test_pickling_subclass(self):
-        return super().test_pickling_subclass()
-
-    def test_reconfigure_newline(self):
-        return super().test_reconfigure_newline()
-
-    def test_newlines_input(self):
-        return super().test_newlines_input()
-
-    def test_reconfigure_defaults(self):
-        return super().test_reconfigure_defaults()
-
-    def test_non_text_encoding_codecs_are_rejected(self):
-        return super().test_non_text_encoding_codecs_are_rejected()
-
-    def test_repr(self):
-        return super().test_repr()
-
-    def test_recursive_repr(self):
-        return super().test_recursive_repr()
-
-    def test_reconfigure_errors(self):
-        return super().test_reconfigure_errors()
-
-    def test_reconfigure_encoding_read(self):
-        return super().test_reconfigure_encoding_read()
-
-    def test_reconfigure_write_through(self):
-        return super().test_reconfigure_write_through()
-
-    def test_reconfigure_line_buffering(self):
-        return super().test_reconfigure_line_buffering()
-
-    def test_reconfigure_write(self):
-        return super().test_reconfigure_write()
-
-    def test_append_bom(self):
-        return super().test_append_bom()
-
-    def test_reconfigure_write_fromascii(self):
-        return super().test_reconfigure_write_fromascii()
-
-    def test_error_through_destructor(self):
-        return super().test_error_through_destructor()
-
-    def test_reconfigure_locale(self):
-        return super().test_reconfigure_locale()
-
 
 class PyTextIOWrapperTest(TextIOWrapperTest):
     io = pyio
     shutdown_error = "LookupError: unknown encoding: ascii"
 
-    def test_constructor(self):
-        return super().test_constructor()
-
     @unittest.expectedFailure  # TODO: RUSTPYTHON; LookupError: unknown encoding: euc_jis_2004
     def test_seek_with_encoder_state(self):
         return super().test_seek_with_encoder_state()
@@ -4433,9 +4413,6 @@ class MiscIOTest(unittest.TestCase):
                 self.open(os_helper.TESTFN, mode)
             self.assertIn('invalid mode', str(cm.exception))
 
-    @unittest.skipIf(
-        support.is_emscripten, "fstat() of a pipe fd is not supported"
-    )
     @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
     def test_open_pipe_with_append(self):
         # bpo-27805: Ignore ESPIPE from lseek() in open().
@@ -4497,7 +4474,7 @@ class MiscIOTest(unittest.TestCase):
             self.assertRaises(ValueError, f.writelines, [])
             self.assertRaises(ValueError, next, f)
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON; cyclic gc
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; cyclic gc
     def test_blockingioerror(self):
         # Various BlockingIOError issues
         class C(str):
@@ -4605,15 +4582,11 @@ class MiscIOTest(unittest.TestCase):
                         with self.assertRaisesRegex(TypeError, msg):
                             pickle.dumps(f, protocol)
 
-    @unittest.skipIf(
-        support.is_emscripten, "fstat() of a pipe fd is not supported"
-    )
+    @unittest.skipIf(support.is_emscripten, "Emscripten corrupts memory when writing to nonblocking fd")
     def test_nonblock_pipe_write_bigbuf(self):
         self._test_nonblock_pipe_write(16*1024)
 
-    @unittest.skipIf(
-        support.is_emscripten, "fstat() of a pipe fd is not supported"
-    )
+    @unittest.skipIf(support.is_emscripten, "Emscripten corrupts memory when writing to nonblocking fd")
     def test_nonblock_pipe_write_smallbuf(self):
         self._test_nonblock_pipe_write(1024)
 
@@ -4750,10 +4723,8 @@ class MiscIOTest(unittest.TestCase):
         proc = assert_python_ok('-X', 'warn_default_encoding', '-c', code)
         warnings = proc.err.splitlines()
         self.assertEqual(len(warnings), 2)
-        self.assertTrue(
-            warnings[0].startswith(b"<string>:5: EncodingWarning: "))
-        self.assertTrue(
-            warnings[1].startswith(b"<string>:8: EncodingWarning: "))
+        self.assertStartsWith(warnings[0], b"<string>:5: EncodingWarning: ")
+        self.assertStartsWith(warnings[1], b"<string>:8: EncodingWarning: ")
 
     def test_text_encoding(self):
         # PEP 597, bpo-47000. io.text_encoding() returns "locale" or "utf-8"
@@ -4834,15 +4805,6 @@ class CMiscIOTest(MiscIOTest):
     def test_daemon_threads_shutdown_stderr_deadlock(self):
         self.check_daemon_threads_shutdown_deadlock('stderr')
 
-    def test_check_encoding_errors(self):
-        return super().test_check_encoding_errors()
-
-    def test_warn_on_dealloc(self):
-        return super().test_warn_on_dealloc()
-
-    def test_warn_on_dealloc_fd(self):
-        return super().test_warn_on_dealloc_fd()
-
 
 class PyMiscIOTest(MiscIOTest):
     io = pyio
@@ -4977,7 +4939,7 @@ class SignalsTest(unittest.TestCase):
                     os.read(r, len(data) * 100)
             exc = cm.exception
             if isinstance(exc, RuntimeError):
-                self.assertTrue(str(exc).startswith("reentrant call"), str(exc))
+                self.assertStartsWith(str(exc), "reentrant call")
         finally:
             signal.alarm(0)
             wio.close()
@@ -5095,13 +5057,13 @@ class SignalsTest(unittest.TestCase):
                 if e.errno != errno.EBADF:
                     raise
 
-    @unittest.skip("TODO: RUSTPYTHON thread 'main' (103833) panicked at crates/vm/src/stdlib/signal.rs:233:43: RefCell already borrowed")
+    @unittest.skip("TODO: RUSTPYTHON; thread 'main' (103833) panicked at crates/vm/src/stdlib/signal.rs:233:43: RefCell already borrowed")
     @requires_alarm
     @support.requires_resource('walltime')
     def test_interrupted_write_retry_buffered(self):
         self.check_interrupted_write_retry(b"x", mode="wb")
 
-    @unittest.skip("TODO: RUSTPYTHON thread 'main' (103833) panicked at crates/vm/src/stdlib/signal.rs:233:43: RefCell already borrowed")
+    @unittest.skip("TODO: RUSTPYTHON; thread 'main' (103833) panicked at crates/vm/src/stdlib/signal.rs:233:43: RefCell already borrowed")
     @requires_alarm
     @support.requires_resource('walltime')
     def test_interrupted_write_retry_text(self):
@@ -5111,9 +5073,9 @@ class SignalsTest(unittest.TestCase):
 class CSignalsTest(SignalsTest):
     io = io
 
-    @unittest.skip("TODO: RUSTPYTHON thread 'main' (103833) panicked at crates/vm/src/stdlib/signal.rs:233:43: RefCell already borrowed")
-    def test_interrupted_read_retry_buffered(self):  # TODO: RUSTPYTHON
-        return super().test_interrupted_read_retry_buffered()  # TODO: RUSTPYTHON
+    @unittest.skip("TODO: RUSTPYTHON; thread 'main' (103833) panicked at crates/vm/src/stdlib/signal.rs:233:43: RefCell already borrowed")
+    def test_interrupted_read_retry_buffered(self):
+        return super().test_interrupted_read_retry_buffered()
 
 class PySignalsTest(SignalsTest):
     io = pyio
@@ -5124,6 +5086,26 @@ class PySignalsTest(SignalsTest):
     test_reentrant_write_text = None
 
 
+class ProtocolsTest(unittest.TestCase):
+    class MyReader:
+        def read(self, sz=-1):
+            return b""
+
+    class MyWriter:
+        def write(self, b: bytes):
+            pass
+
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; AttributeError: module 'io' has no attribute 'Reader'
+    def test_reader_subclass(self):
+        self.assertIsSubclass(self.MyReader, io.Reader)
+        self.assertNotIsSubclass(str, io.Reader)
+
+    @unittest.expectedFailure  # TODO: RUSTPYTHON; AttributeError: module 'io' has no attribute 'Writer'
+    def test_writer_subclass(self):
+        self.assertIsSubclass(self.MyWriter, io.Writer)
+        self.assertNotIsSubclass(str, io.Writer)
+
+
 def load_tests(loader, tests, pattern):
     tests = (CIOTest, PyIOTest, APIMismatchTest,
              CBufferedReaderTest, PyBufferedReaderTest,
@@ -5135,6 +5117,7 @@ def load_tests(loader, tests, pattern):
              CTextIOWrapperTest, PyTextIOWrapperTest,
              CMiscIOTest, PyMiscIOTest,
              CSignalsTest, PySignalsTest, TestIOCTypes,
+             ProtocolsTest,
              )
 
     # Put the namespaces of the IO module we are testing and some useful mock