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
(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'
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()
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().
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)
self.assertRaises((ValueError, OSError),
signal.set_wakeup_fd, fd)
- @unittest.skipIf(support.is_emscripten, "Fixed in next Emscripten release after 4.0.1")
@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
def test_set_wakeup_fd_result(self):
r1, w1 = os.pipe()
self.assertEqual(signal.set_wakeup_fd(-1), w2)
self.assertEqual(signal.set_wakeup_fd(-1), -1)
- @unittest.skipIf(support.is_emscripten, "Fixed in next Emscripten release after 4.0.1")
@unittest.skipUnless(support.has_socket_support, "needs working sockets.")
def test_set_wakeup_fd_socket_result(self):
sock1 = socket.socket()
# On Windows, files are always blocking and Windows does not provide a
# function to test if a socket is in non-blocking mode.
@unittest.skipIf(sys.platform == "win32", "tests specific to POSIX")
- @unittest.skipIf(support.is_emscripten, "Fixed in next Emscripten release after 4.0.1")
@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
def test_set_wakeup_fd_blocking(self):
rfd, wfd = os.pipe()