Lib/test/test_tarfile.py

   1 # -*- coding: iso-8859-15 -*-
   2
   3 import sys
   4 import os
   5 import shutil
   6 import StringIO
   7 from hashlib import md5
   8 import errno
   9
  10 import unittest
  11 import tarfile
  12
  13 from test import test_support
  14
  15 # Check for our compression modules.
  16 try:
  17     import gzip
  18     gzip.GzipFile
  19 except (ImportError, AttributeError):
  20     gzip = None
  21 try:
  22     import bz2
  23 except ImportError:
  24     bz2 = None
  25
  26 def md5sum(data):
  27     return md5(data).hexdigest()
  28
  29 TEMPDIR = os.path.abspath(test_support.TESTFN)
  30 tarname = test_support.findfile("testtar.tar")
  31 gzipname = os.path.join(TEMPDIR, "testtar.tar.gz")
  32 bz2name = os.path.join(TEMPDIR, "testtar.tar.bz2")
  33 tmpname = os.path.join(TEMPDIR, "tmp.tar")
  34
  35 md5_regtype = "65f477c818ad9e15f7feab0c6d37742f"
  36 md5_sparse = "a54fbc4ca4f4399a90e1b27164012fc6"
  37
  38
  39 class ReadTest(unittest.TestCase):
  40
  41     tarname = tarname
  42     mode = "r:"
  43
  44     def setUp(self):
  45         self.tar = tarfile.open(self.tarname, mode=self.mode, encoding="iso8859-1")
  46
  47     def tearDown(self):
  48         self.tar.close()
  49
  50
  51 class UstarReadTest(ReadTest):
  52
  53     def test_fileobj_regular_file(self):
  54         tarinfo = self.tar.getmember("ustar/regtype")
  55         fobj = self.tar.extractfile(tarinfo)
  56         data = fobj.read()
  57         self.assertTrue((len(data), md5sum(data)) == (tarinfo.size, md5_regtype),
  58                 "regular file extraction failed")
  59
  60     def test_fileobj_readlines(self):
  61         self.tar.extract("ustar/regtype", TEMPDIR)
  62         tarinfo = self.tar.getmember("ustar/regtype")
  63         fobj1 = open(os.path.join(TEMPDIR, "ustar/regtype"), "rU")
  64         fobj2 = self.tar.extractfile(tarinfo)
  65
  66         lines1 = fobj1.readlines()
  67         lines2 = fobj2.readlines()
  68         self.assertTrue(lines1 == lines2,
  69                 "fileobj.readlines() failed")
  70         self.assertTrue(len(lines2) == 114,
  71                 "fileobj.readlines() failed")
  72         self.assertTrue(lines2[83] ==
  73                 "I will gladly admit that Python is not the fastest running scripting language.\n",
  74                 "fileobj.readlines() failed")
  75
  76     def test_fileobj_iter(self):
  77         self.tar.extract("ustar/regtype", TEMPDIR)
  78         tarinfo = self.tar.getmember("ustar/regtype")
  79         fobj1 = open(os.path.join(TEMPDIR, "ustar/regtype"), "rU")
  80         fobj2 = self.tar.extractfile(tarinfo)
  81         lines1 = fobj1.readlines()
  82         lines2 = [line for line in fobj2]
  83         self.assertTrue(lines1 == lines2,
  84                      "fileobj.__iter__() failed")
  85
  86     def test_fileobj_seek(self):
  87         self.tar.extract("ustar/regtype", TEMPDIR)
  88         fobj = open(os.path.join(TEMPDIR, "ustar/regtype"), "rb")
  89         data = fobj.read()
  90         fobj.close()
  91
  92         tarinfo = self.tar.getmember("ustar/regtype")
  93         fobj = self.tar.extractfile(tarinfo)
  94
  95         text = fobj.read()
  96         fobj.seek(0)
  97         self.assertTrue(0 == fobj.tell(),
  98                      "seek() to file's start failed")
  99         fobj.seek(2048, 0)
 100         self.assertTrue(2048 == fobj.tell(),
 101                      "seek() to absolute position failed")
 102         fobj.seek(-1024, 1)
 103         self.assertTrue(1024 == fobj.tell(),
 104                      "seek() to negative relative position failed")
 105         fobj.seek(1024, 1)
 106         self.assertTrue(2048 == fobj.tell(),
 107                      "seek() to positive relative position failed")
 108         s = fobj.read(10)
 109         self.assertTrue(s == data[2048:2058],
 110                      "read() after seek failed")
 111         fobj.seek(0, 2)
 112         self.assertTrue(tarinfo.size == fobj.tell(),
 113                      "seek() to file's end failed")
 114         self.assertTrue(fobj.read() == "",
 115                      "read() at file's end did not return empty string")
 116         fobj.seek(-tarinfo.size, 2)
 117         self.assertTrue(0 == fobj.tell(),
 118                      "relative seek() to file's start failed")
 119         fobj.seek(512)
 120         s1 = fobj.readlines()
 121         fobj.seek(512)
 122         s2 = fobj.readlines()
 123         self.assertTrue(s1 == s2,
 124                      "readlines() after seek failed")
 125         fobj.seek(0)
 126         self.assertTrue(len(fobj.readline()) == fobj.tell(),
 127                      "tell() after readline() failed")
 128         fobj.seek(512)
 129         self.assertTrue(len(fobj.readline()) + 512 == fobj.tell(),
 130                      "tell() after seek() and readline() failed")
 131         fobj.seek(0)
 132         line = fobj.readline()
 133         self.assertTrue(fobj.read() == data[len(line):],
 134                      "read() after readline() failed")
 135         fobj.close()
 136
 137     # Test if symbolic and hard links are resolved by extractfile().  The
 138     # test link members each point to a regular member whose data is
 139     # supposed to be exported.
 140     def _test_fileobj_link(self, lnktype, regtype):
 141         a = self.tar.extractfile(lnktype)
 142         b = self.tar.extractfile(regtype)
 143         self.assertEqual(a.name, b.name)
 144
 145     def test_fileobj_link1(self):
 146         self._test_fileobj_link("ustar/lnktype", "ustar/regtype")
 147
 148     def test_fileobj_link2(self):
 149         self._test_fileobj_link("./ustar/linktest2/lnktype", "ustar/linktest1/regtype")
 150
 151     def test_fileobj_symlink1(self):
 152         self._test_fileobj_link("ustar/symtype", "ustar/regtype")
 153
 154     def test_fileobj_symlink2(self):
 155         self._test_fileobj_link("./ustar/linktest2/symtype", "ustar/linktest1/regtype")
 156
 157
 158 class CommonReadTest(ReadTest):
 159
 160     def test_empty_tarfile(self):
 161         # Test for issue6123: Allow opening empty archives.
 162         # This test checks if tarfile.open() is able to open an empty tar
 163         # archive successfully. Note that an empty tar archive is not the
 164         # same as an empty file!
 165         tarfile.open(tmpname, self.mode.replace("r", "w")).close()
 166         try:
 167             tar = tarfile.open(tmpname, self.mode)
 168             tar.getnames()
 169         except tarfile.ReadError:
 170             self.fail("tarfile.open() failed on empty archive")
 171         self.assertListEqual(tar.getmembers(), [])
 172
 173     def test_null_tarfile(self):
 174         # Test for issue6123: Allow opening empty archives.
 175         # This test guarantees that tarfile.open() does not treat an empty
 176         # file as an empty tar archive.
 177         open(tmpname, "wb").close()
 178         self.assertRaises(tarfile.ReadError, tarfile.open, tmpname, self.mode)
 179         self.assertRaises(tarfile.ReadError, tarfile.open, tmpname)
 180
 181     def test_ignore_zeros(self):
 182         # Test TarFile's ignore_zeros option.
 183         if self.mode.endswith(":gz"):
 184             _open = gzip.GzipFile
 185         elif self.mode.endswith(":bz2"):
 186             _open = bz2.BZ2File
 187         else:
 188             _open = open
 189
 190         for char in ('\0', 'a'):
 191             # Test if EOFHeaderError ('\0') and InvalidHeaderError ('a')
 192             # are ignored correctly.
 193             fobj = _open(tmpname, "wb")
 194             fobj.write(char * 1024)
 195             fobj.write(tarfile.TarInfo("foo").tobuf())
 196             fobj.close()
 197
 198             tar = tarfile.open(tmpname, mode="r", ignore_zeros=True)
 199             self.assertListEqual(tar.getnames(), ["foo"],
 200                     "ignore_zeros=True should have skipped the %r-blocks" % char)
 201             tar.close()
 202
 203
 204 class MiscReadTest(CommonReadTest):
 205
 206     def test_no_name_argument(self):
 207         fobj = open(self.tarname, "rb")
 208         tar = tarfile.open(fileobj=fobj, mode=self.mode)
 209         self.assertEqual(tar.name, os.path.abspath(fobj.name))
 210
 211     def test_no_name_attribute(self):
 212         data = open(self.tarname, "rb").read()
 213         fobj = StringIO.StringIO(data)
 214         self.assertRaises(AttributeError, getattr, fobj, "name")
 215         tar = tarfile.open(fileobj=fobj, mode=self.mode)
 216         self.assertEqual(tar.name, None)
 217
 218     def test_empty_name_attribute(self):
 219         data = open(self.tarname, "rb").read()
 220         fobj = StringIO.StringIO(data)
 221         fobj.name = ""
 222         tar = tarfile.open(fileobj=fobj, mode=self.mode)
 223         self.assertEqual(tar.name, None)
 224
 225     def test_fileobj_with_offset(self):
 226         # Skip the first member and store values from the second member
 227         # of the testtar.
 228         tar = tarfile.open(self.tarname, mode=self.mode)
 229         tar.next()
 230         t = tar.next()
 231         name = t.name
 232         offset = t.offset
 233         data = tar.extractfile(t).read()
 234         tar.close()
 235
 236         # Open the testtar and seek to the offset of the second member.
 237         if self.mode.endswith(":gz"):
 238             _open = gzip.GzipFile
 239         elif self.mode.endswith(":bz2"):
 240             _open = bz2.BZ2File
 241         else:
 242             _open = open
 243         fobj = _open(self.tarname, "rb")
 244         fobj.seek(offset)
 245
 246         # Test if the tarfile starts with the second member.
 247         tar = tar.open(self.tarname, mode="r:", fileobj=fobj)
 248         t = tar.next()
 249         self.assertEqual(t.name, name)
 250         # Read to the end of fileobj and test if seeking back to the
 251         # beginning works.
 252         tar.getmembers()
 253         self.assertEqual(tar.extractfile(t).read(), data,
 254                 "seek back did not work")
 255         tar.close()
 256
 257     def test_fail_comp(self):
 258         # For Gzip and Bz2 Tests: fail with a ReadError on an uncompressed file.
 259         if self.mode == "r:":
 260             return
 261         self.assertRaises(tarfile.ReadError, tarfile.open, tarname, self.mode)
 262         fobj = open(tarname, "rb")
 263         self.assertRaises(tarfile.ReadError, tarfile.open, fileobj=fobj, mode=self.mode)
 264
 265     def test_v7_dirtype(self):
 266         # Test old style dirtype member (bug #1336623):
 267         # Old V7 tars create directory members using an AREGTYPE
 268         # header with a "/" appended to the filename field.
 269         tarinfo = self.tar.getmember("misc/dirtype-old-v7")
 270         self.assertTrue(tarinfo.type == tarfile.DIRTYPE,
 271                 "v7 dirtype failed")
 272
 273     def test_xstar_type(self):
 274         # The xstar format stores extra atime and ctime fields inside the
 275         # space reserved for the prefix field. The prefix field must be
 276         # ignored in this case, otherwise it will mess up the name.
 277         try:
 278             self.tar.getmember("misc/regtype-xstar")
 279         except KeyError:
 280             self.fail("failed to find misc/regtype-xstar (mangled prefix?)")
 281
 282     def test_check_members(self):
 283         for tarinfo in self.tar:
 284             self.assertTrue(int(tarinfo.mtime) == 07606136617,
 285                     "wrong mtime for %s" % tarinfo.name)
 286             if not tarinfo.name.startswith("ustar/"):
 287                 continue
 288             self.assertTrue(tarinfo.uname == "tarfile",
 289                     "wrong uname for %s" % tarinfo.name)
 290
 291     def test_find_members(self):
 292         self.assertTrue(self.tar.getmembers()[-1].name == "misc/eof",
 293                 "could not find all members")
 294
 295     def test_extract_hardlink(self):
 296         # Test hardlink extraction (e.g. bug #857297).
 297         tar = tarfile.open(tarname, errorlevel=1, encoding="iso8859-1")
 298
 299         tar.extract("ustar/regtype", TEMPDIR)
 300         try:
 301             tar.extract("ustar/lnktype", TEMPDIR)
 302         except EnvironmentError, e:
 303             if e.errno == errno.ENOENT:
 304                 self.fail("hardlink not extracted properly")
 305
 306         data = open(os.path.join(TEMPDIR, "ustar/lnktype"), "rb").read()
 307         self.assertEqual(md5sum(data), md5_regtype)
 308
 309         try:
 310             tar.extract("ustar/symtype", TEMPDIR)
 311         except EnvironmentError, e:
 312             if e.errno == errno.ENOENT:
 313                 self.fail("symlink not extracted properly")
 314
 315         data = open(os.path.join(TEMPDIR, "ustar/symtype"), "rb").read()
 316         self.assertEqual(md5sum(data), md5_regtype)
 317
 318     def test_extractall(self):
 319         # Test if extractall() correctly restores directory permissions
 320         # and times (see issue1735).
 321         tar = tarfile.open(tarname, encoding="iso8859-1")
 322         directories = [t for t in tar if t.isdir()]
 323         tar.extractall(TEMPDIR, directories)
 324         for tarinfo in directories:
 325             path = os.path.join(TEMPDIR, tarinfo.name)
 326             if sys.platform != "win32":
 327                 # Win32 has no support for fine grained permissions.
 328                 self.assertEqual(tarinfo.mode & 0777, os.stat(path).st_mode & 0777)
 329             self.assertEqual(tarinfo.mtime, os.path.getmtime(path))
 330         tar.close()
 331
 332     def test_init_close_fobj(self):
 333         # Issue #7341: Close the internal file object in the TarFile
 334         # constructor in case of an error. For the test we rely on
 335         # the fact that opening an empty file raises a ReadError.
 336         empty = os.path.join(TEMPDIR, "empty")
 337         open(empty, "wb").write("")
 338
 339         try:
 340             tar = object.__new__(tarfile.TarFile)
 341             try:
 342                 tar.__init__(empty)
 343             except tarfile.ReadError:
 344                 self.assertTrue(tar.fileobj.closed)
 345             else:
 346                 self.fail("ReadError not raised")
 347         finally:
 348             os.remove(empty)
 349
 350
 351 class StreamReadTest(CommonReadTest):
 352
 353     mode="r|"
 354
 355     def test_fileobj_regular_file(self):
 356         tarinfo = self.tar.next() # get "regtype" (can't use getmember)
 357         fobj = self.tar.extractfile(tarinfo)
 358         data = fobj.read()
 359         self.assertTrue((len(data), md5sum(data)) == (tarinfo.size, md5_regtype),
 360                 "regular file extraction failed")
 361
 362     def test_provoke_stream_error(self):
 363         tarinfos = self.tar.getmembers()
 364         f = self.tar.extractfile(tarinfos[0]) # read the first member
 365         self.assertRaises(tarfile.StreamError, f.read)
 366
 367     def test_compare_members(self):
 368         tar1 = tarfile.open(tarname, encoding="iso8859-1")
 369         tar2 = self.tar
 370
 371         while True:
 372             t1 = tar1.next()
 373             t2 = tar2.next()
 374             if t1 is None:
 375                 break
 376             self.assertTrue(t2 is not None, "stream.next() failed.")
 377
 378             if t2.islnk() or t2.issym():
 379                 self.assertRaises(tarfile.StreamError, tar2.extractfile, t2)
 380                 continue
 381
 382             v1 = tar1.extractfile(t1)
 383             v2 = tar2.extractfile(t2)
 384             if v1 is None:
 385                 continue
 386             self.assertTrue(v2 is not None, "stream.extractfile() failed")
 387             self.assertTrue(v1.read() == v2.read(), "stream extraction failed")
 388
 389         tar1.close()
 390
 391
 392 class DetectReadTest(unittest.TestCase):
 393
 394     def _testfunc_file(self, name, mode):
 395         try:
 396             tarfile.open(name, mode)
 397         except tarfile.ReadError:
 398             self.fail()
 399
 400     def _testfunc_fileobj(self, name, mode):
 401         try:
 402             tarfile.open(name, mode, fileobj=open(name, "rb"))
 403         except tarfile.ReadError:
 404             self.fail()
 405
 406     def _test_modes(self, testfunc):
 407         testfunc(tarname, "r")
 408         testfunc(tarname, "r:")
 409         testfunc(tarname, "r:*")
 410         testfunc(tarname, "r|")
 411         testfunc(tarname, "r|*")
 412
 413         if gzip:
 414             self.assertRaises(tarfile.ReadError, tarfile.open, tarname, mode="r:gz")
 415             self.assertRaises(tarfile.ReadError, tarfile.open, tarname, mode="r|gz")
 416             self.assertRaises(tarfile.ReadError, tarfile.open, gzipname, mode="r:")
 417             self.assertRaises(tarfile.ReadError, tarfile.open, gzipname, mode="r|")
 418
 419             testfunc(gzipname, "r")
 420             testfunc(gzipname, "r:*")
 421             testfunc(gzipname, "r:gz")
 422             testfunc(gzipname, "r|*")
 423             testfunc(gzipname, "r|gz")
 424
 425         if bz2:
 426             self.assertRaises(tarfile.ReadError, tarfile.open, tarname, mode="r:bz2")
 427             self.assertRaises(tarfile.ReadError, tarfile.open, tarname, mode="r|bz2")
 428             self.assertRaises(tarfile.ReadError, tarfile.open, bz2name, mode="r:")
 429             self.assertRaises(tarfile.ReadError, tarfile.open, bz2name, mode="r|")
 430
 431             testfunc(bz2name, "r")
 432             testfunc(bz2name, "r:*")
 433             testfunc(bz2name, "r:bz2")
 434             testfunc(bz2name, "r|*")
 435             testfunc(bz2name, "r|bz2")
 436
 437     def test_detect_file(self):
 438         self._test_modes(self._testfunc_file)
 439
 440     def test_detect_fileobj(self):
 441         self._test_modes(self._testfunc_fileobj)
 442
 443
 444 class MemberReadTest(ReadTest):
 445
 446     def _test_member(self, tarinfo, chksum=None, **kwargs):
 447         if chksum is not None:
 448             self.assertTrue(md5sum(self.tar.extractfile(tarinfo).read()) == chksum,
 449                     "wrong md5sum for %s" % tarinfo.name)
 450
 451         kwargs["mtime"] = 07606136617
 452         kwargs["uid"] = 1000
 453         kwargs["gid"] = 100
 454         if "old-v7" not in tarinfo.name:
 455             # V7 tar can't handle alphabetic owners.
 456             kwargs["uname"] = "tarfile"
 457             kwargs["gname"] = "tarfile"
 458         for k, v in kwargs.iteritems():
 459             self.assertTrue(getattr(tarinfo, k) == v,
 460                     "wrong value in %s field of %s" % (k, tarinfo.name))
 461
 462     def test_find_regtype(self):
 463         tarinfo = self.tar.getmember("ustar/regtype")
 464         self._test_member(tarinfo, size=7011, chksum=md5_regtype)
 465
 466     def test_find_conttype(self):
 467         tarinfo = self.tar.getmember("ustar/conttype")
 468         self._test_member(tarinfo, size=7011, chksum=md5_regtype)
 469
 470     def test_find_dirtype(self):
 471         tarinfo = self.tar.getmember("ustar/dirtype")
 472         self._test_member(tarinfo, size=0)
 473
 474     def test_find_dirtype_with_size(self):
 475         tarinfo = self.tar.getmember("ustar/dirtype-with-size")
 476         self._test_member(tarinfo, size=255)
 477
 478     def test_find_lnktype(self):
 479         tarinfo = self.tar.getmember("ustar/lnktype")
 480         self._test_member(tarinfo, size=0, linkname="ustar/regtype")
 481
 482     def test_find_symtype(self):
 483         tarinfo = self.tar.getmember("ustar/symtype")
 484         self._test_member(tarinfo, size=0, linkname="regtype")
 485
 486     def test_find_blktype(self):
 487         tarinfo = self.tar.getmember("ustar/blktype")
 488         self._test_member(tarinfo, size=0, devmajor=3, devminor=0)
 489
 490     def test_find_chrtype(self):
 491         tarinfo = self.tar.getmember("ustar/chrtype")
 492         self._test_member(tarinfo, size=0, devmajor=1, devminor=3)
 493
 494     def test_find_fifotype(self):
 495         tarinfo = self.tar.getmember("ustar/fifotype")
 496         self._test_member(tarinfo, size=0)
 497
 498     def test_find_sparse(self):
 499         tarinfo = self.tar.getmember("ustar/sparse")
 500         self._test_member(tarinfo, size=86016, chksum=md5_sparse)
 501
 502     def test_find_umlauts(self):
 503         tarinfo = self.tar.getmember("ustar/umlauts-ÄÖÜäöüß")
 504         self._test_member(tarinfo, size=7011, chksum=md5_regtype)
 505
 506     def test_find_ustar_longname(self):
 507         name = "ustar/" + "12345/" * 39 + "1234567/longname"
 508         self.assertIn(name, self.tar.getnames())
 509
 510     def test_find_regtype_oldv7(self):
 511         tarinfo = self.tar.getmember("misc/regtype-old-v7")
 512         self._test_member(tarinfo, size=7011, chksum=md5_regtype)
 513
 514     def test_find_pax_umlauts(self):
 515         self.tar = tarfile.open(self.tarname, mode=self.mode, encoding="iso8859-1")
 516         tarinfo = self.tar.getmember("pax/umlauts-ÄÖÜäöüß")
 517         self._test_member(tarinfo, size=7011, chksum=md5_regtype)
 518
 519
 520 class LongnameTest(ReadTest):
 521
 522     def test_read_longname(self):
 523         # Test reading of longname (bug #1471427).
 524         longname = self.subdir + "/" + "123/" * 125 + "longname"
 525         try:
 526             tarinfo = self.tar.getmember(longname)
 527         except KeyError:
 528             self.fail("longname not found")
 529         self.assertTrue(tarinfo.type != tarfile.DIRTYPE, "read longname as dirtype")
 530
 531     def test_read_longlink(self):
 532         longname = self.subdir + "/" + "123/" * 125 + "longname"
 533         longlink = self.subdir + "/" + "123/" * 125 + "longlink"
 534         try:
 535             tarinfo = self.tar.getmember(longlink)
 536         except KeyError:
 537             self.fail("longlink not found")
 538         self.assertTrue(tarinfo.linkname == longname, "linkname wrong")
 539
 540     def test_truncated_longname(self):
 541         longname = self.subdir + "/" + "123/" * 125 + "longname"
 542         tarinfo = self.tar.getmember(longname)
 543         offset = tarinfo.offset
 544         self.tar.fileobj.seek(offset)
 545         fobj = StringIO.StringIO(self.tar.fileobj.read(3 * 512))
 546         self.assertRaises(tarfile.ReadError, tarfile.open, name="foo.tar", fileobj=fobj)
 547
 548     def test_header_offset(self):
 549         # Test if the start offset of the TarInfo object includes
 550         # the preceding extended header.
 551         longname = self.subdir + "/" + "123/" * 125 + "longname"
 552         offset = self.tar.getmember(longname).offset
 553         fobj = open(tarname)
 554         fobj.seek(offset)
 555         tarinfo = tarfile.TarInfo.frombuf(fobj.read(512))
 556         self.assertEqual(tarinfo.type, self.longnametype)
 557
 558
 559 class GNUReadTest(LongnameTest):
 560
 561     subdir = "gnu"
 562     longnametype = tarfile.GNUTYPE_LONGNAME
 563
 564     def test_sparse_file(self):
 565         tarinfo1 = self.tar.getmember("ustar/sparse")
 566         fobj1 = self.tar.extractfile(tarinfo1)
 567         tarinfo2 = self.tar.getmember("gnu/sparse")
 568         fobj2 = self.tar.extractfile(tarinfo2)
 569         self.assertTrue(fobj1.read() == fobj2.read(),
 570                 "sparse file extraction failed")
 571
 572
 573 class PaxReadTest(LongnameTest):
 574
 575     subdir = "pax"
 576     longnametype = tarfile.XHDTYPE
 577
 578     def test_pax_global_headers(self):
 579         tar = tarfile.open(tarname, encoding="iso8859-1")
 580
 581         tarinfo = tar.getmember("pax/regtype1")
 582         self.assertEqual(tarinfo.uname, "foo")
 583         self.assertEqual(tarinfo.gname, "bar")
 584         self.assertEqual(tarinfo.pax_headers.get("VENDOR.umlauts"), u"ÄÖÜäöüß")
 585
 586         tarinfo = tar.getmember("pax/regtype2")
 587         self.assertEqual(tarinfo.uname, "")
 588         self.assertEqual(tarinfo.gname, "bar")
 589         self.assertEqual(tarinfo.pax_headers.get("VENDOR.umlauts"), u"ÄÖÜäöüß")
 590
 591         tarinfo = tar.getmember("pax/regtype3")
 592         self.assertEqual(tarinfo.uname, "tarfile")
 593         self.assertEqual(tarinfo.gname, "tarfile")
 594         self.assertEqual(tarinfo.pax_headers.get("VENDOR.umlauts"), u"ÄÖÜäöüß")
 595
 596     def test_pax_number_fields(self):
 597         # All following number fields are read from the pax header.
 598         tar = tarfile.open(tarname, encoding="iso8859-1")
 599         tarinfo = tar.getmember("pax/regtype4")
 600         self.assertEqual(tarinfo.size, 7011)
 601         self.assertEqual(tarinfo.uid, 123)
 602         self.assertEqual(tarinfo.gid, 123)
 603         self.assertEqual(tarinfo.mtime, 1041808783.0)
 604         self.assertEqual(type(tarinfo.mtime), float)
 605         self.assertEqual(float(tarinfo.pax_headers["atime"]), 1041808783.0)
 606         self.assertEqual(float(tarinfo.pax_headers["ctime"]), 1041808783.0)
 607
 608
 609 class WriteTestBase(unittest.TestCase):
 610     # Put all write tests in here that are supposed to be tested
 611     # in all possible mode combinations.
 612
 613     def test_fileobj_no_close(self):
 614         fobj = StringIO.StringIO()
 615         tar = tarfile.open(fileobj=fobj, mode=self.mode)
 616         tar.addfile(tarfile.TarInfo("foo"))
 617         tar.close()
 618         self.assertTrue(fobj.closed is False, "external fileobjs must never closed")
 619
 620
 621 class WriteTest(WriteTestBase):
 622
 623     mode = "w:"
 624
 625     def test_100_char_name(self):
 626         # The name field in a tar header stores strings of at most 100 chars.
 627         # If a string is shorter than 100 chars it has to be padded with '\0',
 628         # which implies that a string of exactly 100 chars is stored without
 629         # a trailing '\0'.
 630         name = "0123456789" * 10
 631         tar = tarfile.open(tmpname, self.mode)
 632         t = tarfile.TarInfo(name)
 633         tar.addfile(t)
 634         tar.close()
 635
 636         tar = tarfile.open(tmpname)
 637         self.assertTrue(tar.getnames()[0] == name,
 638                 "failed to store 100 char filename")
 639         tar.close()
 640
 641     def test_tar_size(self):
 642         # Test for bug #1013882.
 643         tar = tarfile.open(tmpname, self.mode)
 644         path = os.path.join(TEMPDIR, "file")
 645         fobj = open(path, "wb")
 646         fobj.write("aaa")
 647         fobj.close()
 648         tar.add(path)
 649         tar.close()
 650         self.assertTrue(os.path.getsize(tmpname) > 0,
 651                 "tarfile is empty")
 652
 653     # The test_*_size tests test for bug #1167128.
 654     def test_file_size(self):
 655         tar = tarfile.open(tmpname, self.mode)
 656
 657         path = os.path.join(TEMPDIR, "file")
 658         fobj = open(path, "wb")
 659         fobj.close()
 660         tarinfo = tar.gettarinfo(path)
 661         self.assertEqual(tarinfo.size, 0)
 662
 663         fobj = open(path, "wb")
 664         fobj.write("aaa")
 665         fobj.close()
 666         tarinfo = tar.gettarinfo(path)
 667         self.assertEqual(tarinfo.size, 3)
 668
 669         tar.close()
 670
 671     def test_directory_size(self):
 672         path = os.path.join(TEMPDIR, "directory")
 673         os.mkdir(path)
 674         try:
 675             tar = tarfile.open(tmpname, self.mode)
 676             tarinfo = tar.gettarinfo(path)
 677             self.assertEqual(tarinfo.size, 0)
 678         finally:
 679             os.rmdir(path)
 680
 681     def test_link_size(self):
 682         if hasattr(os, "link"):
 683             link = os.path.join(TEMPDIR, "link")
 684             target = os.path.join(TEMPDIR, "link_target")
 685             fobj = open(target, "wb")
 686             fobj.write("aaa")
 687             fobj.close()
 688             os.link(target, link)
 689             try:
 690                 tar = tarfile.open(tmpname, self.mode)
 691                 # Record the link target in the inodes list.
 692                 tar.gettarinfo(target)
 693                 tarinfo = tar.gettarinfo(link)
 694                 self.assertEqual(tarinfo.size, 0)
 695             finally:
 696                 os.remove(target)
 697                 os.remove(link)
 698
 699     def test_symlink_size(self):
 700         if hasattr(os, "symlink"):
 701             path = os.path.join(TEMPDIR, "symlink")
 702             os.symlink("link_target", path)
 703             try:
 704                 tar = tarfile.open(tmpname, self.mode)
 705                 tarinfo = tar.gettarinfo(path)
 706                 self.assertEqual(tarinfo.size, 0)
 707             finally:
 708                 os.remove(path)
 709
 710     def test_add_self(self):
 711         # Test for #1257255.
 712         dstname = os.path.abspath(tmpname)
 713
 714         tar = tarfile.open(tmpname, self.mode)
 715         self.assertTrue(tar.name == dstname, "archive name must be absolute")
 716
 717         tar.add(dstname)
 718         self.assertTrue(tar.getnames() == [], "added the archive to itself")
 719
 720         cwd = os.getcwd()
 721         os.chdir(TEMPDIR)
 722         tar.add(dstname)
 723         os.chdir(cwd)
 724         self.assertTrue(tar.getnames() == [], "added the archive to itself")
 725
 726     def test_exclude(self):
 727         tempdir = os.path.join(TEMPDIR, "exclude")
 728         os.mkdir(tempdir)
 729         try:
 730             for name in ("foo", "bar", "baz"):
 731                 name = os.path.join(tempdir, name)
 732                 open(name, "wb").close()
 733
 734             exclude = os.path.isfile
 735
 736             tar = tarfile.open(tmpname, self.mode, encoding="iso8859-1")
 737             with test_support.check_warnings(("use the filter argument",
 738                                               DeprecationWarning)):
 739                 tar.add(tempdir, arcname="empty_dir", exclude=exclude)
 740             tar.close()
 741
 742             tar = tarfile.open(tmpname, "r")
 743             self.assertEqual(len(tar.getmembers()), 1)
 744             self.assertEqual(tar.getnames()[0], "empty_dir")
 745         finally:
 746             shutil.rmtree(tempdir)
 747
 748     def test_filter(self):
 749         tempdir = os.path.join(TEMPDIR, "filter")
 750         os.mkdir(tempdir)
 751         try:
 752             for name in ("foo", "bar", "baz"):
 753                 name = os.path.join(tempdir, name)
 754                 open(name, "wb").close()
 755
 756             def filter(tarinfo):
 757                 if os.path.basename(tarinfo.name) == "bar":
 758                     return
 759                 tarinfo.uid = 123
 760                 tarinfo.uname = "foo"
 761                 return tarinfo
 762
 763             tar = tarfile.open(tmpname, self.mode, encoding="iso8859-1")
 764             tar.add(tempdir, arcname="empty_dir", filter=filter)
 765             tar.close()
 766
 767             tar = tarfile.open(tmpname, "r")
 768             for tarinfo in tar:
 769                 self.assertEqual(tarinfo.uid, 123)
 770                 self.assertEqual(tarinfo.uname, "foo")
 771             self.assertEqual(len(tar.getmembers()), 3)
 772             tar.close()
 773         finally:
 774             shutil.rmtree(tempdir)
 775
 776     # Guarantee that stored pathnames are not modified. Don't
 777     # remove ./ or ../ or double slashes. Still make absolute
 778     # pathnames relative.
 779     # For details see bug #6054.
 780     def _test_pathname(self, path, cmp_path=None, dir=False):
 781         # Create a tarfile with an empty member named path
 782         # and compare the stored name with the original.
 783         foo = os.path.join(TEMPDIR, "foo")
 784         if not dir:
 785             open(foo, "w").close()
 786         else:
 787             os.mkdir(foo)
 788
 789         tar = tarfile.open(tmpname, self.mode)
 790         tar.add(foo, arcname=path)
 791         tar.close()
 792
 793         tar = tarfile.open(tmpname, "r")
 794         t = tar.next()
 795         tar.close()
 796
 797         if not dir:
 798             os.remove(foo)
 799         else:
 800             os.rmdir(foo)
 801
 802         self.assertEqual(t.name, cmp_path or path.replace(os.sep, "/"))
 803
 804     def test_pathnames(self):
 805         self._test_pathname("foo")
 806         self._test_pathname(os.path.join("foo", ".", "bar"))
 807         self._test_pathname(os.path.join("foo", "..", "bar"))
 808         self._test_pathname(os.path.join(".", "foo"))
 809         self._test_pathname(os.path.join(".", "foo", "."))
 810         self._test_pathname(os.path.join(".", "foo", ".", "bar"))
 811         self._test_pathname(os.path.join(".", "foo", "..", "bar"))
 812         self._test_pathname(os.path.join(".", "foo", "..", "bar"))
 813         self._test_pathname(os.path.join("..", "foo"))
 814         self._test_pathname(os.path.join("..", "foo", ".."))
 815         self._test_pathname(os.path.join("..", "foo", ".", "bar"))
 816         self._test_pathname(os.path.join("..", "foo", "..", "bar"))
 817
 818         self._test_pathname("foo" + os.sep + os.sep + "bar")
 819         self._test_pathname("foo" + os.sep + os.sep, "foo", dir=True)
 820
 821     def test_abs_pathnames(self):
 822         if sys.platform == "win32":
 823             self._test_pathname("C:\\foo", "foo")
 824         else:
 825             self._test_pathname("/foo", "foo")
 826             self._test_pathname("///foo", "foo")
 827
 828     def test_cwd(self):
 829         # Test adding the current working directory.
 830         cwd = os.getcwd()
 831         os.chdir(TEMPDIR)
 832         try:
 833             open("foo", "w").close()
 834
 835             tar = tarfile.open(tmpname, self.mode)
 836             tar.add(".")
 837             tar.close()
 838
 839             tar = tarfile.open(tmpname, "r")
 840             for t in tar:
 841                 self.assert_(t.name == "." or t.name.startswith("./"))
 842             tar.close()
 843         finally:
 844             os.chdir(cwd)
 845
 846
 847 class StreamWriteTest(WriteTestBase):
 848
 849     mode = "w|"
 850
 851     def test_stream_padding(self):
 852         # Test for bug #1543303.
 853         tar = tarfile.open(tmpname, self.mode)
 854         tar.close()
 855
 856         if self.mode.endswith("gz"):
 857             fobj = gzip.GzipFile(tmpname)
 858             data = fobj.read()
 859             fobj.close()
 860         elif self.mode.endswith("bz2"):
 861             dec = bz2.BZ2Decompressor()
 862             data = open(tmpname, "rb").read()
 863             data = dec.decompress(data)
 864             self.assertTrue(len(dec.unused_data) == 0,
 865                     "found trailing data")
 866         else:
 867             fobj = open(tmpname, "rb")
 868             data = fobj.read()
 869             fobj.close()
 870
 871         self.assertTrue(data.count("\0") == tarfile.RECORDSIZE,
 872                          "incorrect zero padding")
 873
 874     def test_file_mode(self):
 875         # Test for issue #8464: Create files with correct
 876         # permissions.
 877         if sys.platform == "win32" or not hasattr(os, "umask"):
 878             return
 879
 880         if os.path.exists(tmpname):
 881             os.remove(tmpname)
 882
 883         original_umask = os.umask(0022)
 884         try:
 885             tar = tarfile.open(tmpname, self.mode)
 886             tar.close()
 887             mode = os.stat(tmpname).st_mode & 0777
 888             self.assertEqual(mode, 0644, "wrong file permissions")
 889         finally:
 890             os.umask(original_umask)
 891
 892
 893 class GNUWriteTest(unittest.TestCase):
 894     # This testcase checks for correct creation of GNU Longname
 895     # and Longlink extended headers (cp. bug #812325).
 896
 897     def _length(self, s):
 898         blocks, remainder = divmod(len(s) + 1, 512)
 899         if remainder:
 900             blocks += 1
 901         return blocks * 512
 902
 903     def _calc_size(self, name, link=None):
 904         # Initial tar header
 905         count = 512
 906
 907         if len(name) > tarfile.LENGTH_NAME:
 908             # GNU longname extended header + longname
 909             count += 512
 910             count += self._length(name)
 911         if link is not None and len(link) > tarfile.LENGTH_LINK:
 912             # GNU longlink extended header + longlink
 913             count += 512
 914             count += self._length(link)
 915         return count
 916
 917     def _test(self, name, link=None):
 918         tarinfo = tarfile.TarInfo(name)
 919         if link:
 920             tarinfo.linkname = link
 921             tarinfo.type = tarfile.LNKTYPE
 922
 923         tar = tarfile.open(tmpname, "w")
 924         tar.format = tarfile.GNU_FORMAT
 925         tar.addfile(tarinfo)
 926
 927         v1 = self._calc_size(name, link)
 928         v2 = tar.offset
 929         self.assertTrue(v1 == v2, "GNU longname/longlink creation failed")
 930
 931         tar.close()
 932
 933         tar = tarfile.open(tmpname)
 934         member = tar.next()
 935         self.assertIsNotNone(member,
 936                 "unable to read longname member")
 937         self.assertEqual(tarinfo.name, member.name,
 938                 "unable to read longname member")
 939         self.assertEqual(tarinfo.linkname, member.linkname,
 940                 "unable to read longname member")
 941
 942     def test_longname_1023(self):
 943         self._test(("longnam/" * 127) + "longnam")
 944
 945     def test_longname_1024(self):
 946         self._test(("longnam/" * 127) + "longname")
 947
 948     def test_longname_1025(self):
 949         self._test(("longnam/" * 127) + "longname_")
 950
 951     def test_longlink_1023(self):
 952         self._test("name", ("longlnk/" * 127) + "longlnk")
 953
 954     def test_longlink_1024(self):
 955         self._test("name", ("longlnk/" * 127) + "longlink")
 956
 957     def test_longlink_1025(self):
 958         self._test("name", ("longlnk/" * 127) + "longlink_")
 959
 960     def test_longnamelink_1023(self):
 961         self._test(("longnam/" * 127) + "longnam",
 962                    ("longlnk/" * 127) + "longlnk")
 963
 964     def test_longnamelink_1024(self):
 965         self._test(("longnam/" * 127) + "longname",
 966                    ("longlnk/" * 127) + "longlink")
 967
 968     def test_longnamelink_1025(self):
 969         self._test(("longnam/" * 127) + "longname_",
 970                    ("longlnk/" * 127) + "longlink_")
 971
 972
 973 class HardlinkTest(unittest.TestCase):
 974     # Test the creation of LNKTYPE (hardlink) members in an archive.
 975
 976     def setUp(self):
 977         self.foo = os.path.join(TEMPDIR, "foo")
 978         self.bar = os.path.join(TEMPDIR, "bar")
 979
 980         fobj = open(self.foo, "wb")
 981         fobj.write("foo")
 982         fobj.close()
 983
 984         os.link(self.foo, self.bar)
 985
 986         self.tar = tarfile.open(tmpname, "w")
 987         self.tar.add(self.foo)
 988
 989     def tearDown(self):
 990         self.tar.close()
 991         os.remove(self.foo)
 992         os.remove(self.bar)
 993
 994     def test_add_twice(self):
 995         # The same name will be added as a REGTYPE every
 996         # time regardless of st_nlink.
 997         tarinfo = self.tar.gettarinfo(self.foo)
 998         self.assertTrue(tarinfo.type == tarfile.REGTYPE,
 999                 "add file as regular failed")
