Make error handling in BZ2File.{readable,seekable,writable,fileno} consistent with TextIOWrapper.
Also, add tests for these methods.
This commit is contained in:
parent
3ff069ebc6
commit
44ae4a2a22
|
@ -125,6 +125,7 @@ class BZ2File(io.BufferedIOBase):
|
|||
|
||||
def fileno(self):
|
||||
"""Return the file descriptor for the underlying file."""
|
||||
self._check_not_closed()
|
||||
return self._fp.fileno()
|
||||
|
||||
def seekable(self):
|
||||
|
@ -133,10 +134,12 @@ class BZ2File(io.BufferedIOBase):
|
|||
|
||||
def readable(self):
|
||||
"""Return whether the file was opened for reading."""
|
||||
self._check_not_closed()
|
||||
return self._mode in (_MODE_READ, _MODE_READ_EOF)
|
||||
|
||||
def writable(self):
|
||||
"""Return whether the file was opened for writing."""
|
||||
self._check_not_closed()
|
||||
return self._mode == _MODE_WRITE
|
||||
|
||||
# Mode-checking helper functions.
|
||||
|
@ -147,17 +150,14 @@ class BZ2File(io.BufferedIOBase):
|
|||
|
||||
def _check_can_read(self):
|
||||
if not self.readable():
|
||||
self._check_not_closed()
|
||||
raise io.UnsupportedOperation("File not open for reading")
|
||||
|
||||
def _check_can_write(self):
|
||||
if not self.writable():
|
||||
self._check_not_closed()
|
||||
raise io.UnsupportedOperation("File not open for writing")
|
||||
|
||||
def _check_can_seek(self):
|
||||
if not self.seekable():
|
||||
self._check_not_closed()
|
||||
raise io.UnsupportedOperation("Seeking is only supported "
|
||||
"on files open for reading")
|
||||
|
||||
|
|
|
@ -348,8 +348,63 @@ class BZ2FileTest(BaseTest):
|
|||
def testFileno(self):
|
||||
self.createTempFile()
|
||||
with open(self.filename, 'rb') as rawf:
|
||||
with BZ2File(fileobj=rawf) as bz2f:
|
||||
bz2f = BZ2File(fileobj=rawf)
|
||||
try:
|
||||
self.assertEqual(bz2f.fileno(), rawf.fileno())
|
||||
finally:
|
||||
bz2f.close()
|
||||
self.assertRaises(ValueError, bz2f.fileno)
|
||||
|
||||
def testSeekable(self):
|
||||
bz2f = BZ2File(fileobj=BytesIO(self.DATA))
|
||||
try:
|
||||
self.assertTrue(bz2f.seekable())
|
||||
bz2f.read()
|
||||
self.assertTrue(bz2f.seekable())
|
||||
finally:
|
||||
bz2f.close()
|
||||
self.assertRaises(ValueError, bz2f.seekable)
|
||||
|
||||
bz2f = BZ2File(fileobj=BytesIO(), mode="w")
|
||||
try:
|
||||
self.assertFalse(bz2f.seekable())
|
||||
finally:
|
||||
bz2f.close()
|
||||
self.assertRaises(ValueError, bz2f.seekable)
|
||||
|
||||
def testReadable(self):
|
||||
bz2f = BZ2File(fileobj=BytesIO(self.DATA))
|
||||
try:
|
||||
self.assertTrue(bz2f.readable())
|
||||
bz2f.read()
|
||||
self.assertTrue(bz2f.readable())
|
||||
finally:
|
||||
bz2f.close()
|
||||
self.assertRaises(ValueError, bz2f.readable)
|
||||
|
||||
bz2f = BZ2File(fileobj=BytesIO(), mode="w")
|
||||
try:
|
||||
self.assertFalse(bz2f.readable())
|
||||
finally:
|
||||
bz2f.close()
|
||||
self.assertRaises(ValueError, bz2f.readable)
|
||||
|
||||
def testWritable(self):
|
||||
bz2f = BZ2File(fileobj=BytesIO(self.DATA))
|
||||
try:
|
||||
self.assertFalse(bz2f.writable())
|
||||
bz2f.read()
|
||||
self.assertFalse(bz2f.writable())
|
||||
finally:
|
||||
bz2f.close()
|
||||
self.assertRaises(ValueError, bz2f.writable)
|
||||
|
||||
bz2f = BZ2File(fileobj=BytesIO(), mode="w")
|
||||
try:
|
||||
self.assertTrue(bz2f.writable())
|
||||
finally:
|
||||
bz2f.close()
|
||||
self.assertRaises(ValueError, bz2f.writable)
|
||||
|
||||
def testOpenDel(self):
|
||||
self.createTempFile()
|
||||
|
|
Loading…
Reference in New Issue