Issue #9293: I/O streams now raise `io.UnsupportedOperation` when an
unsupported operation is attempted (for example, writing to a file open
only for reading).
diff --git a/Lib/_pyio.py b/Lib/_pyio.py
index 66f990d..12ae4b6 100644
--- a/Lib/_pyio.py
+++ b/Lib/_pyio.py
@@ -243,8 +243,13 @@
return open(*args, **kwargs)
-class UnsupportedOperation(ValueError, IOError):
- pass
+# In normal operation, both `UnsupportedOperation`s should be bound to the
+# same object.
+try:
+ UnsupportedOperation = io.UnsupportedOperation
+except AttributeError:
+ class UnsupportedOperation(ValueError, IOError):
+ pass
class IOBase(metaclass=abc.ABCMeta):
@@ -362,9 +367,8 @@
"""Internal: raise an IOError if file is not seekable
"""
if not self.seekable():
- raise IOError("File or stream is not seekable."
- if msg is None else msg)
-
+ raise UnsupportedOperation("File or stream is not seekable."
+ if msg is None else msg)
def readable(self) -> bool:
"""Return whether object was opened for reading.
@@ -377,8 +381,8 @@
"""Internal: raise an IOError if file is not readable
"""
if not self.readable():
- raise IOError("File or stream is not readable."
- if msg is None else msg)
+ raise UnsupportedOperation("File or stream is not readable."
+ if msg is None else msg)
def writable(self) -> bool:
"""Return whether object was opened for writing.
@@ -391,8 +395,8 @@
"""Internal: raise an IOError if file is not writable
"""
if not self.writable():
- raise IOError("File or stream is not writable."
- if msg is None else msg)
+ raise UnsupportedOperation("File or stream is not writable."
+ if msg is None else msg)
@property
def closed(self):
@@ -1647,7 +1651,7 @@
def tell(self):
if not self._seekable:
- raise IOError("underlying stream is not seekable")
+ raise UnsupportedOperation("underlying stream is not seekable")
if not self._telling:
raise IOError("telling position disabled by next() call")
self.flush()
@@ -1726,17 +1730,17 @@
if self.closed:
raise ValueError("tell on closed file")
if not self._seekable:
- raise IOError("underlying stream is not seekable")
+ raise UnsupportedOperation("underlying stream is not seekable")
if whence == 1: # seek relative to current position
if cookie != 0:
- raise IOError("can't do nonzero cur-relative seeks")
+ raise UnsupportedOperation("can't do nonzero cur-relative seeks")
# Seeking to the current position should attempt to
# sync the underlying buffer with the current position.
whence = 0
cookie = self.tell()
if whence == 2: # seek relative to end of file
if cookie != 0:
- raise IOError("can't do nonzero end-relative seeks")
+ raise UnsupportedOperation("can't do nonzero end-relative seeks")
self.flush()
position = self.buffer.seek(0, 2)
self._set_decoded_chars('')
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
index 76c8536..c9f7582 100644
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -179,6 +179,23 @@
pass
+class MockUnseekableIO:
+ def seekable(self):
+ return False
+
+ def seek(self, *args):
+ raise self.UnsupportedOperation("not seekable")
+
+ def tell(self, *args):
+ raise self.UnsupportedOperation("not seekable")
+
+class CMockUnseekableIO(MockUnseekableIO, io.BytesIO):
+ UnsupportedOperation = io.UnsupportedOperation
+
+class PyMockUnseekableIO(MockUnseekableIO, pyio.BytesIO):
+ UnsupportedOperation = pyio.UnsupportedOperation
+
+
class MockNonBlockWriterIO:
def __init__(self):
@@ -304,16 +321,26 @@
def test_invalid_operations(self):
# Try writing on a file opened in read mode and vice-versa.
+ exc = self.UnsupportedOperation
for mode in ("w", "wb"):
with self.open(support.TESTFN, mode) as fp:
- self.assertRaises(IOError, fp.read)
- self.assertRaises(IOError, fp.readline)
+ self.assertRaises(exc, fp.read)
+ self.assertRaises(exc, fp.readline)
+ with self.open(support.TESTFN, "wb", buffering=0) as fp:
+ self.assertRaises(exc, fp.read)
+ self.assertRaises(exc, fp.readline)
+ with self.open(support.TESTFN, "rb", buffering=0) as fp:
+ self.assertRaises(exc, fp.write, b"blah")
+ self.assertRaises(exc, fp.writelines, [b"blah\n"])
with self.open(support.TESTFN, "rb") as fp:
- self.assertRaises(IOError, fp.write, b"blah")
- self.assertRaises(IOError, fp.writelines, [b"blah\n"])
+ self.assertRaises(exc, fp.write, b"blah")
+ self.assertRaises(exc, fp.writelines, [b"blah\n"])
with self.open(support.TESTFN, "r") as fp:
- self.assertRaises(IOError, fp.write, "blah")
- self.assertRaises(IOError, fp.writelines, ["blah\n"])
+ self.assertRaises(exc, fp.write, "blah")
+ self.assertRaises(exc, fp.writelines, ["blah\n"])
+ # Non-zero seeking from current or end pos
+ self.assertRaises(exc, fp.seek, 1, self.SEEK_CUR)
+ self.assertRaises(exc, fp.seek, -1, self.SEEK_END)
def test_raw_file_io(self):
with self.open(support.TESTFN, "wb", buffering=0) as f:
@@ -670,6 +697,11 @@
b.close()
self.assertRaises(ValueError, b.flush)
+ def test_unseekable(self):
+ bufio = self.tp(self.MockUnseekableIO(b"A" * 10))
+ self.assertRaises(self.UnsupportedOperation, bufio.tell)
+ self.assertRaises(self.UnsupportedOperation, bufio.seek, 0)
+
class BufferedReaderTest(unittest.TestCase, CommonBufferedTests):
read_mode = "rb"
@@ -1433,6 +1465,9 @@
BufferedReaderTest.test_misbehaved_io(self)
BufferedWriterTest.test_misbehaved_io(self)
+ # You can't construct a BufferedRandom over a non-seekable stream.
+ test_unseekable = None
+
class CBufferedRandomTest(BufferedRandomTest):
tp = io.BufferedRandom
@@ -2177,6 +2212,11 @@
txt.close()
self.assertRaises(ValueError, txt.flush)
+ def test_unseekable(self):
+ txt = self.TextIOWrapper(self.MockUnseekableIO(self.testdata))
+ self.assertRaises(self.UnsupportedOperation, txt.tell)
+ self.assertRaises(self.UnsupportedOperation, txt.seek, 0)
+
class CTextIOWrapperTest(TextIOWrapperTest):
def test_initialization(self):
@@ -2550,7 +2590,7 @@
# Put the namespaces of the IO module we are testing and some useful mock
# classes in the __dict__ of each test.
mocks = (MockRawIO, MisbehavedRawIO, MockFileIO, CloseFailureIO,
- MockNonBlockWriterIO)
+ MockNonBlockWriterIO, MockUnseekableIO)
all_members = io.__all__ + ["IncrementalNewlineDecoder"]
c_io_ns = {name : getattr(io, name) for name in all_members}
py_io_ns = {name : getattr(pyio, name) for name in all_members}