mirror of
https://github.com/python/cpython.git
synced 2025-09-26 18:29:57 +00:00
Add proper tests for closed files to various I/O operations,
restoring a disabled test. This was necessary to make test_pickle.py pass.
This commit is contained in:
parent
2b6a97e69c
commit
4b5386f398
2 changed files with 41 additions and 14 deletions
21
Lib/io.py
21
Lib/io.py
|
@ -157,6 +157,10 @@ def open(file, mode="r", buffering=None, encoding=None, newline=None):
|
||||||
return text
|
return text
|
||||||
|
|
||||||
|
|
||||||
|
class UnsupportedOperation(ValueError, IOError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
class IOBase:
|
class IOBase:
|
||||||
|
|
||||||
"""Base class for all I/O classes.
|
"""Base class for all I/O classes.
|
||||||
|
@ -177,8 +181,8 @@ class IOBase:
|
||||||
|
|
||||||
def _unsupported(self, name: str) -> IOError:
|
def _unsupported(self, name: str) -> IOError:
|
||||||
"""Internal: raise an exception for unsupported operations."""
|
"""Internal: raise an exception for unsupported operations."""
|
||||||
raise IOError("%s.%s() not supported" % (self.__class__.__name__,
|
raise UnsupportedOperation("%s.%s() not supported" %
|
||||||
name))
|
(self.__class__.__name__, name))
|
||||||
|
|
||||||
### Positioning ###
|
### Positioning ###
|
||||||
|
|
||||||
|
@ -327,6 +331,8 @@ class IOBase:
|
||||||
return res
|
return res
|
||||||
|
|
||||||
def __iter__(self):
|
def __iter__(self):
|
||||||
|
if self.closed:
|
||||||
|
raise ValueError("__iter__ on closed file")
|
||||||
return self
|
return self
|
||||||
|
|
||||||
def __next__(self):
|
def __next__(self):
|
||||||
|
@ -348,6 +354,8 @@ class IOBase:
|
||||||
return lines
|
return lines
|
||||||
|
|
||||||
def writelines(self, lines):
|
def writelines(self, lines):
|
||||||
|
if self.closed:
|
||||||
|
raise ValueError("write to closed file")
|
||||||
for line in lines:
|
for line in lines:
|
||||||
self.write(line)
|
self.write(line)
|
||||||
|
|
||||||
|
@ -587,6 +595,7 @@ class _BufferedIOMixin(BufferedIOBase):
|
||||||
self.raw.flush()
|
self.raw.flush()
|
||||||
|
|
||||||
def close(self):
|
def close(self):
|
||||||
|
if not self.closed:
|
||||||
self.flush()
|
self.flush()
|
||||||
self.raw.close()
|
self.raw.close()
|
||||||
|
|
||||||
|
@ -644,6 +653,8 @@ class BytesIO(BufferedIOBase):
|
||||||
return self.read(n)
|
return self.read(n)
|
||||||
|
|
||||||
def write(self, b):
|
def write(self, b):
|
||||||
|
if self.closed:
|
||||||
|
raise ValueError("write to closed file")
|
||||||
n = len(b)
|
n = len(b)
|
||||||
newpos = self._pos + n
|
newpos = self._pos + n
|
||||||
self._buffer[self._pos:newpos] = b
|
self._buffer[self._pos:newpos] = b
|
||||||
|
@ -779,6 +790,8 @@ class BufferedWriter(_BufferedIOMixin):
|
||||||
self._write_buf = b""
|
self._write_buf = b""
|
||||||
|
|
||||||
def write(self, b):
|
def write(self, b):
|
||||||
|
if self.closed:
|
||||||
|
raise ValueError("write to closed file")
|
||||||
if not isinstance(b, bytes):
|
if not isinstance(b, bytes):
|
||||||
if hasattr(b, "__index__"):
|
if hasattr(b, "__index__"):
|
||||||
raise TypeError("Can't write object of type %s" %
|
raise TypeError("Can't write object of type %s" %
|
||||||
|
@ -809,6 +822,8 @@ class BufferedWriter(_BufferedIOMixin):
|
||||||
return written
|
return written
|
||||||
|
|
||||||
def flush(self):
|
def flush(self):
|
||||||
|
if self.closed:
|
||||||
|
raise ValueError("flush of closed file")
|
||||||
written = 0
|
written = 0
|
||||||
try:
|
try:
|
||||||
while self._write_buf:
|
while self._write_buf:
|
||||||
|
@ -1040,6 +1055,8 @@ class TextIOWrapper(TextIOBase):
|
||||||
return self.buffer.isatty()
|
return self.buffer.isatty()
|
||||||
|
|
||||||
def write(self, s: str):
|
def write(self, s: str):
|
||||||
|
if self.closed:
|
||||||
|
raise ValueError("write to closed file")
|
||||||
# XXX What if we were just reading?
|
# XXX What if we were just reading?
|
||||||
b = s.encode(self._encoding)
|
b = s.encode(self._encoding)
|
||||||
if isinstance(b, str):
|
if isinstance(b, str):
|
||||||
|
|
|
@ -89,21 +89,31 @@ class AutoFileTests(unittest.TestCase):
|
||||||
self.assert_(f.closed)
|
self.assert_(f.closed)
|
||||||
|
|
||||||
def testMethods(self):
|
def testMethods(self):
|
||||||
methods = ['fileno', 'flush', 'isatty', '__next__', 'read', 'readinto',
|
methods = [('fileno', ()),
|
||||||
'readline', 'readlines', 'seek', 'tell', 'truncate',
|
('flush', ()),
|
||||||
'write', '__iter__']
|
('isatty', ()),
|
||||||
if sys.platform.startswith('atheos'):
|
('__next__', ()),
|
||||||
methods.remove('truncate')
|
('read', ()),
|
||||||
|
('write', (b"",)),
|
||||||
|
('readline', ()),
|
||||||
|
('readlines', ()),
|
||||||
|
('seek', (0,)),
|
||||||
|
('tell', ()),
|
||||||
|
('write', (b"",)),
|
||||||
|
('writelines', ([],)),
|
||||||
|
('__iter__', ()),
|
||||||
|
]
|
||||||
|
if not sys.platform.startswith('atheos'):
|
||||||
|
methods.append(('truncate', ()))
|
||||||
|
|
||||||
# __exit__ should close the file
|
# __exit__ should close the file
|
||||||
self.f.__exit__(None, None, None)
|
self.f.__exit__(None, None, None)
|
||||||
self.assert_(self.f.closed)
|
self.assert_(self.f.closed)
|
||||||
|
|
||||||
## for methodname in methods:
|
for methodname, args in methods:
|
||||||
## method = getattr(self.f, methodname)
|
method = getattr(self.f, methodname)
|
||||||
## # should raise on closed file
|
# should raise on closed file
|
||||||
## self.assertRaises(ValueError, method)
|
self.assertRaises(ValueError, method, *args)
|
||||||
## self.assertRaises(ValueError, self.f.writelines, [])
|
|
||||||
|
|
||||||
# file is closed, __exit__ shouldn't do anything
|
# file is closed, __exit__ shouldn't do anything
|
||||||
self.assertEquals(self.f.__exit__(None, None, None), None)
|
self.assertEquals(self.f.__exit__(None, None, None), None)
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue