diff options
Diffstat (limited to 'Lib/test/test_io.py')
| -rw-r--r-- | Lib/test/test_io.py | 117 |
1 files changed, 57 insertions, 60 deletions
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py index 455a87b..ac3b6d1 100644 --- a/Lib/test/test_io.py +++ b/Lib/test/test_io.py @@ -445,9 +445,6 @@ 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" - ) @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()") def test_optional_abilities(self): # Test for OSError when optional APIs are not supported @@ -501,57 +498,65 @@ class IOTest(unittest.TestCase): (text_reader, "r"), (text_writer, "w"), (self.BytesIO, "rws"), (self.StringIO, "rws"), ) - 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) - - 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) + 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 readable: - obj.read(1) - obj.read() - else: - self.assertRaises(OSError, obj.read, 1) - self.assertRaises(OSError, obj.read) + if isinstance(obj, self.TextIOBase): + data = "3" + elif isinstance(obj, (self.BufferedIOBase, self.RawIOBase)): + data = b"3" + else: + self.fail("Unknown base class") - 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 "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): + if test == pipe_writer and not threading_helper.can_start_thread: + skipTest() + with test() as obj: + do_test(test, obj, abilities) - 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' @@ -3927,7 +3932,6 @@ class TextIOWrapperTest(unittest.TestCase): self.assertEqual(res + f.readline(), 'foo\nbar\n') @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()") - @unittest.skipIf(support.is_emscripten, "Fixed in next Emscripten release after 4.0.1") def test_read_non_blocking(self): import os r, w = os.pipe() @@ -4242,9 +4246,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(). @@ -4413,15 +4414,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) |