import sys import os import shutil import tempfile import unittest import tarfile from test import test_support # Check for our compression modules. try: import gzip gzip.GzipFile except (ImportError, AttributeError): gzip = None try: import bz2 except ImportError: bz2 = None def path(path): return test_support.findfile(path) testtar = path("testtar.tar") tempdir = os.path.join(tempfile.gettempdir(), "testtar" + os.extsep + "dir") tempname = test_support.TESTFN membercount = 10 def tarname(comp=""): if not comp: return testtar return os.path.join(tempdir, "%s%s%s" % (testtar, os.extsep, comp)) def dirname(): if not os.path.exists(tempdir): os.mkdir(tempdir) return tempdir def tmpname(): return tempname class BaseTest(unittest.TestCase): comp = '' mode = 'r' sep = ':' def setUp(self): mode = self.mode + self.sep + self.comp self.tar = tarfile.open(tarname(self.comp), mode) def tearDown(self): self.tar.close() class ReadTest(BaseTest): def test(self): """Test member extraction. """ members = 0 for tarinfo in self.tar: members += 1 if not tarinfo.isreg(): continue f = self.tar.extractfile(tarinfo) self.assert_(len(f.read()) == tarinfo.size, "size read does not match expected size") f.close() self.assert_(members == membercount, "could not find all members") def test_sparse(self): """Test sparse member extraction. """ if self.sep != "|": f1 = self.tar.extractfile("S-SPARSE") f2 = self.tar.extractfile("S-SPARSE-WITH-NULLS") self.assert_(f1.read() == f2.read(), "_FileObject failed on sparse file member") def test_readlines(self): """Test readlines() method of _FileObject. """ if self.sep != "|": filename = "0-REGTYPE-TEXT" self.tar.extract(filename, dirname()) lines1 = file(os.path.join(dirname(), filename), "rU").readlines() lines2 = self.tar.extractfile(filename).readlines() self.assert_(lines1 == lines2, "_FileObject.readline() does not work correctly") def test_iter(self): # Test iteration over ExFileObject. if self.sep != "|": filename = "0-REGTYPE-TEXT" self.tar.extract(filename, dirname()) lines1 = file(os.path.join(dirname(), filename), "rU").readlines() lines2 = [line for line in self.tar.extractfile(filename)] self.assert_(lines1 == lines2, "ExFileObject iteration does not work correctly") def test_seek(self): """Test seek() method of _FileObject, incl. random reading. """ if self.sep != "|": filename = "0-REGTYPE" self.tar.extract(filename, dirname()) data = file(os.path.join(dirname(), filename), "rb").read() tarinfo = self.tar.getmember(filename) fobj = self.tar.extractfile(tarinfo) text = fobj.read() fobj.seek(0) self.assert_(0 == fobj.tell(), "seek() to file's start failed") fobj.seek(2048, 0) self.assert_(2048 == fobj.tell(), "seek() to absolute position failed") fobj.seek(-1024, 1) self.assert_(1024 == fobj.tell(), "seek() to negative relative position failed") fobj.seek(1024, 1) self.assert_(2048 == fobj.tell(), "seek() to positive relative position failed") s = fobj.read(10) self.assert_(s == data[2048:2058], "read() after seek failed") fobj.seek(0, 2) self.assert_(tarinfo.size == fobj.tell(), "seek() to file's end failed") self.assert_(fobj.read() == "", "read() at file's end did not return empty string") fobj.seek(-tarinfo.size, 2) self.assert_(0 == fobj.tell(), "relative seek() to file's start failed") fobj.seek(512) s1 = fobj.readlines() fobj.seek(512) s2 = fobj.readlines() self.assert_(s1 == s2, "readlines() after seek failed") fobj.close() class ReadStreamTest(ReadTest): sep = "|" def test(self): """Test member extraction, and for StreamError when seeking backwards. """ ReadTest.test(self) tarinfo = self.tar.getmembers()[0] f = self.tar.extractfile(tarinfo) self.assertRaises(tarfile.StreamError, f.read) def test_stream(self): """Compare the normal tar and the stream tar. """ stream = self.tar tar = tarfile.open(tarname(), 'r') while 1: t1 = tar.next() t2 = stream.next() if t1 is None: break self.assert_(t2 is not None, "stream.next() failed.") if t2.islnk() or t2.issym(): self.assertRaises(tarfile.StreamError, stream.extractfile, t2) continue v1 = tar.extractfile(t1) v2 = stream.extractfile(t2) if v1 is None: continue self.assert_(v2 is not None, "stream.extractfile() failed") self.assert_(v1.read() == v2.read(), "stream extraction failed") stream.close() class ReadAsteriskTest(ReadTest): def setUp(self): mode = self.mode + self.sep + "*" self.tar = tarfile.open(tarname(self.comp), mode) class ReadStreamAsteriskTest(ReadStreamTest): def setUp(self): mode = self.mode + self.sep + "*" self.tar = tarfile.open(tarname(self.comp), mode) class WriteTest(BaseTest): mode = 'w' def setUp(self): mode = self.mode + self.sep + self.comp self.src = tarfile.open(tarname(self.comp), 'r') self.dstname = tmpname() self.dst = tarfile.open(self.dstname, mode) def tearDown(self): self.src.close() self.dst.close() def test_posix(self): self.dst.posix = 1 self._test() def test_nonposix(self): self.dst.posix = 0 self._test() def test_small(self): self.dst.add(os.path.join(os.path.dirname(__file__),"cfgparser.1")) self.dst.close() self.assertNotEqual(os.stat(self.dstname).st_size, 0) def _test(self): for tarinfo in self.src: if not tarinfo.isreg(): continue f = self.src.extractfile(tarinfo) if self.dst.posix and len(tarinfo.name) > tarfile.LENGTH_NAME: self.assertRaises(ValueError, self.dst.addfile, tarinfo, f) else: self.dst.addfile(tarinfo, f) class WriteStreamTest(WriteTest): sep = '|' class WriteGNULongTest(unittest.TestCase): """This testcase checks for correct creation of GNU Longname and Longlink extensions. It creates a tarfile and adds empty members with either long names, long linknames or both and compares the size of the tarfile with the expected size. It checks for SF bug #812325 in TarFile._create_gnulong(). While I was writing this testcase, I noticed a second bug in the same method: Long{names,links} weren't null-terminated which lead to bad tarfiles when their length was a multiple of 512. This is tested as well. """ def setUp(self): self.tar = tarfile.open(tmpname(), "w") self.tar.posix = False def tearDown(self): self.tar.close() def _length(self, s): blocks, remainder = divmod(len(s) + 1, 512) if remainder: blocks += 1 return blocks * 512 def _calc_size(self, name, link=None): # initial tar header count = 512 if len(name) > tarfile.LENGTH_NAME: # gnu longname extended header + longname count += 512 count += self._length(name) if link is not None and len(link) > tarfile.LENGTH_LINK: # gnu longlink extended header + longlink count += 512 count += self._length(link) return count def _test(self, name, link=None): tarinfo = tarfile.TarInfo(name) if link: tarinfo.linkname = link tarinfo.type = tarfile.LNKTYPE self.tar.addfile(tarinfo) v1 = self._calc_size(name, link) v2 = self.tar.offset self.assertEqual(v1, v2, "GNU longname/longlink creation failed") def test_longname_1023(self): self._test(("longnam/" * 127) + "longnam") def test_longname_1024(self): self._test(("longnam/" * 127) + "longname") def test_longname_1025(self): self._test(("longnam/" * 127) + "longname_") def test_longlink_1023(self): self._test("name", ("longlnk/" * 127) + "longlnk") def test_longlink_1024(self): self._test("name", ("longlnk/" * 127) + "longlink") def test_longlink_1025(self): self._test("name", ("longlnk/" * 127) + "longlink_") def test_longnamelink_1023(self): self._test(("longnam/" * 127) + "longnam", ("longlnk/" * 127) + "longlnk") def test_longnamelink_1024(self): self._test(("longnam/" * 127) + "longname", ("longlnk/" * 127) + "longlink") def test_longnamelink_1025(self): self._test(("longnam/" * 127) + "longname_", ("longlnk/" * 127) + "longlink_") class ExtractHardlinkTest(BaseTest): def test_hardlink(self): """Test hardlink extraction (bug #857297) """ # Prevent errors from being caught self.tar.errorlevel = 1 self.tar.extract("0-REGTYPE", dirname()) try: # Extract 1-LNKTYPE which is a hardlink to 0-REGTYPE self.tar.extract("1-LNKTYPE", dirname()) except EnvironmentError, e: import errno if e.errno == errno.ENOENT: self.fail("hardlink not extracted properly") # Gzip TestCases class ReadTestGzip(ReadTest): comp = "gz" class ReadStreamTestGzip(ReadStreamTest): comp = "gz" class WriteTestGzip(WriteTest): comp = "gz" class WriteStreamTestGzip(WriteStreamTest): comp = "gz" class ReadAsteriskTestGzip(ReadAsteriskTest): comp = "gz" class ReadStreamAsteriskTestGzip(ReadStreamAsteriskTest): comp = "gz" # Filemode test cases class FileModeTest(unittest.TestCase): def test_modes(self): self.assertEqual(tarfile.filemode(0755), '-rwxr-xr-x') self.assertEqual(tarfile.filemode(07111), '---s--s--t') if bz2: # Bzip2 TestCases class ReadTestBzip2(ReadTestGzip): comp = "bz2" class ReadStreamTestBzip2(ReadStreamTestGzip): comp = "bz2" class WriteTestBzip2(WriteTest): comp = "bz2" class WriteStreamTestBzip2(WriteStreamTestGzip): comp = "bz2" class ReadAsteriskTestBzip2(ReadAsteriskTest): comp = "bz2" class ReadStreamAsteriskTestBzip2(ReadStreamAsteriskTest): comp = "bz2" # If importing gzip failed, discard the Gzip TestCases. if not gzip: del ReadTestGzip del ReadStreamTestGzip del WriteTestGzip del WriteStreamTestGzip def test_main(): if gzip: # create testtar.tar.gz gzip.open(tarname("gz"), "wb").write(file(tarname(), "rb").read()) if bz2: # create testtar.tar.bz2 bz2.BZ2File(tarname("bz2"), "wb").write(file(tarname(), "rb").read()) tests = [ FileModeTest, ReadTest, ReadStreamTest, ReadAsteriskTest, ReadStreamAsteriskTest, WriteTest, WriteStreamTest, WriteGNULongTest, ] if hasattr(os, "link"): tests.append(ExtractHardlinkTest) if gzip: tests.extend([ ReadTestGzip, ReadStreamTestGzip, WriteTestGzip, WriteStreamTestGzip, ReadAsteriskTestGzip, ReadStreamAsteriskTestGzip ]) if bz2: tests.extend([ ReadTestBzip2, ReadStreamTestBzip2, WriteTestBzip2, WriteStreamTestBzip2, ReadAsteriskTestBzip2, ReadStreamAsteriskTestBzip2 ]) try: test_support.run_unittest(*tests) finally: if gzip: os.remove(tarname("gz")) if bz2: os.remove(tarname("bz2")) if os.path.exists(dirname()): shutil.rmtree(dirname()) if os.path.exists(tmpname()): os.remove(tmpname()) if __name__ == "__main__": test_main()