1000
1001     def test_add_hardlink(self):
1002         tarinfo = self.tar.gettarinfo(self.bar)
1003         self.assertTrue(tarinfo.type == tarfile.LNKTYPE,
1004                 "add file as hardlink failed")
1005
1006     def test_dereference_hardlink(self):
1007         self.tar.dereference = True
1008         tarinfo = self.tar.gettarinfo(self.bar)
1009         self.assertTrue(tarinfo.type == tarfile.REGTYPE,
1010                 "dereferencing hardlink failed")
1011
1012
1013 class PaxWriteTest(GNUWriteTest):
1014
1015     def _test(self, name, link=None):
1016         # See GNUWriteTest.
1017         tarinfo = tarfile.TarInfo(name)
1018         if link:
1019             tarinfo.linkname = link
1020             tarinfo.type = tarfile.LNKTYPE
1021
1022         tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT)
1023         tar.addfile(tarinfo)
1024         tar.close()
1025
1026         tar = tarfile.open(tmpname)
1027         if link:
1028             l = tar.getmembers()[0].linkname
1029             self.assertTrue(link == l, "PAX longlink creation failed")
1030         else:
1031             n = tar.getmembers()[0].name
1032             self.assertTrue(name == n, "PAX longname creation failed")
1033
1034     def test_pax_global_header(self):
1035         pax_headers = {
1036                 u"foo": u"bar",
1037                 u"uid": u"0",
1038                 u"mtime": u"1.23",
1039                 u"test": u"äöü",
1040                 u"äöü": u"test"}
1041
1042         tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT,
1043                 pax_headers=pax_headers)
1044         tar.addfile(tarfile.TarInfo("test"))
1045         tar.close()
1046
1047         # Test if the global header was written correctly.
1048         tar = tarfile.open(tmpname, encoding="iso8859-1")
1049         self.assertEqual(tar.pax_headers, pax_headers)
1050         self.assertEqual(tar.getmembers()[0].pax_headers, pax_headers)
1051
1052         # Test if all the fields are unicode.
1053         for key, val in tar.pax_headers.iteritems():
1054             self.assertTrue(type(key) is unicode)
1055             self.assertTrue(type(val) is unicode)
1056             if key in tarfile.PAX_NUMBER_FIELDS:
1057                 try:
1058                     tarfile.PAX_NUMBER_FIELDS[key](val)
1059                 except (TypeError, ValueError):
1060                     self.fail("unable to convert pax header field")
1061
1062     def test_pax_extended_header(self):
1063         # The fields from the pax header have priority over the
1064         # TarInfo.
1065         pax_headers = {u"path": u"foo", u"uid": u"123"}
1066
1067         tar = tarfile.open(tmpname, "w", format=tarfile.PAX_FORMAT, encoding="iso8859-1")
1068         t = tarfile.TarInfo()
1069         t.name = u"äöü"     # non-ASCII
1070         t.uid = 8**8        # too large
1071         t.pax_headers = pax_headers
1072         tar.addfile(t)
1073         tar.close()
1074
1075         tar = tarfile.open(tmpname, encoding="iso8859-1")
1076         t = tar.getmembers()[0]
1077         self.assertEqual(t.pax_headers, pax_headers)
1078         self.assertEqual(t.name, "foo")
1079         self.assertEqual(t.uid, 123)
1080
1081
1082 class UstarUnicodeTest(unittest.TestCase):
1083     # All *UnicodeTests FIXME
1084
1085     format = tarfile.USTAR_FORMAT
1086
1087     def test_iso8859_1_filename(self):
1088         self._test_unicode_filename("iso8859-1")
1089
1090     def test_utf7_filename(self):
1091         self._test_unicode_filename("utf7")
1092
1093     def test_utf8_filename(self):
1094         self._test_unicode_filename("utf8")
1095
1096     def _test_unicode_filename(self, encoding):
1097         tar = tarfile.open(tmpname, "w", format=self.format, encoding=encoding, errors="strict")
1098         name = u"äöü"
1099         tar.addfile(tarfile.TarInfo(name))
1100         tar.close()
1101
1102         tar = tarfile.open(tmpname, encoding=encoding)
1103         self.assertTrue(type(tar.getnames()[0]) is not unicode)
1104         self.assertEqual(tar.getmembers()[0].name, name.encode(encoding))
1105         tar.close()
1106
1107     def test_unicode_filename_error(self):
1108         tar = tarfile.open(tmpname, "w", format=self.format, encoding="ascii", errors="strict")
1109         tarinfo = tarfile.TarInfo()
1110
1111         tarinfo.name = "äöü"
1112         if self.format == tarfile.PAX_FORMAT:
1113             self.assertRaises(UnicodeError, tar.addfile, tarinfo)
1114         else:
1115             tar.addfile(tarinfo)
1116
1117         tarinfo.name = u"äöü"
1118         self.assertRaises(UnicodeError, tar.addfile, tarinfo)
1119
1120         tarinfo.name = "foo"
1121         tarinfo.uname = u"äöü"
1122         self.assertRaises(UnicodeError, tar.addfile, tarinfo)
1123
1124     def test_unicode_argument(self):
1125         tar = tarfile.open(tarname, "r", encoding="iso8859-1", errors="strict")
1126         for t in tar:
1127             self.assertTrue(type(t.name) is str)
1128             self.assertTrue(type(t.linkname) is str)
1129             self.assertTrue(type(t.uname) is str)
1130             self.assertTrue(type(t.gname) is str)
1131         tar.close()
1132
1133     def test_uname_unicode(self):
1134         for name in (u"äöü", "äöü"):
1135             t = tarfile.TarInfo("foo")
1136             t.uname = name
1137             t.gname = name
1138
1139             fobj = StringIO.StringIO()
1140             tar = tarfile.open("foo.tar", mode="w", fileobj=fobj, format=self.format, encoding="iso8859-1")
1141             tar.addfile(t)
1142             tar.close()
1143             fobj.seek(0)
1144
1145             tar = tarfile.open("foo.tar", fileobj=fobj, encoding="iso8859-1")
1146             t = tar.getmember("foo")
1147             self.assertEqual(t.uname, "äöü")
1148             self.assertEqual(t.gname, "äöü")
1149
1150
1151 class GNUUnicodeTest(UstarUnicodeTest):
1152
1153     format = tarfile.GNU_FORMAT
1154
1155
1156 class PaxUnicodeTest(UstarUnicodeTest):
1157
1158     format = tarfile.PAX_FORMAT
1159
1160     def _create_unicode_name(self, name):
1161         tar = tarfile.open(tmpname, "w", format=self.format)
1162         t = tarfile.TarInfo()
1163         t.pax_headers["path"] = name
1164         tar.addfile(t)
1165         tar.close()
1166
1167     def test_error_handlers(self):
1168         # Test if the unicode error handlers work correctly for characters
1169         # that cannot be expressed in a given encoding.
1170         self._create_unicode_name(u"äöü")
1171
1172         for handler, name in (("utf-8", u"äöü".encode("utf8")),
1173                     ("replace", "???"), ("ignore", "")):
1174             tar = tarfile.open(tmpname, format=self.format, encoding="ascii",
1175                     errors=handler)
1176             self.assertEqual(tar.getnames()[0], name)
1177
1178         self.assertRaises(UnicodeError, tarfile.open, tmpname,
1179                 encoding="ascii", errors="strict")
1180
1181     def test_error_handler_utf8(self):
1182         # Create a pathname that has one component representable using
1183         # iso8859-1 and the other only in iso8859-15.
1184         self._create_unicode_name(u"äöü/¤")
1185
1186         tar = tarfile.open(tmpname, format=self.format, encoding="iso8859-1",
1187                 errors="utf-8")
1188         self.assertEqual(tar.getnames()[0], "äöü/" + u"¤".encode("utf8"))
1189
1190
1191 class AppendTest(unittest.TestCase):
1192     # Test append mode (cp. patch #1652681).
1193
1194     def setUp(self):
1195         self.tarname = tmpname
1196         if os.path.exists(self.tarname):
1197             os.remove(self.tarname)
1198
1199     def _add_testfile(self, fileobj=None):
1200         tar = tarfile.open(self.tarname, "a", fileobj=fileobj)
1201         tar.addfile(tarfile.TarInfo("bar"))
1202         tar.close()
1203
1204     def _create_testtar(self, mode="w:"):
1205         src = tarfile.open(tarname, encoding="iso8859-1")
1206         t = src.getmember("ustar/regtype")
1207         t.name = "foo"
1208         f = src.extractfile(t)
1209         tar = tarfile.open(self.tarname, mode)
1210         tar.addfile(t, f)
1211         tar.close()
1212
1213     def _test(self, names=["bar"], fileobj=None):
1214         tar = tarfile.open(self.tarname, fileobj=fileobj)
1215         self.assertEqual(tar.getnames(), names)
1216
1217     def test_non_existing(self):
1218         self._add_testfile()
1219         self._test()
1220
1221     def test_empty(self):
1222         tarfile.open(self.tarname, "w:").close()
1223         self._add_testfile()
1224         self._test()
1225
1226     def test_empty_fileobj(self):
1227         fobj = StringIO.StringIO("\0" * 1024)
1228         self._add_testfile(fobj)
1229         fobj.seek(0)
1230         self._test(fileobj=fobj)
1231
1232     def test_fileobj(self):
1233         self._create_testtar()
1234         data = open(self.tarname).read()
1235         fobj = StringIO.StringIO(data)
1236         self._add_testfile(fobj)
1237         fobj.seek(0)
1238         self._test(names=["foo", "bar"], fileobj=fobj)
1239
1240     def test_existing(self):
1241         self._create_testtar()
1242         self._add_testfile()
1243         self._test(names=["foo", "bar"])
1244
1245     def test_append_gz(self):
1246         if gzip is None:
1247             return
1248         self._create_testtar("w:gz")
1249         self.assertRaises(tarfile.ReadError, tarfile.open, tmpname, "a")
1250
1251     def test_append_bz2(self):
1252         if bz2 is None:
1253             return
1254         self._create_testtar("w:bz2")
1255         self.assertRaises(tarfile.ReadError, tarfile.open, tmpname, "a")
1256
1257     # Append mode is supposed to fail if the tarfile to append to
1258     # does not end with a zero block.
1259     def _test_error(self, data):
1260         open(self.tarname, "wb").write(data)
1261         self.assertRaises(tarfile.ReadError, self._add_testfile)
1262
1263     def test_null(self):
1264         self._test_error("")
1265
1266     def test_incomplete(self):
1267         self._test_error("\0" * 13)
1268
1269     def test_premature_eof(self):
1270         data = tarfile.TarInfo("foo").tobuf()
1271         self._test_error(data)
1272
1273     def test_trailing_garbage(self):
1274         data = tarfile.TarInfo("foo").tobuf()
1275         self._test_error(data + "\0" * 13)
1276
1277     def test_invalid(self):
1278         self._test_error("a" * 512)
1279
1280
1281 class LimitsTest(unittest.TestCase):
1282
1283     def test_ustar_limits(self):
1284         # 100 char name
1285         tarinfo = tarfile.TarInfo("0123456789" * 10)
1286         tarinfo.tobuf(tarfile.USTAR_FORMAT)
1287
1288         # 101 char name that cannot be stored
1289         tarinfo = tarfile.TarInfo("0123456789" * 10 + "0")
1290         self.assertRaises(ValueError, tarinfo.tobuf, tarfile.USTAR_FORMAT)
1291
1292         # 256 char name with a slash at pos 156
1293         tarinfo = tarfile.TarInfo("123/" * 62 + "longname")
1294         tarinfo.tobuf(tarfile.USTAR_FORMAT)
1295
1296         # 256 char name that cannot be stored
1297         tarinfo = tarfile.TarInfo("1234567/" * 31 + "longname")
1298         self.assertRaises(ValueError, tarinfo.tobuf, tarfile.USTAR_FORMAT)
1299
1300         # 512 char name
1301         tarinfo = tarfile.TarInfo("123/" * 126 + "longname")
1302         self.assertRaises(ValueError, tarinfo.tobuf, tarfile.USTAR_FORMAT)
1303
1304         # 512 char linkname
1305         tarinfo = tarfile.TarInfo("longlink")
1306         tarinfo.linkname = "123/" * 126 + "longname"
1307         self.assertRaises(ValueError, tarinfo.tobuf, tarfile.USTAR_FORMAT)
1308
1309         # uid > 8 digits
1310         tarinfo = tarfile.TarInfo("name")
1311         tarinfo.uid = 010000000
1312         self.assertRaises(ValueError, tarinfo.tobuf, tarfile.USTAR_FORMAT)
1313
1314     def test_gnu_limits(self):
1315         tarinfo = tarfile.TarInfo("123/" * 126 + "longname")
1316         tarinfo.tobuf(tarfile.GNU_FORMAT)
1317
1318         tarinfo = tarfile.TarInfo("longlink")
1319         tarinfo.linkname = "123/" * 126 + "longname"
1320         tarinfo.tobuf(tarfile.GNU_FORMAT)
1321
1322         # uid >= 256 ** 7
1323         tarinfo = tarfile.TarInfo("name")
1324         tarinfo.uid = 04000000000000000000L
1325         self.assertRaises(ValueError, tarinfo.tobuf, tarfile.GNU_FORMAT)
1326
1327     def test_pax_limits(self):
1328         tarinfo = tarfile.TarInfo("123/" * 126 + "longname")
1329         tarinfo.tobuf(tarfile.PAX_FORMAT)
1330
1331         tarinfo = tarfile.TarInfo("longlink")
1332         tarinfo.linkname = "123/" * 126 + "longname"
1333         tarinfo.tobuf(tarfile.PAX_FORMAT)
1334
1335         tarinfo = tarfile.TarInfo("name")
1336         tarinfo.uid = 04000000000000000000L
1337         tarinfo.tobuf(tarfile.PAX_FORMAT)
1338
1339
1340 class ContextManagerTest(unittest.TestCase):
1341
1342     def test_basic(self):
1343         with tarfile.open(tarname) as tar:
1344             self.assertFalse(tar.closed, "closed inside runtime context")
1345         self.assertTrue(tar.closed, "context manager failed")
1346
1347     def test_closed(self):
1348         # The __enter__() method is supposed to raise IOError
1349         # if the TarFile object is already closed.
1350         tar = tarfile.open(tarname)
1351         tar.close()
1352         with self.assertRaises(IOError):
1353             with tar:
1354                 pass
1355
1356     def test_exception(self):
1357         # Test if the IOError exception is passed through properly.
1358         with self.assertRaises(Exception) as exc:
1359             with tarfile.open(tarname) as tar:
1360                 raise IOError
1361         self.assertIsInstance(exc.exception, IOError,
1362                               "wrong exception raised in context manager")
1363         self.assertTrue(tar.closed, "context manager failed")
1364
1365     def test_no_eof(self):
1366         # __exit__() must not write end-of-archive blocks if an
1367         # exception was raised.
1368         try:
1369             with tarfile.open(tmpname, "w") as tar:
1370                 raise Exception
1371         except:
1372             pass
1373         self.assertEqual(os.path.getsize(tmpname), 0,
1374                 "context manager wrote an end-of-archive block")
1375         self.assertTrue(tar.closed, "context manager failed")
1376
1377     def test_eof(self):
1378         # __exit__() must write end-of-archive blocks, i.e. call
1379         # TarFile.close() if there was no error.
1380         with tarfile.open(tmpname, "w"):
1381             pass
1382         self.assertNotEqual(os.path.getsize(tmpname), 0,
1383                 "context manager wrote no end-of-archive block")
1384
1385     def test_fileobj(self):
1386         # Test that __exit__() did not close the external file
1387         # object.
1388         fobj = open(tmpname, "wb")
1389         try:
1390             with tarfile.open(fileobj=fobj, mode="w") as tar:
1391                 raise Exception
1392         except:
1393             pass
1394         self.assertFalse(fobj.closed, "external file object was closed")
1395         self.assertTrue(tar.closed, "context manager failed")
1396         fobj.close()
1397
1398
1399 class LinkEmulationTest(ReadTest):
1400
1401     # Test for issue #8741 regression. On platforms that do not support
1402     # symbolic or hard links tarfile tries to extract these types of members as
1403     # the regular files they point to.
1404     def _test_link_extraction(self, name):
1405         self.tar.extract(name, TEMPDIR)
1406         data = open(os.path.join(TEMPDIR, name), "rb").read()
1407         self.assertEqual(md5sum(data), md5_regtype)
1408
1409     def test_hardlink_extraction1(self):
1410         self._test_link_extraction("ustar/lnktype")
1411
1412     def test_hardlink_extraction2(self):
1413         self._test_link_extraction("./ustar/linktest2/lnktype")
1414
1415     def test_symlink_extraction1(self):
1416         self._test_link_extraction("ustar/symtype")
1417
1418     def test_symlink_extraction2(self):
1419         self._test_link_extraction("./ustar/linktest2/symtype")
1420
1421
1422 class GzipMiscReadTest(MiscReadTest):
1423     tarname = gzipname
1424     mode = "r:gz"
1425 class GzipUstarReadTest(UstarReadTest):
1426     tarname = gzipname
1427     mode = "r:gz"
1428 class GzipStreamReadTest(StreamReadTest):
1429     tarname = gzipname
1430     mode = "r|gz"
1431 class GzipWriteTest(WriteTest):
1432     mode = "w:gz"
1433 class GzipStreamWriteTest(StreamWriteTest):
1434     mode = "w|gz"
1435
1436
1437 class Bz2MiscReadTest(MiscReadTest):
1438     tarname = bz2name
1439     mode = "r:bz2"
1440 class Bz2UstarReadTest(UstarReadTest):
1441     tarname = bz2name
1442     mode = "r:bz2"
1443 class Bz2StreamReadTest(StreamReadTest):
1444     tarname = bz2name
1445     mode = "r|bz2"
1446 class Bz2WriteTest(WriteTest):
1447     mode = "w:bz2"
1448 class Bz2StreamWriteTest(StreamWriteTest):
1449     mode = "w|bz2"
1450
1451 class Bz2PartialReadTest(unittest.TestCase):
1452     # Issue5068: The _BZ2Proxy.read() method loops forever
1453     # on an empty or partial bzipped file.
1454
1455     def _test_partial_input(self, mode):
1456         class MyStringIO(StringIO.StringIO):
1457             hit_eof = False
1458             def read(self, n):
1459                 if self.hit_eof:
1460                     raise AssertionError("infinite loop detected in tarfile.open()")
1461                 self.hit_eof = self.pos == self.len
1462                 return StringIO.StringIO.read(self, n)
1463             def seek(self, *args):
1464                 self.hit_eof = False
1465                 return StringIO.StringIO.seek(self, *args)
1466
1467         data = bz2.compress(tarfile.TarInfo("foo").tobuf())
1468         for x in range(len(data) + 1):
1469             try:
1470                 tarfile.open(fileobj=MyStringIO(data[:x]), mode=mode)
1471             except tarfile.ReadError:
1472                 pass # we have no interest in ReadErrors
1473
1474     def test_partial_input(self):
1475         self._test_partial_input("r")
1476
1477     def test_partial_input_bz2(self):
1478         self._test_partial_input("r:bz2")
1479
1480
1481 def test_main():
1482     os.makedirs(TEMPDIR)
1483
1484     tests = [
1485         UstarReadTest,
1486         MiscReadTest,
1487         StreamReadTest,
1488         DetectReadTest,
1489         MemberReadTest,
1490         GNUReadTest,
1491         PaxReadTest,
1492         WriteTest,
1493         StreamWriteTest,
1494         GNUWriteTest,
1495         PaxWriteTest,
1496         UstarUnicodeTest,
1497         GNUUnicodeTest,
1498         PaxUnicodeTest,
1499         AppendTest,
1500         LimitsTest,
1501         ContextManagerTest,
1502     ]
1503
1504     if hasattr(os, "link"):
1505         tests.append(HardlinkTest)
1506     else:
1507         tests.append(LinkEmulationTest)
1508
1509     fobj = open(tarname, "rb")
1510     data = fobj.read()
1511     fobj.close()
1512
1513     if gzip:
1514         # Create testtar.tar.gz and add gzip-specific tests.
1515         tar = gzip.open(gzipname, "wb")
1516         tar.write(data)
1517         tar.close()
1518
1519         tests += [
1520             GzipMiscReadTest,
1521             GzipUstarReadTest,
1522             GzipStreamReadTest,
1523             GzipWriteTest,
1524             GzipStreamWriteTest,
1525         ]
1526
1527     if bz2:
1528         # Create testtar.tar.bz2 and add bz2-specific tests.
1529         tar = bz2.BZ2File(bz2name, "wb")
1530         tar.write(data)
1531         tar.close()
1532
1533         tests += [
1534             Bz2MiscReadTest,
1535             Bz2UstarReadTest,
1536             Bz2StreamReadTest,
1537             Bz2WriteTest,
1538             Bz2StreamWriteTest,
1539             Bz2PartialReadTest,
1540         ]
1541
1542     try:
1543         test_support.run_unittest(*tests)
1544     finally:
1545         if os.path.exists(TEMPDIR):
1546             shutil.rmtree(TEMPDIR)
1547
1548 if __name__ == "__main__":
1549     test_main()