Merge branch 'zip-symlinks3'

avocado/utils/archive.py

@@ -11,16 +11,21 @@
 #
 # Copyright: Red Hat Inc. 2014
 # Author: Ruda Moura <rmoura@redhat.com>
-
-
 """
 Module to help extract and create compressed archives.
 """
 
+import logging
 import os
+import platform
+import stat
 import tarfile
 import zipfile
 
+
+LOG = logging.getLogger(__name__)
+
+
 try:
     import lzma
     LZMA_CAPABLE = True
@@ -159,6 +164,30 @@ class ArchiveFile(object):
         :param path: destination path.
         """
         self._engine.extractall(path)
+        if self.is_zip:
+            self._update_zip_extra_attrs(path)
+
+    def _update_zip_extra_attrs(self, dst_dir):
+        if platform.system() != "Linux":
+            LOG.warn("Attr handling in zip files only supported on Linux.")
+            return
+        # Walk all files and re-create files as symlinks
+        for path, info in self._engine.NameToInfo.iteritems():
+            dst = os.path.join(dst_dir, path)
+            if not os.path.exists(dst):
+                LOG.warn("Paths in this zip file are stored in unsupported "
+                         "format, not updating the attributes. (%s)", msg)
+                return
+            attr = info.external_attr >> 16
+            if attr & stat.S_IFLNK == stat.S_IFLNK:
+                dst = os.path.join(dst_dir, path)
+                src = open(dst, 'r').read()
+                os.remove(dst)
+                os.symlink(src, dst)
+                continue    # Don't override any other attributes on links
+            mode = attr & 511   # Mask only permissions
+            if mode and mode != 436:  # If mode is stored and is not default
+                os.chmod(dst, mode)
 
     def close(self):
         """

selftests/unit/test_archive.py

@@ -93,8 +93,50 @@ class ArchiveTest(unittest.TestCase):
     def test_tbz2_2_file(self):
         self.compress_and_check_file('.tbz2')
 
+    def test_zip_extra_attrs(self):
+        """
+        Check that utils.archive reflects extra attrs of file like symlinks
+        and file permissions.
+        """
+        def get_path(*args):
+            """ Get path with decompressdir prefix """
+            return os.path.join(self.decompressdir, *args)
+        # File types
+        zip_path = os.path.abspath(os.path.join(os.path.dirname(__file__),
+                                                os.path.pardir, ".data",
+                                                "test_archive__symlinks.zip"))
+        archive.uncompress(zip_path, self.decompressdir)
+        self.assertTrue(os.path.islink(get_path("link_to_dir")))
+        self.assertTrue(os.path.islink(get_path("link_to_file")))
+        self.assertTrue(os.path.islink(get_path("link_to_file2")))
+        self.assertTrue(os.path.islink(get_path("dir", "2nd_link_to_file")))
+        self.assertTrue(os.path.islink(get_path("dir",
+                                                "link_to_link_to_file2")))
+        self.assertTrue(os.path.islink(get_path("dir", "2nd_link_to_file")))
+        self.assertTrue(os.path.islink(get_path("link_to_dir",
+                                                "2nd_link_to_file")))
+        self.assertTrue(os.path.isfile(get_path("file")))
+        self.assertTrue(os.path.isfile(get_path("dir", "file2")))
+        self.assertTrue(os.path.isfile(get_path("link_to_dir", "file2")))
+        act = os.path.realpath(get_path("link_to_dir",
+                                        "link_to_link_to_file2"))
+        exp = get_path("dir", "file2")
+        self.assertEqual(act, exp)
+        self.assertEqual(os.path.realpath(get_path("link_to_dir")),
+                         get_path("dir"))
+        # File permissions
+        self.assertEqual(os.stat(get_path("dir", "file2")).st_mode & 0o777,
+                         0o664)
+        self.assertEqual(os.stat(get_path("file")).st_mode & 0o777, 0o753)
+        self.assertEqual(os.stat(get_path("dir")).st_mode & 0o777, 0o775)
+        self.assertEqual(os.stat(get_path("link_to_file2")).st_mode & 0o777,
+                         0o664)
+        self.assertEqual(os.stat(get_path("link_to_dir")).st_mode & 0o777,
+                         0o775)
+        self.assertEqual(os.stat(get_path("link_to_file")).st_mode & 0o777,
+                         0o753)
+
     def tearDown(self):
-        pass
         try:
             shutil.rmtree(self.basedir)
         except OSError: