From e948314a3edb2bd4bcf1502a7af38a302168d2cd Mon Sep 17 00:00:00 2001 From: CPython Developers <> Date: Thu, 5 Feb 2026 20:46:35 +0900 Subject: [PATCH] Update test_io from v3.14.3 --- Lib/test/test_io.py | 325 +++++++++++++++++++++----------------------- 1 file changed, 154 insertions(+), 171 deletions(-) diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py index ba9360200..08cc3f655 100644 --- a/Lib/test/test_io.py +++ b/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":5: EncodingWarning: ")) - self.assertTrue( - warnings[1].startswith(b":8: EncodingWarning: ")) + self.assertStartsWith(warnings[0], b":5: EncodingWarning: ") + self.assertStartsWith(warnings[1], b":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