traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Issue #10233: Close file objects in a timely manner in the tarfile module 

and its test suite.
This commit is contained in:
Antoine Pitrou 2010-10-29 23:49:49 +00:00
parent 749afa95ce
commit e1eca4e3f5
3 changed files with 58 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
Lib/tarfile.py
View File

@ -1800,20 +1800,18 @@ class TarFile(object):
except (ImportError, AttributeError): except (ImportError, AttributeError):
raise CompressionError("gzip module is not available") raise CompressionError("gzip module is not available")
if fileobj is None: extfileobj = fileobj is not None
fileobj = bltn_open(name, mode + "b")
extfileobj = False
else:
extfileobj = True
try: try:
t = cls.taropen(name, mode, fileobj = gzip.GzipFile(name, mode + "b", compresslevel, fileobj)
gzip.GzipFile(name, mode, compresslevel, fileobj), t = cls.taropen(name, mode, fileobj, **kwargs)
**kwargs)
except IOError: except IOError:
if not extfileobj: if not extfileobj:
fileobj.close() fileobj.close()
raise ReadError("not a gzip file") raise ReadError("not a gzip file")
except:
if not extfileobj:
fileobj.close()
raise
t._extfileobj = extfileobj t._extfileobj = extfileobj
return t return t

75
Lib/test/test_tarfile.py
View File

