mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-20 02:50:14 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Make BZ2File's fileobj support easier to use.

Browse source 
The fileobj argument was added during the 3.3 development cycle, so this change
does not break backward compatibility with 3.2.
This commit is contained in:
Nadeem Vawda 2012-06-04 23:31:20 +02:00
parent 68721019ef
commit aebcdba829
5 changed files with 49 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

17
Lib/bz2.py
View file

@ -39,13 +39,12 @@ class BZ2File(io.BufferedIOBase):
returned as bytes, and data to be written should be given as bytes.
"""
def __init__(self, filename=None, mode="r", buffering=None,
compresslevel=9, *, fileobj=None):
def __init__(self, filename, mode="r", buffering=None, compresslevel=9):
"""Open a bzip2-compressed file.
If filename is given, open the named file. Otherwise, operate on
the file object given by fileobj. Exactly one of these two
parameters should be provided.
If filename is a str or bytes object, is gives the name of the file to
be opened. Otherwise, it should be a file object, which will be used to
read or write the compressed data.
mode can be 'r' for reading (default), 'w' for (over)writing, or
'a' for appending.
@ -91,15 +90,15 @@ class BZ2File(io.BufferedIOBase):
else:
raise ValueError("Invalid mode: {!r}".format(mode))
if filename is not None and fileobj is None:
if isinstance(filename, (str, bytes)):
self._fp = open(filename, mode)
self._closefp = True
self._mode = mode_code
elif fileobj is not None and filename is None:
self._fp = fileobj
elif hasattr(filename, "read") or hasattr(filename, "write"):
self._fp = filename
self._mode = mode_code
else:
raise ValueError("Must give exactly one of filename and fileobj")
raise TypeError("filename must be a str or bytes object, or a file")
def close(self):
"""Flush and close the file.

4
Lib/tarfile.py
View file

@ -1657,8 +1657,8 @@ class TarFile(object):
except ImportError:
raise CompressionError("bz2 module is not available")
fileobj = bz2.BZ2File(filename=name if fileobj is None else None,
mode=mode, fileobj=fileobj, compresslevel=compresslevel)
fileobj = bz2.BZ2File(fileobj or name, mode,
compresslevel=compresslevel)
try:
t = cls.taropen(name, mode, fileobj, **kwargs)

40
Lib/test/test_bz2.py
View file

@ -81,6 +81,20 @@ class BZ2FileTest(BaseTest):
with open(self.filename, "wb") as f:
f.write(self.DATA * streams)
def testBadArgs(self):
with self.assertRaises(TypeError):
BZ2File(123.456)
with self.assertRaises(ValueError):
BZ2File("/dev/null", "z")
with self.assertRaises(ValueError):
BZ2File("/dev/null", "rx")
with self.assertRaises(ValueError):
BZ2File("/dev/null", "rbt")
with self.assertRaises(ValueError):
BZ2File("/dev/null", compresslevel=0)
with self.assertRaises(ValueError):
BZ2File("/dev/null", compresslevel=10)
def testRead(self):
self.createTempFile()
with BZ2File(self.filename) as bz2f:
@ -348,7 +362,7 @@ class BZ2FileTest(BaseTest):
def testFileno(self):
self.createTempFile()
with open(self.filename, 'rb') as rawf:
bz2f = BZ2File(fileobj=rawf)
bz2f = BZ2File(rawf)
try:
self.assertEqual(bz2f.fileno(), rawf.fileno())
finally:
@ -356,7 +370,7 @@ class BZ2FileTest(BaseTest):
self.assertRaises(ValueError, bz2f.fileno)
def testSeekable(self):
bz2f = BZ2File(fileobj=BytesIO(self.DATA))
bz2f = BZ2File(BytesIO(self.DATA))
try:
self.assertTrue(bz2f.seekable())
bz2f.read()
@ -365,7 +379,7 @@ class BZ2FileTest(BaseTest):
bz2f.close()
self.assertRaises(ValueError, bz2f.seekable)
bz2f = BZ2File(fileobj=BytesIO(), mode="w")
bz2f = BZ2File(BytesIO(), mode="w")
try:
self.assertFalse(bz2f.seekable())
finally:
@ -374,7 +388,7 @@ class BZ2FileTest(BaseTest):
src = BytesIO(self.DATA)
src.seekable = lambda: False
bz2f = BZ2File(fileobj=src)
bz2f = BZ2File(src)
try:
self.assertFalse(bz2f.seekable())
finally:
@ -382,7 +396,7 @@ class BZ2FileTest(BaseTest):
self.assertRaises(ValueError, bz2f.seekable)
def testReadable(self):
bz2f = BZ2File(fileobj=BytesIO(self.DATA))
bz2f = BZ2File(BytesIO(self.DATA))
try:
self.assertTrue(bz2f.readable())
bz2f.read()
@ -391,7 +405,7 @@ class BZ2FileTest(BaseTest):
bz2f.close()
self.assertRaises(ValueError, bz2f.readable)
bz2f = BZ2File(fileobj=BytesIO(), mode="w")
bz2f = BZ2File(BytesIO(), mode="w")
try:
self.assertFalse(bz2f.readable())
finally:
@ -399,7 +413,7 @@ class BZ2FileTest(BaseTest):
self.assertRaises(ValueError, bz2f.readable)
def testWritable(self):
bz2f = BZ2File(fileobj=BytesIO(self.DATA))
bz2f = BZ2File(BytesIO(self.DATA))
try:
self.assertFalse(bz2f.writable())
bz2f.read()
@ -408,7 +422,7 @@ class BZ2FileTest(BaseTest):
bz2f.close()
self.assertRaises(ValueError, bz2f.writable)
bz2f = BZ2File(fileobj=BytesIO(), mode="w")
bz2f = BZ2File(BytesIO(), mode="w")
try:
self.assertTrue(bz2f.writable())
finally:
@ -512,14 +526,14 @@ class BZ2FileTest(BaseTest):
def testReadBytesIO(self):
with BytesIO(self.DATA) as bio:
with BZ2File(fileobj=bio) as bz2f:
with BZ2File(bio) as bz2f:
self.assertRaises(TypeError, bz2f.read, None)
self.assertEqual(bz2f.read(), self.TEXT)
self.assertFalse(bio.closed)
def testPeekBytesIO(self):
with BytesIO(self.DATA) as bio:
with BZ2File(fileobj=bio) as bz2f:
with BZ2File(bio) as bz2f:
pdata = bz2f.peek()
self.assertNotEqual(len(pdata), 0)
self.assertTrue(self.TEXT.startswith(pdata))
@ -527,7 +541,7 @@ class BZ2FileTest(BaseTest):
def testWriteBytesIO(self):
with BytesIO() as bio:
with BZ2File(fileobj=bio, mode="w") as bz2f:
with BZ2File(bio, "w") as bz2f:
self.assertRaises(TypeError, bz2f.write)
bz2f.write(self.TEXT)
self.assertEqual(self.decompress(bio.getvalue()), self.TEXT)
@ -535,14 +549,14 @@ class BZ2FileTest(BaseTest):
def testSeekForwardBytesIO(self):
with BytesIO(self.DATA) as bio:
with BZ2File(fileobj=bio) as bz2f:
with BZ2File(bio) as bz2f:
self.assertRaises(TypeError, bz2f.seek)
bz2f.seek(150)
self.assertEqual(bz2f.read(), self.TEXT[150:])
def testSeekBackwardsBytesIO(self):
with BytesIO(self.DATA) as bio:
with BZ2File(fileobj=bio) as bz2f:
with BZ2File(bio) as bz2f:
bz2f.read(500)
bz2f.seek(-150, 1)
self.assertEqual(bz2f.read(), self.TEXT[500-150:])