mirror of https://github.com/python/cpython.git
Try to fix test_tarfile issues on Windows buildbots by closing file
objects explicitly instead of letting them linger on.
This commit is contained in:
Antoine Pitrou
parent
e5768cf348
commit
95f5560b46
|
|
@ -1764,14 +1764,19 @@ def gzopen(cls, name, mode="r", fileobj=None, compresslevel=9, **kwargs):
|
|||
|
||||
if fileobj is None:
|
||||
fileobj = bltn_open(name, mode + "b")
|
||||
extfileobj = False
|
||||
else:
|
||||
extfileobj = True
|
||||
|
||||
try:
|
||||
t = cls.taropen(name, mode,
|
||||
gzip.GzipFile(name, mode, compresslevel, fileobj),
|
||||
**kwargs)
|
||||
except IOError:
|
||||
if not extfileobj:
|
||||
fileobj.close()
|
||||
raise ReadError("not a gzip file")
|
||||
t._extfileobj = False
|
||||
t._extfileobj = extfileobj
|
||||
return t
|
||||
|
||||
@classmethod
|
||||
|
|
@ -1795,6 +1800,7 @@ def bz2open(cls, name, mode="r", fileobj=None, compresslevel=9, **kwargs):
|
|||
try:
|
||||
t = cls.taropen(name, mode, fileobj, **kwargs)
|
||||
except (IOError, EOFError):
|
||||
fileobj.close()
|
||||
raise ReadError("not a bzip2 file")
|
||||
t._extfileobj = False
|
||||
return t
|
||||
|
|
|
|||
|
|
@ -59,10 +59,10 @@ def test_fileobj_regular_file(self):
|
|||
def test_fileobj_readlines(self):
|
||||
self.tar.extract("ustar/regtype", TEMPDIR)
|
||||
tarinfo = self.tar.getmember("ustar/regtype")
|
||||
fobj1 = open(os.path.join(TEMPDIR, "ustar/regtype"), "r")
|
||||
with open(os.path.join(TEMPDIR, "ustar/regtype"), "r") as fobj1:
|
||||
lines1 = fobj1.readlines()
|
||||
fobj2 = io.TextIOWrapper(self.tar.extractfile(tarinfo))
|
||||
|
||||
lines1 = fobj1.readlines()
|
||||
lines2 = fobj2.readlines()
|
||||
self.assertTrue(lines1 == lines2,
|
||||
"fileobj.readlines() failed")
|
||||
|
|
@ -75,18 +75,17 @@ def test_fileobj_readlines(self):
|
|||
def test_fileobj_iter(self):
|
||||
self.tar.extract("ustar/regtype", TEMPDIR)
|
||||
tarinfo = self.tar.getmember("ustar/regtype")
|
||||
fobj1 = open(os.path.join(TEMPDIR, "ustar/regtype"), "rU")
|
||||
fobj2 = self.tar.extractfile(tarinfo)
|
||||
with open(os.path.join(TEMPDIR, "ustar/regtype"), "rU") as fobj1:
|
||||
lines1 = fobj1.readlines()
|
||||
fobj2 = self.tar.extractfile(tarinfo)
|
||||
lines2 = list(io.TextIOWrapper(fobj2))
|
||||
self.assertTrue(lines1 == lines2,
|
||||
"fileobj.__iter__() failed")
|
||||
|
||||
def test_fileobj_seek(self):
|
||||
self.tar.extract("ustar/regtype", TEMPDIR)
|
||||
fobj = open(os.path.join(TEMPDIR, "ustar/regtype"), "rb")
|
||||
with open(os.path.join(TEMPDIR, "ustar/regtype"), "rb") as fobj:
|
||||
data = fobj.read()
|
||||
fobj.close()
|
||||
|
||||
tarinfo = self.tar.getmember("ustar/regtype")
|
||||
fobj = self.tar.extractfile(tarinfo)
|
||||
|
|
@ -161,19 +160,24 @@ def test_empty_tarfile(self):
|
|||
# This test checks if tarfile.open() is able to open an empty tar
|
||||
# archive successfully. Note that an empty tar archive is not the
|
||||
# same as an empty file!
|
||||
tarfile.open(tmpname, self.mode.replace("r", "w")).close()
|
||||
with tarfile.open(tmpname, self.mode.replace("r", "w")):
|
||||
pass
|
||||
try:
|
||||
tar = tarfile.open(tmpname, self.mode)
|
||||
tar.getnames()
|
||||
except tarfile.ReadError:
|
||||
self.fail("tarfile.open() failed on empty archive")
|
||||
else:
|
||||
self.assertListEqual(tar.getmembers(), [])
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_null_tarfile(self):
|
||||
# Test for issue6123: Allow opening empty archives.
|
||||
# This test guarantees that tarfile.open() does not treat an empty
|
||||
# file as an empty tar archive.
|
||||
open(tmpname, "wb").close()
|
||||
with open(tmpname, "wb"):
|
||||
pass
|
||||
self.assertRaises(tarfile.ReadError, tarfile.open, tmpname, self.mode)
|
||||
self.assertRaises(tarfile.ReadError, tarfile.open, tmpname)
|
||||
|
||||
|
|
@ -189,33 +193,36 @@ def test_ignore_zeros(self):
|
|||
for char in (b'\0', b'a'):
|
||||
# Test if EOFHeaderError ('\0') and InvalidHeaderError ('a')
|
||||
# are ignored correctly.
|
||||
fobj = _open(tmpname, "wb")
|
||||
with _open(tmpname, "wb") as fobj:
|
||||
fobj.write(char * 1024)
|
||||
fobj.write(tarfile.TarInfo("foo").tobuf())
|
||||
fobj.close()
|
||||
|
||||
tar = tarfile.open(tmpname, mode="r", ignore_zeros=True)
|
||||
try:
|
||||
self.assertListEqual(tar.getnames(), ["foo"],
|
||||
"ignore_zeros=True should have skipped the %r-blocks" % char)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
|
||||
class MiscReadTest(CommonReadTest):
|
||||
|
||||
def test_no_name_argument(self):
|
||||
fobj = open(self.tarname, "rb")
|
||||
with open(self.tarname, "rb") as fobj:
|
||||
tar = tarfile.open(fileobj=fobj, mode=self.mode)
|
||||
self.assertEqual(tar.name, os.path.abspath(fobj.name))
|
||||
|
||||
def test_no_name_attribute(self):
|
||||
data = open(self.tarname, "rb").read()
|
||||
with open(self.tarname, "rb") as fobj:
|
||||
data = fobj.read()
|
||||
fobj = io.BytesIO(data)
|
||||
self.assertRaises(AttributeError, getattr, fobj, "name")
|
||||
tar = tarfile.open(fileobj=fobj, mode=self.mode)
|
||||
self.assertEqual(tar.name, None)
|
||||
|
||||
def test_empty_name_attribute(self):
|
||||
data = open(self.tarname, "rb").read()
|
||||
with open(self.tarname, "rb") as fobj:
|
||||
data = fobj.read()
|
||||
fobj = io.BytesIO(data)
|
||||
fobj.name = ""
|
||||
tar = tarfile.open(fileobj=fobj, mode=self.mode)
|
||||
|
|
@ -225,11 +232,13 @@ def test_fileobj_with_offset(self):
|
|||
# Skip the first member and store values from the second member
|
||||
# of the testtar.
|
||||
tar = tarfile.open(self.tarname, mode=self.mode)
|
||||
try:
|
||||
tar.next()
|
||||
t = tar.next()
|
||||
name = t.name
|
||||
offset = t.offset
|
||||
data = tar.extractfile(t).read()
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
# Open the testtar and seek to the offset of the second member.
|
||||
|
|
@ -240,6 +249,7 @@ def test_fileobj_with_offset(self):
|
|||
else:
|
||||
_open = open
|
||||
fobj = _open(self.tarname, "rb")
|
||||
try:
|
||||
fobj.seek(offset)
|
||||
|
||||
# Test if the tarfile starts with the second member.
|
||||
|
|
@ -252,14 +262,17 @@ def test_fileobj_with_offset(self):
|
|||
self.assertEqual(tar.extractfile(t).read(), data,
|
||||
"seek back did not work")
|
||||
tar.close()
|
||||
finally:
|
||||
fobj.close()
|
||||
|
||||
def test_fail_comp(self):
|
||||
# For Gzip and Bz2 Tests: fail with a ReadError on an uncompressed file.
|
||||
if self.mode == "r:":
|
||||
return
|
||||
self.assertRaises(tarfile.ReadError, tarfile.open, tarname, self.mode)
|
||||
fobj = open(tarname, "rb")
|
||||
self.assertRaises(tarfile.ReadError, tarfile.open, fileobj=fobj, mode=self.mode)
|
||||
with open(tarname, "rb") as fobj:
|
||||
self.assertRaises(tarfile.ReadError, tarfile.open,
|
||||
fileobj=fobj, mode=self.mode)
|
||||
|
||||
def test_v7_dirtype(self):
|
||||
# Test old style dirtype member (bug #1336623):
|
||||
|
|
@ -298,6 +311,7 @@ def test_extract_hardlink(self):
|
|||
# Test hardlink extraction (e.g. bug #857297).
|
||||
tar = tarfile.open(tarname, errorlevel=1, encoding="iso8859-1")
|
||||
|
||||
try:
|
||||
tar.extract("ustar/regtype", TEMPDIR)
|
||||
try:
|
||||
tar.extract("ustar/lnktype", TEMPDIR)
|
||||
|
|
@ -316,11 +330,14 @@ def test_extract_hardlink(self):
|
|||
|
||||
data = open(os.path.join(TEMPDIR, "ustar/symtype"), "rb").read()
|
||||
self.assertEqual(md5sum(data), md5_regtype)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_extractall(self):
|
||||
# Test if extractall() correctly restores directory permissions
|
||||
# and times (see issue1735).
|
||||
tar = tarfile.open(tarname, encoding="iso8859-1")
|
||||
try:
|
||||
directories = [t for t in tar if t.isdir()]
|
||||
tar.extractall(TEMPDIR, directories)
|
||||
for tarinfo in directories:
|
||||
|
|
@ -329,6 +346,7 @@ def test_extractall(self):
|
|||
# Win32 has no support for fine grained permissions.
|
||||
self.assertEqual(tarinfo.mode & 0o777, os.stat(path).st_mode & 0o777)
|
||||
self.assertEqual(tarinfo.mtime, os.path.getmtime(path))
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_init_close_fobj(self):
|
||||
|
|
@ -336,7 +354,8 @@ def test_init_close_fobj(self):
|
|||
# constructor in case of an error. For the test we rely on
|
||||
# the fact that opening an empty file raises a ReadError.
|
||||
empty = os.path.join(TEMPDIR, "empty")
|
||||
open(empty, "wb").write(b"")
|
||||
with open(empty, "wb") as fobj:
|
||||
fobj.write(b"")
|
||||
|
||||
try:
|
||||
tar = object.__new__(tarfile.TarFile)
|
||||
|
|
@ -347,7 +366,7 @@ def test_init_close_fobj(self):
|
|||
else:
|
||||
self.fail("ReadError not raised")
|
||||
finally:
|
||||
os.remove(empty)
|
||||
support.unlink(empty)
|
||||
|
||||
|
||||
class StreamReadTest(CommonReadTest):
|
||||
|
|
@ -368,6 +387,7 @@ def test_provoke_stream_error(self):
|
|||
|
||||
def test_compare_members(self):
|
||||
tar1 = tarfile.open(tarname, encoding="iso8859-1")
|
||||
try:
|
||||
tar2 = self.tar
|
||||
|
||||
while True:
|
||||
|
|
@ -387,7 +407,7 @@ def test_compare_members(self):
|
|||
continue
|
||||
self.assertTrue(v2 is not None, "stream.extractfile() failed")
|
||||
self.assertEqual(v1.read(), v2.read(), "stream extraction failed")
|
||||
|
||||
finally:
|
||||
tar1.close()
|
||||
|
||||
|
||||
|
|
@ -395,15 +415,19 @@ class DetectReadTest(unittest.TestCase):
|
|||
|
||||
def _testfunc_file(self, name, mode):
|
||||
try:
|
||||
tarfile.open(name, mode)
|
||||
tar = tarfile.open(name, mode)
|
||||
except tarfile.ReadError as e:
|
||||
self.fail()
|
||||
else:
|
||||
tar.close()
|
||||
|
||||
def _testfunc_fileobj(self, name, mode):
|
||||
try:
|
||||
tarfile.open(name, mode, fileobj=open(name, "rb"))
|
||||
tar = tarfile.open(name, mode, fileobj=open(name, "rb"))
|
||||
except tarfile.ReadError as e:
|
||||
self.fail()
|
||||
else:
|
||||
tar.close()
|
||||
|
||||
def _test_modes(self, testfunc):
|
||||
testfunc(tarname, "r")
|
||||
|
|
@ -579,7 +603,7 @@ class PaxReadTest(LongnameTest):
|
|||
|
||||
def test_pax_global_headers(self):
|
||||
tar = tarfile.open(tarname, encoding="iso8859-1")
|
||||
|
||||
try:
|
||||
tarinfo = tar.getmember("pax/regtype1")
|
||||
self.assertEqual(tarinfo.uname, "foo")
|
||||
self.assertEqual(tarinfo.gname, "bar")
|
||||
|
|
@ -594,10 +618,13 @@ def test_pax_global_headers(self):
|
|||
self.assertEqual(tarinfo.uname, "tarfile")
|
||||
self.assertEqual(tarinfo.gname, "tarfile")
|
||||
self.assertEqual(tarinfo.pax_headers.get("VENDOR.umlauts"), "\xc4\xd6\xdc\xe4\xf6\xfc\xdf")
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_pax_number_fields(self):
|
||||
# All following number fields are read from the pax header.
|
||||
tar = tarfile.open(tarname, encoding="iso8859-1")
|
||||
try:
|
||||
tarinfo = tar.getmember("pax/regtype4")
|
||||
self.assertEqual(tarinfo.size, 7011)
|
||||
self.assertEqual(tarinfo.uid, 123)
|
||||
|
|
@ -606,6 +633,8 @@ def test_pax_number_fields(self):
|
|||
self.assertEqual(type(tarinfo.mtime), float)
|
||||
self.assertEqual(float(tarinfo.pax_headers["atime"]), 1041808783.0)
|
||||
self.assertEqual(float(tarinfo.pax_headers["ctime"]), 1041808783.0)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
|
||||
class WriteTestBase(unittest.TestCase):
|
||||
|
|
@ -631,23 +660,28 @@ def test_100_char_name(self):
|
|||
# a trailing '\0'.
|
||||
name = "0123456789" * 10
|
||||
tar = tarfile.open(tmpname, self.mode)
|
||||
try:
|
||||
t = tarfile.TarInfo(name)
|
||||
tar.addfile(t)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tar = tarfile.open(tmpname)
|
||||
try:
|
||||
self.assertTrue(tar.getnames()[0] == name,
|
||||
"failed to store 100 char filename")
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_tar_size(self):
|
||||
# Test for bug #1013882.
|
||||
tar = tarfile.open(tmpname, self.mode)
|
||||
try:
|
||||
path = os.path.join(TEMPDIR, "file")
|
||||
fobj = open(path, "wb")
|
||||
with open(path, "wb") as fobj:
|
||||
fobj.write(b"aaa")
|
||||
fobj.close()
|
||||
tar.add(path)
|
||||
finally:
|
||||
tar.close()
|
||||
self.assertTrue(os.path.getsize(tmpname) > 0,
|
||||
"tarfile is empty")
|
||||
|
|
@ -655,19 +689,18 @@ def test_tar_size(self):
|
|||
# The test_*_size tests test for bug #1167128.
|
||||
def test_file_size(self):
|
||||
tar = tarfile.open(tmpname, self.mode)
|
||||
|
||||
try:
|
||||
path = os.path.join(TEMPDIR, "file")
|
||||
fobj = open(path, "wb")
|
||||
fobj.close()
|
||||
with open(path, "wb"):
|
||||
pass
|
||||
tarinfo = tar.gettarinfo(path)
|
||||
self.assertEqual(tarinfo.size, 0)
|
||||
|
||||
fobj = open(path, "wb")
|
||||
with open(path, "wb") as fobj:
|
||||
fobj.write(b"aaa")
|
||||
fobj.close()
|
||||
tarinfo = tar.gettarinfo(path)
|
||||
self.assertEqual(tarinfo.size, 3)
|
||||
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_directory_size(self):
|
||||
|
|
@ -675,8 +708,11 @@ def test_directory_size(self):
|
|||
os.mkdir(path)
|
||||
try:
|
||||
tar = tarfile.open(tmpname, self.mode)
|
||||
try:
|
||||
tarinfo = tar.gettarinfo(path)
|
||||
self.assertEqual(tarinfo.size, 0)
|
||||
finally:
|
||||
tar.close()
|
||||
finally:
|
||||
os.rmdir(path)
|
||||
|
||||
|
|
@ -684,16 +720,18 @@ def test_link_size(self):
|
|||
if hasattr(os, "link"):
|
||||
link = os.path.join(TEMPDIR, "link")
|
||||
target = os.path.join(TEMPDIR, "link_target")
|
||||
fobj = open(target, "wb")
|
||||
with open(target, "wb") as fobj:
|
||||
fobj.write(b"aaa")
|
||||
fobj.close()
|
||||
os.link(target, link)
|
||||
try:
|
||||
tar = tarfile.open(tmpname, self.mode)
|
||||
try:
|
||||
# Record the link target in the inodes list.
|
||||
tar.gettarinfo(target)
|
||||
tarinfo = tar.gettarinfo(link)
|
||||
self.assertEqual(tarinfo.size, 0)
|
||||
finally:
|
||||
tar.close()
|
||||
finally:
|
||||
os.remove(target)
|
||||
os.remove(link)
|
||||
|
|
@ -704,18 +742,20 @@ def test_symlink_size(self):
|
|||
os.symlink("link_target", path)
|
||||
try:
|
||||
tar = tarfile.open(tmpname, self.mode)
|
||||
try:
|
||||
tarinfo = tar.gettarinfo(path)
|
||||
self.assertEqual(tarinfo.size, 0)
|
||||
finally:
|
||||
tar.close()
|
||||
finally:
|
||||
os.remove(path)
|
||||
|
||||
def test_add_self(self):
|
||||
# Test for #1257255.
|
||||
dstname = os.path.abspath(tmpname)
|
||||
|
||||
tar = tarfile.open(tmpname, self.mode)
|
||||
try:
|
||||
self.assertTrue(tar.name == dstname, "archive name must be absolute")
|
||||
|
||||
tar.add(dstname)
|
||||
self.assertTrue(tar.getnames() == [], "added the archive to itself")
|
||||
|
||||
|
|
@ -724,6 +764,8 @@ def test_add_self(self):
|
|||
tar.add(dstname)
|
||||
os.chdir(cwd)
|
||||
self.assertTrue(tar.getnames() == [], "added the archive to itself")
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_exclude(self):
|
||||
tempdir = os.path.join(TEMPDIR, "exclude")
|
||||
|
|
@ -736,14 +778,19 @@ def test_exclude(self):
|
|||
exclude = os.path.isfile
|
||||
|
||||
tar = tarfile.open(tmpname, self.mode, encoding="iso8859-1")
|
||||
try:
|
||||
with support.check_warnings(("use the filter argument",
|
||||
DeprecationWarning)):
|
||||
tar.add(tempdir, arcname="empty_dir", exclude=exclude)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tar = tarfile.open(tmpname, "r")
|
||||
try:
|
||||
self.assertEqual(len(tar.getmembers()), 1)
|
||||
self.assertEqual(tar.getnames()[0], "empty_dir")
|
||||
finally:
|
||||
tar.close()
|
||||
finally:
|
||||
shutil.rmtree(tempdir)
|
||||
|
||||
|
|
@ -763,14 +810,18 @@ def filter(tarinfo):
|
|||
return tarinfo
|
||||
|
||||
tar = tarfile.open(tmpname, self.mode, encoding="iso8859-1")
|
||||
try:
|
||||
tar.add(tempdir, arcname="empty_dir", filter=filter)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tar = tarfile.open(tmpname, "r")
|
||||
try:
|
||||
for tarinfo in tar:
|
||||
self.assertEqual(tarinfo.uid, 123)
|
||||
self.assertEqual(tarinfo.uname, "foo")
|
||||
self.assertEqual(len(tar.getmembers()), 3)
|
||||
finally:
|
||||
tar.close()
|
||||
finally:
|
||||
shutil.rmtree(tempdir)
|
||||
|
|
@ -789,11 +840,15 @@ def _test_pathname(self, path, cmp_path=None, dir=False):
|
|||
os.mkdir(foo)
|
||||
|
||||
tar = tarfile.open(tmpname, self.mode)
|
||||
try:
|
||||
tar.add(foo, arcname=path)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tar = tarfile.open(tmpname, "r")
|
||||
try:
|
||||
t = tar.next()
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
if not dir:
|
||||
|
|
@ -832,15 +887,17 @@ def test_cwd(self):
|
|||
cwd = os.getcwd()
|
||||
os.chdir(TEMPDIR)
|
||||
try:
|
||||
open("foo", "w").close()
|
||||
|
||||
tar = tarfile.open(tmpname, self.mode)
|
||||
try:
|
||||
tar.add(".")
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tar = tarfile.open(tmpname, "r")
|
||||
try:
|
||||
for t in tar:
|
||||
self.assert_(t.name == "." or t.name.startswith("./"))
|
||||
finally:
|
||||
tar.close()
|
||||
finally:
|
||||
os.chdir(cwd)
|
||||
|
|
@ -856,19 +913,18 @@ def test_stream_padding(self):
|
|||
tar.close()
|
||||
|
||||
if self.mode.endswith("gz"):
|
||||
fobj = gzip.GzipFile(tmpname)
|
||||
with gzip.GzipFile(tmpname) as fobj:
|
||||
data = fobj.read()
|
||||
fobj.close()
|
||||
elif self.mode.endswith("bz2"):
|
||||
dec = bz2.BZ2Decompressor()
|
||||
data = open(tmpname, "rb").read()
|
||||
with open(tmpname, "rb") as fobj:
|
||||
data = fobj.read()
|
||||
data = dec.decompress(data)
|
||||
self.assertTrue(len(dec.unused_data) == 0,
|
||||
"found trailing data")
|
||||
else:
|
||||
fobj = open(tmpname, "rb")
|
||||
with open(tmpname, "rb") as fobj:
|
||||
data = fobj.read()
|
||||
fobj.close()
|
||||
|
||||
self.assertTrue(data.count(b"\0") == tarfile.RECORDSIZE,
|
||||
"incorrect zero padding")
|
||||
|
|
@ -923,16 +979,18 @@ def _test(self, name, link=None):
|
|||
tarinfo.type = tarfile.LNKTYPE
|
||||
|
||||
tar = tarfile.open(tmpname, "w")
|
||||
try:
|
||||
tar.format = tarfile.GNU_FORMAT
|
||||
tar.addfile(tarinfo)
|
||||
|
||||
v1 = self._calc_size(name, link)
|
||||
v2 = tar.offset
|
||||
self.assertTrue(v1 == v2, "GNU longname/longlink creation failed")
|
||||
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tar = tarfile.open(tmpname)
|
||||
try:
|
||||
member = tar.next()
|
||||
self.assertIsNotNone(member,
|
||||
"unable to read longname member")
|
||||
|
|
@ -940,6 +998,8 @@ def _test(self, name, link=None):
|
|||
"unable to read longname member")
|
||||
self.assertEqual(tarinfo.linkname, member.linkname,
|
||||
"unable to read longname member")
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_longname_1023(self):
|
||||
self._test(("longnam/" * 127) + "longnam")
|
||||
|
|
@ -979,9 +1039,8 @@ def setUp(self):
|
|||
self.foo = os.path.join(TEMPDIR, "foo")
|
||||
self.bar = os.path.join(TEMPDIR, "bar")
|
||||
|
||||
fobj = open(self.foo, "wb")
|
||||
with open(self.foo, "wb") as fobj:
|
||||
fobj.write(b"foo")
|
||||
fobj.close()
|
||||
|
||||
os.link(self.foo, self.bar)
|
||||
|
||||
|
|
@ -990,8 +1049,8 @@ def setUp(self):
|
|||
|
||||
def tearDown(self):
|
||||
self.tar.close()
|
||||
os.remove(self.foo)
|
||||
os.remove(self.bar)
|
||||
support.unlink(self.foo)
|
||||
support.unlink(self.bar)
|
||||
|
||||
def test_add_twice(self):
|
||||
# The same name will be added as a REGTYPE every
|
||||
|
|
@ -1022,16 +1081,21 @@ def _test(self, name, link=None):
|
|||
tarinfo.type = tarfile.LNKTYPE
|
||||
|
||||
tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT)
|
||||
try:
|
||||
tar.addfile(tarinfo)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tar = tarfile.open(tmpname)
|
||||
try:
|
||||
if link:
|
||||
l = tar.getmembers()[0].linkname
|
||||
self.assertTrue(link == l, "PAX longlink creation failed")
|
||||
else:
|
||||
n = tar.getmembers()[0].name
|
||||
self.assertTrue(name == n, "PAX longname creation failed")
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_pax_global_header(self):
|
||||
pax_headers = {
|
||||
|
|
@ -1043,14 +1107,16 @@ def test_pax_global_header(self):
|
|||
|
||||
tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT,
|
||||
pax_headers=pax_headers)
|
||||
try:
|
||||
tar.addfile(tarfile.TarInfo("test"))
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
# Test if the global header was written correctly.
|
||||
tar = tarfile.open(tmpname, encoding="iso8859-1")
|
||||
try:
|
||||
self.assertEqual(tar.pax_headers, pax_headers)
|
||||
self.assertEqual(tar.getmembers()[0].pax_headers, pax_headers)
|
||||
|
||||
# Test if all the fields are strings.
|
||||
for key, val in tar.pax_headers.items():
|
||||
self.assertTrue(type(key) is not bytes)
|
||||
|
|
@ -1060,6 +1126,8 @@ def test_pax_global_header(self):
|
|||
tarfile.PAX_NUMBER_FIELDS[key](val)
|
||||
except (TypeError, ValueError):
|
||||
self.fail("unable to convert pax header field")
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_pax_extended_header(self):
|
||||
# The fields from the pax header have priority over the
|
||||
|
|
@ -1067,18 +1135,23 @@ def test_pax_extended_header(self):
|
|||
pax_headers = {"path": "foo", "uid": "123"}
|
||||
|
||||
tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT, encoding="iso8859-1")
|
||||
try:
|
||||
t = tarfile.TarInfo()
|
||||
t.name = "\xe4\xf6\xfc" # non-ASCII
|
||||
t.uid = 8**8 # too large
|
||||
t.pax_headers = pax_headers
|
||||
tar.addfile(t)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tar = tarfile.open(tmpname, encoding="iso8859-1")
|
||||
try:
|
||||
t = tar.getmembers()[0]
|
||||
self.assertEqual(t.pax_headers, pax_headers)
|
||||
self.assertEqual(t.name, "foo")
|
||||
self.assertEqual(t.uid, 123)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
|
||||
class UstarUnicodeTest(unittest.TestCase):
|
||||
|
|
@ -1096,12 +1169,16 @@ def test_utf8_filename(self):
|
|||
|
||||
def _test_unicode_filename(self, encoding):
|
||||
tar = tarfile.open(tmpname, "w", format=self.format, encoding=encoding, errors="strict")
|
||||
try:
|
||||
name = "\xe4\xf6\xfc"
|
||||
tar.addfile(tarfile.TarInfo(name))
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tar = tarfile.open(tmpname, encoding=encoding)
|
||||
try:
|
||||
self.assertEqual(tar.getmembers()[0].name, name)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_unicode_filename_error(self):
|
||||
|
|
@ -1110,6 +1187,7 @@ def test_unicode_filename_error(self):
|
|||
return
|
||||
|
||||
tar = tarfile.open(tmpname, "w", format=self.format, encoding="ascii", errors="strict")
|
||||
try:
|
||||
tarinfo = tarfile.TarInfo()
|
||||
|
||||
tarinfo.name = "\xe4\xf6\xfc"
|
||||
|
|
@ -1118,14 +1196,18 @@ def test_unicode_filename_error(self):
|
|||
tarinfo.name = "foo"
|
||||
tarinfo.uname = "\xe4\xf6\xfc"
|
||||
self.assertRaises(UnicodeError, tar.addfile, tarinfo)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_unicode_argument(self):
|
||||
tar = tarfile.open(tarname, "r", encoding="iso8859-1", errors="strict")
|
||||
try:
|
||||
for t in tar:
|
||||
self.assertTrue(type(t.name) is str)
|
||||
self.assertTrue(type(t.linkname) is str)
|
||||
self.assertTrue(type(t.uname) is str)
|
||||
self.assertTrue(type(t.gname) is str)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
def test_uname_unicode(self):
|
||||
|
|
@ -1134,10 +1216,13 @@ def test_uname_unicode(self):
|
|||
t.gname = "\xe4\xf6\xfc"
|
||||
|
||||
tar = tarfile.open(tmpname, mode="w", format=self.format, encoding="iso8859-1")
|
||||
try:
|
||||
tar.addfile(t)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tar = tarfile.open(tmpname, encoding="iso8859-1")
|
||||
try:
|
||||
t = tar.getmember("foo")
|
||||
self.assertEqual(t.uname, "\xe4\xf6\xfc")
|
||||
self.assertEqual(t.gname, "\xe4\xf6\xfc")
|
||||
|
|
@ -1147,6 +1232,8 @@ def test_uname_unicode(self):
|
|||
t = tar.getmember("foo")
|
||||
self.assertEqual(t.uname, "\udce4\udcf6\udcfc")
|
||||
self.assertEqual(t.gname, "\udce4\udcf6\udcfc")
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
|
||||
class GNUUnicodeTest(UstarUnicodeTest):
|
||||
|
|
@ -1189,21 +1276,19 @@ def setUp(self):
|
|||
os.remove(self.tarname)
|
||||
|
||||
def _add_testfile(self, fileobj=None):
|
||||
tar = tarfile.open(self.tarname, "a", fileobj=fileobj)
|
||||
with tarfile.open(self.tarname, "a", fileobj=fileobj) as tar:
|
||||
tar.addfile(tarfile.TarInfo("bar"))
|
||||
tar.close()
|
||||
|
||||
def _create_testtar(self, mode="w:"):
|
||||
src = tarfile.open(tarname, encoding="iso8859-1")
|
||||
with tarfile.open(tarname, encoding="iso8859-1") as src:
|
||||
t = src.getmember("ustar/regtype")
|
||||
t.name = "foo"
|
||||
f = src.extractfile(t)
|
||||
tar = tarfile.open(self.tarname, mode)
|
||||
with tarfile.open(self.tarname, mode) as tar:
|
||||
tar.addfile(t, f)
|
||||
tar.close()
|
||||
|
||||
def _test(self, names=["bar"], fileobj=None):
|
||||
tar = tarfile.open(self.tarname, fileobj=fileobj)
|
||||
with tarfile.open(self.tarname, fileobj=fileobj) as tar:
|
||||
self.assertEqual(tar.getnames(), names)
|
||||
|
||||
def test_non_existing(self):
|
||||
|
|
@ -1223,7 +1308,8 @@ def test_empty_fileobj(self):
|
|||
|
||||
def test_fileobj(self):
|
||||
self._create_testtar()
|
||||
data = open(self.tarname, "rb").read()
|
||||
with open(self.tarname, "rb") as fobj:
|
||||
data = fobj.read()
|
||||
fobj = io.BytesIO(data)
|
||||
self._add_testfile(fobj)
|
||||
fobj.seek(0)
|
||||
|
|
@ -1249,7 +1335,8 @@ def test_append_bz2(self):
|
|||
# Append mode is supposed to fail if the tarfile to append to
|
||||
# does not end with a zero block.
|
||||
def _test_error(self, data):
|
||||
open(self.tarname, "wb").write(data)
|
||||
with open(self.tarname, "wb") as fobj:
|
||||
fobj.write(data)
|
||||
self.assertRaises(tarfile.ReadError, self._add_testfile)
|
||||
|
||||
def test_null(self):
|
||||
|
|
@ -1390,7 +1477,7 @@ def test_eof(self):
|
|||
def test_fileobj(self):
|
||||
# Test that __exit__() did not close the external file
|
||||
# object.
|
||||
fobj = open(tmpname, "wb")
|
||||
with open(tmpname, "wb") as fobj:
|
||||
try:
|
||||
with tarfile.open(fileobj=fobj, mode="w") as tar:
|
||||
raise Exception
|
||||
|
|
@ -1398,7 +1485,6 @@ def test_fileobj(self):
|
|||
pass
|
||||
self.assertFalse(fobj.closed, "external file object was closed")
|
||||
self.assertTrue(tar.closed, "context manager failed")
|
||||
fobj.close()
|
||||
|
||||
|
||||
class LinkEmulationTest(ReadTest):
|
||||
|
|
@ -1495,6 +1581,7 @@ def test_partial_input_bz2(self):
|
|||
|
||||
|
||||
def test_main():
|
||||
support.unlink(TEMPDIR)
|
||||
os.makedirs(TEMPDIR)
|
||||
|
||||
tests = [
|
||||
|
|
@ -1523,15 +1610,14 @@ def test_main():
|
|||
else:
|
||||
tests.append(LinkEmulationTest)
|
||||
|
||||
fobj = open(tarname, "rb")
|
||||
with open(tarname, "rb") as fobj:
|
||||
data = fobj.read()
|
||||
fobj.close()
|
||||
|
||||
if gzip:
|
||||
# Create testtar.tar.gz and add gzip-specific tests.
|
||||
tar = gzip.open(gzipname, "wb")
|
||||
support.unlink(gzipname)
|
||||
with gzip.open(gzipname, "wb") as tar:
|
||||
tar.write(data)
|
||||
tar.close()
|
||||
|
||||
tests += [
|
||||
GzipMiscReadTest,
|
||||
|
|
@ -1543,8 +1629,11 @@ def test_main():
|
|||
|
||||
if bz2:
|
||||
# Create testtar.tar.bz2 and add bz2-specific tests.
|
||||
support.unlink(bz2name)
|
||||
tar = bz2.BZ2File(bz2name, "wb")
|
||||
try:
|
||||
tar.write(data)
|
||||
finally:
|
||||
tar.close()
|
||||
|
||||
tests += [
|
||||
|
|
|
|||
Loading…
Reference in New Issue