@ -52,25 +52,32 @@ class UstarReadTest(ReadTest):
def test_fileobj_regular_file(self): def test_fileobj_regular_file(self):
tarinfo = self.tar.getmember("ustar/regtype") tarinfo = self.tar.getmember("ustar/regtype")
fobj = self.tar.extractfile(tarinfo) fobj = self.tar.extractfile(tarinfo)
data = fobj.read() try:
self.assertTrue((len(data), md5sum(data)) == (tarinfo.size, md5_regtype), data = fobj.read()
"regular file extraction failed") self.assertTrue((len(data), md5sum(data)) == (tarinfo.size, md5_regtype),
"regular file extraction failed")
finally:
fobj.close()
def test_fileobj_readlines(self): def test_fileobj_readlines(self):
self.tar.extract("ustar/regtype", TEMPDIR) self.tar.extract("ustar/regtype", TEMPDIR)
tarinfo = self.tar.getmember("ustar/regtype") tarinfo = self.tar.getmember("ustar/regtype")
with open(os.path.join(TEMPDIR, "ustar/regtype"), "r") as fobj1: with open(os.path.join(TEMPDIR, "ustar/regtype"), "r") as fobj1:
lines1 = fobj1.readlines() lines1 = fobj1.readlines()
fobj2 = io.TextIOWrapper(self.tar.extractfile(tarinfo))
lines2 = fobj2.readlines() fobj = self.tar.extractfile(tarinfo)
self.assertTrue(lines1 == lines2, try:
"fileobj.readlines() failed") fobj2 = io.TextIOWrapper(fobj)
self.assertTrue(len(lines2) == 114, lines2 = fobj2.readlines()
"fileobj.readlines() failed") self.assertTrue(lines1 == lines2,
self.assertTrue(lines2[83] == "fileobj.readlines() failed")
"I will gladly admit that Python is not the fastest running scripting language.\n", self.assertTrue(len(lines2) == 114,
"fileobj.readlines() failed") "fileobj.readlines() failed")
self.assertTrue(lines2[83] ==
"I will gladly admit that Python is not the fastest running scripting language.\n",
"fileobj.readlines() failed")
finally:
fobj.close()
def test_fileobj_iter(self): def test_fileobj_iter(self):
self.tar.extract("ustar/regtype", TEMPDIR) self.tar.extract("ustar/regtype", TEMPDIR)
@ -78,9 +85,12 @@ class UstarReadTest(ReadTest):
with open(os.path.join(TEMPDIR, "ustar/regtype"), "rU") as fobj1: with open(os.path.join(TEMPDIR, "ustar/regtype"), "rU") as fobj1:
lines1 = fobj1.readlines() lines1 = fobj1.readlines()
fobj2 = self.tar.extractfile(tarinfo) fobj2 = self.tar.extractfile(tarinfo)
lines2 = list(io.TextIOWrapper(fobj2)) try:
self.assertTrue(lines1 == lines2, lines2 = list(io.TextIOWrapper(fobj2))
"fileobj.__iter__() failed") self.assertTrue(lines1 == lines2,
"fileobj.__iter__() failed")
finally:
fobj2.close()
def test_fileobj_seek(self): def test_fileobj_seek(self):
self.tar.extract("ustar/regtype", TEMPDIR) self.tar.extract("ustar/regtype", TEMPDIR)
@ -138,7 +148,11 @@ class UstarReadTest(ReadTest):
def _test_fileobj_link(self, lnktype, regtype): def _test_fileobj_link(self, lnktype, regtype):
a = self.tar.extractfile(lnktype) a = self.tar.extractfile(lnktype)
b = self.tar.extractfile(regtype) b = self.tar.extractfile(regtype)
self.assertEqual(a.name, b.name) try:
self.assertEqual(a.name, b.name)
finally:
a.close()
b.close()
def test_fileobj_link1(self): def test_fileobj_link1(self):
self._test_fileobj_link("ustar/lnktype", "ustar/regtype") self._test_fileobj_link("ustar/lnktype", "ustar/regtype")
@ -225,8 +239,8 @@ class MiscReadTest(CommonReadTest):
data = fobj.read() data = fobj.read()
fobj = io.BytesIO(data) fobj = io.BytesIO(data)
fobj.name = "" fobj.name = ""
tar = tarfile.open(fileobj=fobj, mode=self.mode) with tarfile.open(fileobj=fobj, mode=self.mode) as tar:
self.assertEqual(tar.name, None) self.assertEqual(tar.name, None)
def test_fileobj_with_offset(self): def test_fileobj_with_offset(self):
# Skip the first member and store values from the second member # Skip the first member and store values from the second member
@ -237,7 +251,9 @@ class MiscReadTest(CommonReadTest):
t = tar.next() t = tar.next()
name = t.name name = t.name
offset = t.offset offset = t.offset
data = tar.extractfile(t).read() f = tar.extractfile(t)
data = f.read()
f.close()
finally: finally:
tar.close() tar.close()
@ -319,7 +335,8 @@ class MiscReadTest(CommonReadTest):
if e.errno == errno.ENOENT: if e.errno == errno.ENOENT:
self.fail("hardlink not extracted properly") self.fail("hardlink not extracted properly")
data = open(os.path.join(TEMPDIR, "ustar/lnktype"), "rb").read() with open(os.path.join(TEMPDIR, "ustar/lnktype"), "rb") as f:
data = f.read()
self.assertEqual(md5sum(data), md5_regtype) self.assertEqual(md5sum(data), md5_regtype)
try: try:
@ -328,7 +345,8 @@ class MiscReadTest(CommonReadTest):
if e.errno == errno.ENOENT: if e.errno == errno.ENOENT:
self.fail("symlink not extracted properly") self.fail("symlink not extracted properly")
data = open(os.path.join(TEMPDIR, "ustar/symtype"), "rb").read() with open(os.path.join(TEMPDIR, "ustar/symtype"), "rb") as f:
data = f.read()
self.assertEqual(md5sum(data), md5_regtype) self.assertEqual(md5sum(data), md5_regtype)
finally: finally:
tar.close() tar.close()
@ -604,10 +622,10 @@ class LongnameTest(ReadTest):
# the preceding extended header. # the preceding extended header.
longname = self.subdir + "/" + "123/" * 125 + "longname" longname = self.subdir + "/" + "123/" * 125 + "longname"
offset = self.tar.getmember(longname).offset offset = self.tar.getmember(longname).offset
fobj = open(tarname, "rb") with open(tarname, "rb") as fobj:
fobj.seek(offset) fobj.seek(offset)
tarinfo = tarfile.TarInfo.frombuf(fobj.read(512), "iso8859-1", "strict") tarinfo = tarfile.TarInfo.frombuf(fobj.read(512), "iso8859-1", "strict")
self.assertEqual(tarinfo.type, self.longnametype) self.assertEqual(tarinfo.type, self.longnametype)
class GNUReadTest(LongnameTest): class GNUReadTest(LongnameTest):
@ -1353,8 +1371,11 @@ class AppendTest(unittest.TestCase):
t = src.getmember("ustar/regtype") t = src.getmember("ustar/regtype")
t.name = "foo" t.name = "foo"
f = src.extractfile(t) f = src.extractfile(t)
with tarfile.open(self.tarname, mode) as tar: try:
tar.addfile(t, f) with tarfile.open(self.tarname, mode) as tar:
tar.addfile(t, f)
finally:
f.close()
def _test(self, names=["bar"], fileobj=None): def _test(self, names=["bar"], fileobj=None):
with tarfile.open(self.tarname, fileobj=fileobj) as tar: with tarfile.open(self.tarname, fileobj=fileobj) as tar:

3
Misc/NEWS
View File

@ -54,6 +54,9 @@ Core and Builtins
Library Library
------- -------
- Issue #10233: Close file objects in a timely manner in the tarfile module
and its test suite.
- Issue #10093: ResourceWarnings are now issued when files and sockets are - Issue #10093: ResourceWarnings are now issued when files and sockets are
deallocated without explicit closing. These warnings are silenced by deallocated without explicit closing. These warnings are silenced by
default, except in pydebug mode. default, except in pydebug mode.