diff -r e22d0ff286f9 Lib/_pyio.py --- a/Lib/_pyio.py Sat Jul 05 11:11:09 2014 +0300 +++ b/Lib/_pyio.py Sat Jul 05 23:28:03 2014 +0300 @@ -7,11 +7,16 @@ import codecs import errno import array +import stat # Import _thread instead of threading to reduce startup cost try: from _thread import allocate_lock as Lock except ImportError: from _dummy_thread import allocate_lock as Lock +try: + from ctypes import INT_MAX +except ImportError: + INT_MAX = 2**31 - 1 import io from io import (__all__, SEEK_SET, SEEK_CUR, SEEK_END) @@ -1382,6 +1387,375 @@ return BufferedWriter.write(self, b) +class FileIO(RawIOBase): + _fd = -1 + _created = False + _readable = False + _writable = False + _appending = False + _seekable = None + _closefd = True + + def __init__(self, file, mode='r', closefd=True, opener=None): + """\ +Open a file. The mode can be 'r', 'w', 'x' or 'a' for reading (default), +writing, exclusive creation or appending. The file will be created if it +doesn't exist when opened for writing or appending; it will be truncated +when opened for writing. A `FileExistsError` will be raised if it already +exists when opened for creating. Opening a file for creating implies +writing so this mode behaves in a similar way to 'w'.Add a '+' to the mode +to allow simultaneous reading and writing. A custom opener can be used by +passing a callable as *opener*. The underlying file descriptor for the file +object is then obtained by calling opener with (*name*, *flags*). +*opener* must return an open file descriptor (passing os.open as *opener* +results in functionality similar to passing None).""" + if self._fd >= 0: + # Have to close the existing file first. + try: + if self._closefd: + os.close(self._fd) + finally: + self._fd = -1 + + if isinstance(file, float): + raise TypeError('integer argument expected, got float') + elif isinstance(file, int): + fd = file + if fd < 0: + raise ValueError('Negative filedescriptor') + else: + fd = -1 + #if 0 in os.fsencode(file): + #raise TypeError('embedded NUL character') + + if not isinstance(mode, str): + raise TypeError('invalid mode: %s' % (mode,)) + if not set(mode) <= set('xrwab+'): + raise ValueError('invalid mode: %s' % (mode,)) + if sum(c in 'rwax' for c in mode) != 1 or mode.count('+') > 1: + raise ValueError('Must have exactly one of create/read/write/append ' + 'mode and at most one plus') + + if 'x' in mode: + self._created = True + self._writable = True + flags = os.O_EXCL | os.O_CREAT + elif 'r' in mode: + self._readable = True + flags = 0 + elif 'w' in mode: + self._writable = True + flags = os.O_CREAT | os.O_TRUNC + elif 'a' in mode: + self._writable = True + self._appending = True + flags = os.O_APPEND | os.O_CREAT + + if '+' in mode: + self._readable = True + self._writable = True + + if self._readable and self._writable: + flags |= os.O_RDWR + elif self._readable: + flags |= os.O_RDONLY + else: + flags |= os.O_WRONLY + + binary_flag = getattr(os, 'O_BINARY', 0) + flags |= binary_flag + + noinherit_flag = (getattr(os, 'O_NOINHERIT', 0) or + getattr(os, 'O_CLOEXEC', 0)) + flags |= noinherit_flag + + owned_fd = None + try: + if fd >= 0: + if hasattr(os, 'fstat'): + try: + os.fstat(fd) + except OSError as e: + if e.errno == errno.EBADF: + raise + self._closefd = closefd + else: + self._closefd = True + if not closefd: + raise ValueError('Cannot use closefd=False with file name') + if opener is None: + fd = os.open(file, flags, 0o666) + else: + fd = opener(file, flags) + if not isinstance(fd, int): + raise TypeError('expected integer from opener') + if fd < 0: + raise OSError('Negative file descriptor') + owned_fd = fd + if not noinherit_flag: + os.set_inheritable(fd, False) + + self._blksize = DEFAULT_BUFFER_SIZE + try: + fdfstat = os.fstat(fd) + except (OSError, AttributeError): + pass + else: + try: + if stat.S_ISDIR(fdfstat.st_mode): + raise IsADirectoryError(errno.EISDIR, + os.strerror(errno.EISDIR), file) + except AttributeError: + pass + try: + if fdfstat.st_blksize > 1: + self._blksize = fdfstat.st_blksize; + except AttributeError: + pass + + if binary_flag: + try: + import msvcrt + except ImportError: + pass + else: + # don't translate newlines (\r\n <=> \n) + msvcrt.setmode(fd, os.O_BINARY) + + self.name = file + if self._appending: + # For consistent behaviour, we explicitly seek to the + # end of file (otherwise, it might be done only on the + # first write()). + os.lseek(fd, 0, SEEK_END) + except: + if owned_fd is not None: + os.close(owned_fd) + raise + self._fd = fd + + def __del__(self): + if self._fd >= 0 and self._closefd and not self.closed: + import warnings + warnings.warn('unclosed file %r' % (self.name,), ResourceWarning, + stacklevel=2) + self.close() + + def __getstate__(self): + raise TypeError("cannot serialize '%s' object", self.__class__.__name__) + + def __repr__(self): + class_name = '%s.%s' % (self.__class__.__module__, + self.__class__.__qualname__) + if self.closed: + return '<%s [closed]>' % class_name + try: + name = self.name + except AttributeError: + return'<%s fd=%d mode=%r>' % (class_name, self._fd, self.mode) + else: + return'<%s name=%r mode=%r>' % (class_name, name, self.mode) + + def _checkReadable(self): + if not self._readable: + raise UnsupportedOperation('File not open for reading') + + def _checkWritable(self, msg=None): + if not self._writable: + raise UnsupportedOperation('File not open for writing') + + def read(self, size=-1): + """\ +Read at most size bytes, returned as bytes. + +Only makes one system call, so less data may be returned than requested +In non-blocking mode, returns None if no data is available. +On end-of-file, returns ''.""" + self._checkClosed() + self._checkReadable() + if size is None or size < 0: + return self.readall() + if size > INT_MAX: + # os.read() size is limited to a C int + size = INT_MAX + while True: + try: + return os.read(self._fd, size) + except InterruptedError: + continue + except BlockingIOError: + return None + + def readall(self): + """\ +Read all data from the file, returned as bytes. + +In non-blocking mode, returns as much as is immediately available, +or None if no data is available. On end-of-file, returns ''.""" + self._checkClosed() + self._checkReadable() + bufsize = DEFAULT_BUFFER_SIZE + try: + pos = os.lseek(self._fd, 0, SEEK_CUR) + end = os.fstat(self._fd).st_size + if end >= pos and end - pos < sys.maxsize: + bufsize = end - pos + 1 + except: + pass + + result = b'' + while True: + if len(result) >= bufsize: + bufsize += max(bufsize, DEFAULT_BUFFER_SIZE) + try: + n = bufsize - len(result) + b = os.read(self._fd, min(n, INT_MAX)) + except InterruptedError: + continue + except BlockingIOError: + if result: + break + return None + if not b: + break + result += b + + return result + + def readinto(self, b): + "Same as RawIOBase.readinto()." + data = self.read(len(b)) + n = len(data) + try: + b[:n] = data + except TypeError as err: + import array + if not isinstance(b, array.array): + raise err + b[:n] = array.array('b', data) + return n + + def write(self, b): + """\ +Write bytes b to file, return number written. + +Only makes one system call, so not all of the data may be written. +The number of bytes actually written is returned.""" + self._checkClosed() + self._checkWritable() + try: + return os.write(self._fd, b) + except BlockingIOError: + return None + + def seek(self, pos, whence=SEEK_SET): + """\ +Move to new file position. + +Argument offset is a byte count. Optional argument whence defaults to +0 (offset from start of file, offset should be >= 0); other values are 1 +(move relative to current position, positive or negative), and 2 (move +relative to end of file, usually negative, although many platforms allow +seeking beyond the end of a file). +Note that not all file objects are seekable.""" + if isinstance(pos, float): + raise TypeError('an integer is required') + self._checkClosed() + return os.lseek(self._fd, pos, whence) + + def tell(self): + "tell() -> int. Current file position" + self._checkClosed() + return os.lseek(self._fd, 0, SEEK_CUR) + + def truncate(self, size=None): + """\ +Truncate the file to at most size bytes. + +Size defaults to the current file position, as returned by tell(). +The current file position is changed to the value of size.""" + self._checkClosed() + self._checkWritable() + if size is None: + size = self.tell() + os.ftruncate(self._fd, size) + return size + + def close(self): + """\ +Close the file. + +A closed file cannot be used for further I/O operations. close() may be +called more than once without error. Changes the fileno to -1.""" + if not self.closed: + try: + if self._closefd: + os.close(self._fd) + finally: + super().close() + + def seekable(self): + "True if file supports random-access." + self._checkClosed() + if self._seekable is None: + try: + self.tell() + except OSError: + self._seekable = False + else: + self._seekable = True + return self._seekable + + def readable(self): + "True if file was opened in a read mode." + self._checkClosed() + return self._readable + + def writable(self): + "writable() -> bool. True if file was opened in a write mode." + self._checkClosed() + return self._writable + + def fileno(self): + """\ +"File descriptor". + +This is needed for lower-level file interfaces, such the fcntl module.""" + self._checkClosed() + return self._fd + + def isatty(self): + "True if the file is connected to a tty device." + self._checkClosed() + return os.isatty(self._fd) + + @property + def closefd(self): + "True if the file descriptor will be closed" + return self._closefd + + @property + def mode(self): + "String giving the file mode" + if self._created: + if self._readable: + return 'xb+' + else: + return 'xb' + elif self._appending: + if self._readable: + return 'ab+' + else: + return 'ab' + elif self._readable: + if self._writable: + return 'rb+' + else: + return 'rb' + else: + return 'wb' + + class TextIOBase(IOBase): """Base class for text I/O. diff -r e22d0ff286f9 Lib/test/test_file_eintr.py --- a/Lib/test/test_file_eintr.py Sat Jul 05 11:11:09 2014 +0300 +++ b/Lib/test/test_file_eintr.py Sat Jul 05 23:28:03 2014 +0300 @@ -18,11 +18,12 @@ import unittest # Test import all of the things we're about to try testing up front. -from _io import FileIO +import _io +import _pyio @unittest.skipUnless(os.name == 'posix', 'tests requires a posix system.') -class TestFileIOSignalInterrupt(unittest.TestCase): +class TestFileIOSignalInterrupt: def setUp(self): self._process = None @@ -38,8 +39,9 @@ subclasseses should override this to test different IO objects. """ - return ('import _io ;' - 'infile = _io.FileIO(sys.stdin.fileno(), "rb")') + return ('import %s as io ;' + 'infile = io.FileIO(sys.stdin.fileno(), "rb")' % + self.modname) def fail_with_process_info(self, why, stdout=b'', stderr=b'', communicate=True): @@ -179,11 +181,19 @@ expected=b'hello\nworld!\n')) +class CTestFileIOSignalInterrupt(TestFileIOSignalInterrupt, unittest.TestCase): + modname = '_io' + +class PyTestFileIOSignalInterrupt(TestFileIOSignalInterrupt, unittest.TestCase): + modname = '_pyio' + + class TestBufferedIOSignalInterrupt(TestFileIOSignalInterrupt): def _generate_infile_setup_code(self): """Returns the infile = ... line of code to make a BufferedReader.""" - return ('infile = open(sys.stdin.fileno(), "rb") ;' - 'import _io ;assert isinstance(infile, _io.BufferedReader)') + return ('import %s as io ;infile = io.open(sys.stdin.fileno(), "rb") ;' + 'assert isinstance(infile, io.BufferedReader)' % + self.modname) def test_readall(self): """BufferedReader.read() must handle signals and not lose data.""" @@ -193,12 +203,20 @@ read_method_name='read', expected=b'hello\nworld!\n')) +class CTestBufferedIOSignalInterrupt(TestBufferedIOSignalInterrupt, unittest.TestCase): + modname = '_io' + +class PyTestBufferedIOSignalInterrupt(TestBufferedIOSignalInterrupt, unittest.TestCase): + modname = '_pyio' + class TestTextIOSignalInterrupt(TestFileIOSignalInterrupt): def _generate_infile_setup_code(self): """Returns the infile = ... line of code to make a TextIOWrapper.""" - return ('infile = open(sys.stdin.fileno(), "rt", newline=None) ;' - 'import _io ;assert isinstance(infile, _io.TextIOWrapper)') + return ('import %s as io ;' + 'infile = io.open(sys.stdin.fileno(), "rt", newline=None) ;' + 'assert isinstance(infile, io.TextIOWrapper)' % + self.modname) def test_readline(self): """readline() must handle signals and not lose data.""" @@ -224,6 +242,12 @@ read_method_name='read', expected="hello\nworld!\n")) +class CTestTextIOSignalInterrupt(TestTextIOSignalInterrupt, unittest.TestCase): + modname = '_io' + +class PyTestTextIOSignalInterrupt(TestTextIOSignalInterrupt, unittest.TestCase): + modname = '_pyio' + def test_main(): test_cases = [ diff -r e22d0ff286f9 Lib/test/test_fileio.py --- a/Lib/test/test_fileio.py Sat Jul 05 11:11:09 2014 +0300 +++ b/Lib/test/test_fileio.py Sat Jul 05 23:28:03 2014 +0300 @@ -12,13 +12,15 @@ from test.support import TESTFN, check_warnings, run_unittest, make_bad_fd, cpython_only from collections import UserList -from _io import FileIO as _FileIO +import _io # C implementation of io +import _pyio # Python implementation of io -class AutoFileTests(unittest.TestCase): + +class AutoFileTests: # file tests for which a test file is automatically set up def setUp(self): - self.f = _FileIO(TESTFN, 'w') + self.f = self.FileIO(TESTFN, 'w') def tearDown(self): if self.f: @@ -74,7 +76,7 @@ self.f.write(bytes([1, 2])) self.f.close() a = array('b', b'x'*10) - self.f = _FileIO(TESTFN, 'r') + self.f = self.FileIO(TESTFN, 'r') n = self.f.readinto(a) self.assertEqual(array('b', [1, 2]), a[:n]) @@ -82,7 +84,7 @@ l = [b'123', b'456'] self.f.writelines(l) self.f.close() - self.f = _FileIO(TESTFN, 'rb') + self.f = self.FileIO(TESTFN, 'rb') buf = self.f.read() self.assertEqual(buf, b'123456') @@ -90,7 +92,7 @@ l = UserList([b'123', b'456']) self.f.writelines(l) self.f.close() - self.f = _FileIO(TESTFN, 'rb') + self.f = self.FileIO(TESTFN, 'rb') buf = self.f.read() self.assertEqual(buf, b'123456') @@ -102,7 +104,7 @@ def test_none_args(self): self.f.write(b"hi\nbye\nabc") self.f.close() - self.f = _FileIO(TESTFN, 'r') + self.f = self.FileIO(TESTFN, 'r') self.assertEqual(self.f.read(None), b"hi\nbye\nabc") self.f.seek(0) self.assertEqual(self.f.readline(None), b"hi\n") @@ -112,13 +114,13 @@ self.assertRaises(TypeError, self.f.write, "Hello!") def testRepr(self): - self.assertEqual(repr(self.f), "<_io.FileIO name=%r mode=%r>" - % (self.f.name, self.f.mode)) + self.assertEqual(repr(self.f), "<%s.FileIO name=%r mode=%r>" + % (self.modulename, self.f.name, self.f.mode)) del self.f.name - self.assertEqual(repr(self.f), "<_io.FileIO fd=%r mode=%r>" - % (self.f.fileno(), self.f.mode)) + self.assertEqual(repr(self.f), "<%s.FileIO fd=%r mode=%r>" + % (self.modulename, self.f.fileno(), self.f.mode)) self.f.close() - self.assertEqual(repr(self.f), "<_io.FileIO [closed]>") + self.assertEqual(repr(self.f), "<%s.FileIO [closed]>" % (self.modulename,)) def testErrors(self): f = self.f @@ -128,15 +130,15 @@ self.assertRaises(ValueError, f.read, 10) # Open for reading f.close() self.assertTrue(f.closed) - f = _FileIO(TESTFN, 'r') + f = self.FileIO(TESTFN, 'r') self.assertRaises(TypeError, f.readinto, "") self.assertTrue(not f.closed) f.close() self.assertTrue(f.closed) def testMethods(self): - methods = ['fileno', 'isatty', 'read', 'readinto', - 'seek', 'tell', 'truncate', 'write', 'seekable', + methods = ['fileno', 'isatty', 'read', + 'tell', 'truncate', 'seekable', 'readable', 'writable'] self.f.close() @@ -146,13 +148,16 @@ method = getattr(self.f, methodname) # should raise on closed file self.assertRaises(ValueError, method) + self.assertRaises(ValueError, self.f.readinto, bytearray()) + self.assertRaises(ValueError, self.f.seek, 0, os.SEEK_CUR) + self.assertRaises(ValueError, self.f.write, b'') def testOpendir(self): # Issue 3703: opening a directory should fill the errno # Windows always returns "[Errno 13]: Permission denied # Unix uses fstat and returns "[Errno 21]: Is a directory" try: - _FileIO('.', 'r') + self.FileIO('.', 'r') except OSError as e: self.assertNotEqual(e.errno, 0) self.assertEqual(e.filename, ".") @@ -163,7 +168,7 @@ def testOpenDirFD(self): fd = os.open('.', os.O_RDONLY) with self.assertRaises(OSError) as cm: - _FileIO(fd, 'r') + self.FileIO(fd, 'r') os.close(fd) self.assertEqual(cm.exception.errno, errno.EISDIR) @@ -248,7 +253,7 @@ self.f.close() except OSError: pass - self.f = _FileIO(TESTFN, 'r') + self.f = self.FileIO(TESTFN, 'r') os.close(self.f.fileno()) return self.f @@ -268,23 +273,32 @@ a = array('b', b'x'*10) f.readinto(a) -class OtherFileTests(unittest.TestCase): +class CAutoFileTests(AutoFileTests, unittest.TestCase): + FileIO = _io.FileIO + modulename = '_io' + +class PyAutoFileTests(AutoFileTests, unittest.TestCase): + FileIO = _pyio.FileIO + modulename = '_pyio' + + +class OtherFileTests: def testAbles(self): try: - f = _FileIO(TESTFN, "w") + f = self.FileIO(TESTFN, "w") self.assertEqual(f.readable(), False) self.assertEqual(f.writable(), True) self.assertEqual(f.seekable(), True) f.close() - f = _FileIO(TESTFN, "r") + f = self.FileIO(TESTFN, "r") self.assertEqual(f.readable(), True) self.assertEqual(f.writable(), False) self.assertEqual(f.seekable(), True) f.close() - f = _FileIO(TESTFN, "a+") + f = self.FileIO(TESTFN, "a+") self.assertEqual(f.readable(), True) self.assertEqual(f.writable(), True) self.assertEqual(f.seekable(), True) @@ -293,7 +307,7 @@ if sys.platform != "win32": try: - f = _FileIO("/dev/tty", "a") + f = self.FileIO("/dev/tty", "a") except OSError: # When run in a cron job there just aren't any # ttys, so skip the test. This also handles other @@ -316,7 +330,7 @@ # check invalid mode strings for mode in ("", "aU", "wU+", "rw", "rt"): try: - f = _FileIO(TESTFN, mode) + f = self.FileIO(TESTFN, mode) except ValueError: pass else: @@ -332,7 +346,7 @@ ('ab+', 'ab+'), ('a+b', 'ab+'), ('r', 'rb'), ('rb', 'rb'), ('rb+', 'rb+'), ('r+b', 'rb+')]: # read modes are last so that TESTFN will exist first - with _FileIO(TESTFN, modes[0]) as f: + with self.FileIO(TESTFN, modes[0]) as f: self.assertEqual(f.mode, modes[1]) finally: if os.path.exists(TESTFN): @@ -340,7 +354,7 @@ def testUnicodeOpen(self): # verify repr works for unicode too - f = _FileIO(str(TESTFN), "w") + f = self.FileIO(str(TESTFN), "w") f.close() os.unlink(TESTFN) @@ -350,7 +364,7 @@ fn = TESTFN.encode("ascii") except UnicodeEncodeError: self.skipTest('could not encode %r to ascii' % TESTFN) - f = _FileIO(fn, "w") + f = self.FileIO(fn, "w") try: f.write(b"abc") f.close() @@ -361,28 +375,22 @@ def testConstructorHandlesNULChars(self): fn_with_NUL = 'foo\0bar' - self.assertRaises(TypeError, _FileIO, fn_with_NUL, 'w') - self.assertRaises(TypeError, _FileIO, bytes(fn_with_NUL, 'ascii'), 'w') + self.assertRaises(TypeError, self.FileIO, fn_with_NUL, 'w') + with self.assertRaises((TypeError, ValueError)): + self.FileIO(bytes(fn_with_NUL, 'ascii'), 'w') def testInvalidFd(self): - self.assertRaises(ValueError, _FileIO, -10) - self.assertRaises(OSError, _FileIO, make_bad_fd()) + self.assertRaises(ValueError, self.FileIO, -10) + self.assertRaises(OSError, self.FileIO, make_bad_fd()) if sys.platform == 'win32': import msvcrt self.assertRaises(OSError, msvcrt.get_osfhandle, make_bad_fd()) - @cpython_only - def testInvalidFd_overflow(self): - # Issue 15989 - import _testcapi - self.assertRaises(TypeError, _FileIO, _testcapi.INT_MAX + 1) - self.assertRaises(TypeError, _FileIO, _testcapi.INT_MIN - 1) - def testBadModeArgument(self): # verify that we get a sensible error message for bad mode argument bad_mode = "qwerty" try: - f = _FileIO(TESTFN, bad_mode) + f = self.FileIO(TESTFN, bad_mode) except ValueError as msg: if msg.args[0] != 0: s = str(msg) @@ -395,7 +403,7 @@ self.fail("no error for invalid mode: %s" % bad_mode) def testTruncate(self): - f = _FileIO(TESTFN, 'w') + f = self.FileIO(TESTFN, 'w') f.write(bytes(bytearray(range(10)))) self.assertEqual(f.tell(), 10) f.truncate(5) @@ -410,11 +418,11 @@ def bug801631(): # SF bug # "file.truncate fault on windows" - f = _FileIO(TESTFN, 'w') + f = self.FileIO(TESTFN, 'w') f.write(bytes(range(11))) f.close() - f = _FileIO(TESTFN,'r+') + f = self.FileIO(TESTFN,'r+') data = f.read(5) if data != bytes(range(5)): self.fail("Read on file opened for update failed %r" % data) @@ -454,19 +462,19 @@ pass def testInvalidInit(self): - self.assertRaises(TypeError, _FileIO, "1", 0, 0) + self.assertRaises(TypeError, self.FileIO, "1", 0, 0) def testWarnings(self): with check_warnings(quiet=True) as w: self.assertEqual(w.warnings, []) - self.assertRaises(TypeError, _FileIO, []) + self.assertRaises(TypeError, self.FileIO, []) self.assertEqual(w.warnings, []) - self.assertRaises(ValueError, _FileIO, "/some/invalid/name", "rt") + self.assertRaises(ValueError, self.FileIO, "/some/invalid/name", "rt") self.assertEqual(w.warnings, []) def testUnclosedFDOnException(self): class MyException(Exception): pass - class MyFileIO(_FileIO): + class MyFileIO(self.FileIO): def __setattr__(self, name, value): if name == "name": raise MyException("blocked setting name") @@ -475,12 +483,28 @@ self.assertRaises(MyException, MyFileIO, fd) os.close(fd) # should not raise OSError(EBADF) +class COtherFileTests(OtherFileTests, unittest.TestCase): + FileIO = _io.FileIO + modulename = '_io' + + @cpython_only + def testInvalidFd_overflow(self): + # Issue 15989 + import _testcapi + self.assertRaises(TypeError, self.FileIO, _testcapi.INT_MAX + 1) + self.assertRaises(TypeError, self.FileIO, _testcapi.INT_MIN - 1) + +class PyOtherFileTests(OtherFileTests, unittest.TestCase): + FileIO = _pyio.FileIO + modulename = '_pyio' + def test_main(): # Historically, these tests have been sloppy about removing TESTFN. # So get rid of it no matter what. try: - run_unittest(AutoFileTests, OtherFileTests) + run_unittest(CAutoFileTests, PyAutoFileTests, + COtherFileTests, PyOtherFileTests) finally: if os.path.exists(TESTFN): os.unlink(TESTFN) diff -r e22d0ff286f9 Lib/test/test_io.py --- a/Lib/test/test_io.py Sat Jul 05 11:11:09 2014 +0300 +++ b/Lib/test/test_io.py Sat Jul 05 23:28:03 2014 +0300 @@ -368,7 +368,8 @@ def test_open_handles_NUL_chars(self): fn_with_NUL = 'foo\0bar' self.assertRaises(TypeError, self.open, fn_with_NUL, 'w') - self.assertRaises(TypeError, self.open, bytes(fn_with_NUL, 'ascii'), 'w') + with self.assertRaises((TypeError, ValueError)): + self.open(bytes(fn_with_NUL, 'ascii'), 'w') def test_raw_file_io(self): with self.open(support.TESTFN, "wb", buffering=0) as f: