home *** CD-ROM | disk | FTP | other *** search
/ PC World 2005 June / PCWorld_2005-06_cd.bin / software / vyzkuste / firewally / firewally.exe / framework-2.3.exe / test_tarfile.py < prev    next >
Text File  |  2003-12-30  |  8KB  |  276 lines
  1. import sys
  2. import os
  3. import shutil
  4. import tempfile
  5.  
  6. import unittest
  7. import tarfile
  8.  
  9. from test import test_support
  10.  
  11. # Check for our compression modules.
  12. try:
  13.     import gzip
  14.     gzip.GzipFile
  15. except (ImportError, AttributeError):
  16.     gzip = None
  17. try:
  18.     import bz2
  19. except ImportError:
  20.     bz2 = None
  21.  
  22. def path(path):
  23.     return test_support.findfile(path)
  24.  
  25. testtar = path("testtar.tar")
  26. tempdir = os.path.join(tempfile.gettempdir(), "testtar" + os.extsep + "dir")
  27. tempname = test_support.TESTFN
  28. membercount = 10
  29.  
  30. def tarname(comp=""):
  31.     if not comp:
  32.         return testtar
  33.     return os.path.join(tempdir, "%s%s%s" % (testtar, os.extsep, comp))
  34.  
  35. def dirname():
  36.     if not os.path.exists(tempdir):
  37.         os.mkdir(tempdir)
  38.     return tempdir
  39.  
  40. def tmpname():
  41.     return tempname
  42.  
  43.  
  44. class BaseTest(unittest.TestCase):
  45.     comp = ''
  46.     mode = 'r'
  47.     sep = ':'
  48.  
  49.     def setUp(self):
  50.         mode = self.mode + self.sep + self.comp
  51.         self.tar = tarfile.open(tarname(self.comp), mode)
  52.  
  53.     def tearDown(self):
  54.         self.tar.close()
  55.  
  56. class ReadTest(BaseTest):
  57.  
  58.     def test(self):
  59.         """Test member extraction.
  60.         """
  61.         members = 0
  62.         for tarinfo in self.tar:
  63.             members += 1
  64.             if not tarinfo.isreg():
  65.                 continue
  66.             f = self.tar.extractfile(tarinfo)
  67.             self.assert_(len(f.read()) == tarinfo.size,
  68.                          "size read does not match expected size")
  69.             f.close()
  70.  
  71.         self.assert_(members == membercount,
  72.                      "could not find all members")
  73.  
  74.     def test_sparse(self):
  75.         """Test sparse member extraction.
  76.         """
  77.         if self.sep != "|":
  78.             f1 = self.tar.extractfile("S-SPARSE")
  79.             f2 = self.tar.extractfile("S-SPARSE-WITH-NULLS")
  80.             self.assert_(f1.read() == f2.read(),
  81.                          "_FileObject failed on sparse file member")
  82.  
  83.     def test_readlines(self):
  84.         """Test readlines() method of _FileObject.
  85.         """
  86.         if self.sep != "|":
  87.             filename = "0-REGTYPE-TEXT"
  88.             self.tar.extract(filename, dirname())
  89.             lines1 = file(os.path.join(dirname(), filename), "rU").readlines()
  90.             lines2 = self.tar.extractfile(filename).readlines()
  91.             self.assert_(lines1 == lines2,
  92.                          "_FileObject.readline() does not work correctly")
  93.  
  94.     def test_seek(self):
  95.         """Test seek() method of _FileObject, incl. random reading.
  96.         """
  97.         if self.sep != "|":
  98.             filename = "0-REGTYPE"
  99.             self.tar.extract(filename, dirname())
  100.             data = file(os.path.join(dirname(), filename), "rb").read()
  101.  
  102.             tarinfo = self.tar.getmember(filename)
  103.             fobj = self.tar.extractfile(tarinfo)
  104.  
  105.             text = fobj.read()
  106.             fobj.seek(0)
  107.             self.assert_(0 == fobj.tell(),
  108.                          "seek() to file's start failed")
  109.             fobj.seek(2048, 0)
  110.             self.assert_(2048 == fobj.tell(),
  111.                          "seek() to absolute position failed")
  112.             fobj.seek(-1024, 1)
  113.             self.assert_(1024 == fobj.tell(),
  114.                          "seek() to negative relative position failed")
  115.             fobj.seek(1024, 1)
  116.             self.assert_(2048 == fobj.tell(),
  117.                          "seek() to positive relative position failed")
  118.             s = fobj.read(10)
  119.             self.assert_(s == data[2048:2058],
  120.                          "read() after seek failed")
  121.             fobj.seek(0, 2)
  122.             self.assert_(tarinfo.size == fobj.tell(),
  123.                          "seek() to file's end failed")
  124.             self.assert_(fobj.read() == "",
  125.                          "read() at file's end did not return empty string")
  126.             fobj.seek(-tarinfo.size, 2)
  127.             self.assert_(0 == fobj.tell(),
  128.                          "relative seek() to file's start failed")
  129.             fobj.seek(512)
  130.             s1 = fobj.readlines()
  131.             fobj.seek(512)
  132.             s2 = fobj.readlines()
  133.             self.assert_(s1 == s2,
  134.                          "readlines() after seek failed")
  135.             fobj.close()
  136.  
  137. class ReadStreamTest(ReadTest):
  138.     sep = "|"
  139.  
  140.     def test(self):
  141.         """Test member extraction, and for StreamError when
  142.            seeking backwards.
  143.         """
  144.         ReadTest.test(self)
  145.         tarinfo = self.tar.getmembers()[0]
  146.         f = self.tar.extractfile(tarinfo)
  147.         self.assertRaises(tarfile.StreamError, f.read)
  148.  
  149.     def test_stream(self):
  150.         """Compare the normal tar and the stream tar.
  151.         """
  152.         stream = self.tar
  153.         tar = tarfile.open(tarname(), 'r')
  154.  
  155.         while 1:
  156.             t1 = tar.next()
  157.             t2 = stream.next()
  158.             if t1 is None:
  159.                 break
  160.             self.assert_(t2 is not None, "stream.next() failed.")
  161.  
  162.             if t2.islnk() or t2.issym():
  163.                 self.assertRaises(tarfile.StreamError, stream.extractfile, t2)
  164.                 continue
  165.             v1 = tar.extractfile(t1)
  166.             v2 = stream.extractfile(t2)
  167.             if v1 is None:
  168.                 continue
  169.             self.assert_(v2 is not None, "stream.extractfile() failed")
  170.             self.assert_(v1.read() == v2.read(), "stream extraction failed")
  171.  
  172.         stream.close()
  173.  
  174. class WriteTest(BaseTest):
  175.     mode = 'w'
  176.  
  177.     def setUp(self):
  178.         mode = self.mode + self.sep + self.comp
  179.         self.src = tarfile.open(tarname(self.comp), 'r')
  180.         self.dst = tarfile.open(tmpname(), mode)
  181.  
  182.     def tearDown(self):
  183.         self.src.close()
  184.         self.dst.close()
  185.  
  186.     def test_posix(self):
  187.         self.dst.posix = 1
  188.         self._test()
  189.  
  190.     def test_nonposix(self):
  191.         self.dst.posix = 0
  192.         self._test()
  193.  
  194.     def _test(self):
  195.         for tarinfo in self.src:
  196.             if not tarinfo.isreg():
  197.                 continue
  198.             f = self.src.extractfile(tarinfo)
  199.             if self.dst.posix and len(tarinfo.name) > tarfile.LENGTH_NAME:
  200.                 self.assertRaises(ValueError, self.dst.addfile,
  201.                                  tarinfo, f)
  202.             else:
  203.                 self.dst.addfile(tarinfo, f)
  204.  
  205. class WriteStreamTest(WriteTest):
  206.     sep = '|'
  207.  
  208. # Gzip TestCases
  209. class ReadTestGzip(ReadTest):
  210.     comp = "gz"
  211. class ReadStreamTestGzip(ReadStreamTest):
  212.     comp = "gz"
  213. class WriteTestGzip(WriteTest):
  214.     comp = "gz"
  215. class WriteStreamTestGzip(WriteStreamTest):
  216.     comp = "gz"
  217.  
  218. if bz2:
  219.     # Bzip2 TestCases
  220.     class ReadTestBzip2(ReadTestGzip):
  221.         comp = "bz2"
  222.     class ReadStreamTestBzip2(ReadStreamTestGzip):
  223.         comp = "bz2"
  224.     class WriteTestBzip2(WriteTest):
  225.         comp = "bz2"
  226.     class WriteStreamTestBzip2(WriteStreamTestGzip):
  227.         comp = "bz2"
  228.  
  229. # If importing gzip failed, discard the Gzip TestCases.
  230. if not gzip:
  231.     del ReadTestGzip
  232.     del ReadStreamTestGzip
  233.     del WriteTestGzip
  234.     del WriteStreamTestGzip
  235.  
  236. def test_main():
  237.     if gzip:
  238.         # create testtar.tar.gz
  239.         gzip.open(tarname("gz"), "wb").write(file(tarname(), "rb").read())
  240.     if bz2:
  241.         # create testtar.tar.bz2
  242.         bz2.BZ2File(tarname("bz2"), "wb").write(file(tarname(), "rb").read())
  243.  
  244.     tests = [
  245.         ReadTest,
  246.         ReadStreamTest,
  247.         WriteTest,
  248.         WriteStreamTest
  249.     ]
  250.  
  251.     if gzip:
  252.         tests.extend([
  253.             ReadTestGzip, ReadStreamTestGzip,
  254.             WriteTestGzip, WriteStreamTestGzip
  255.         ])
  256.  
  257.     if bz2:
  258.         tests.extend([
  259.             ReadTestBzip2, ReadStreamTestBzip2,
  260.             WriteTestBzip2, WriteStreamTestBzip2
  261.         ])
  262.     try:
  263.         test_support.run_unittest(*tests)
  264.     finally:
  265.         if gzip:
  266.             os.remove(tarname("gz"))
  267.         if bz2:
  268.             os.remove(tarname("bz2"))
  269.         if os.path.exists(dirname()):
  270.             shutil.rmtree(dirname())
  271.         if os.path.exists(tmpname()):
  272.             os.remove(tmpname())
  273.  
  274. if __name__ == "__main__":
  275.     test_main()
  276.