Split VFS.lchmod into chmod_symlink and chmod_not_symlink

This makes the code overall clearer.

4 files changed, 58 insertions, 24 deletions

diff --git a/obnamlib/metadata.py b/obnamlib/metadata.py
index 9bb2d0d9..c055087f 100644
--- a/obnamlib/metadata.py
+++ b/obnamlib/metadata.py
@@ -25,8 +25,6 @@ import tracing
 
 import obnamlib
 
-# lchmod is not available on Linux:
-LCHMOD_AVAILABLE = getattr(os, "lchmod", None) is not None
 
 metadata_verify_fields = (
     'st_mode', 'st_mtime_sec', 'st_mtime_nsec',
@@ -249,13 +247,15 @@ def set_metadata(fs, filename, metadata, getuid=None):
     if getuid() == 0:
         fs.lchown(filename, metadata.st_uid, metadata.st_gid)
 
-    if LCHMOD_AVAILABLE or not symlink:
-        # If we are not the owner, and not root, do not restore setuid/setgid.
-        mode = metadata.st_mode
-        if getuid() not in (0, metadata.st_uid): # pragma: no cover
-            mode = mode & (~stat.S_ISUID)
-            mode = mode & (~stat.S_ISGID)
-        fs.lchmod(filename, mode)
+    # If we are not the owner, and not root, do not restore setuid/setgid.
+    mode = metadata.st_mode
+    if getuid() not in (0, metadata.st_uid): # pragma: no cover
+        mode = mode & (~stat.S_ISUID)
+        mode = mode & (~stat.S_ISGID)
+    if symlink:
+        fs.chmod_symlink(filename, mode)
+    else:
+        fs.chmod_not_symlink(filename, mode)
 
     if metadata.xattr: # pragma: no cover
         set_xattrs_from_blob(fs, filename, metadata.xattr)
diff --git a/obnamlib/plugins/sftp_plugin.py b/obnamlib/plugins/sftp_plugin.py
index f038771e..576ba7e8 100644
--- a/obnamlib/plugins/sftp_plugin.py
+++ b/obnamlib/plugins/sftp_plugin.py
@@ -476,7 +476,14 @@ class SftpFS(obnamlib.VirtualFileSystem):
             self.sftp.chown(pathname, uid, gid)
 
     @ioerror_to_oserror
-    def lchmod(self, pathname, mode):
+    def chmod_symlink(self, pathname, mode):
+        # SFTP and/or paramiko don't have lchmod at all, so we can't
+        # actually do this. However, we at least check that pathname
+        # exists.
+        self.lstat(pathname)
+
+    @ioerror_to_oserror
+    def chmod_not_symlink(self, pathname, mode):
         self._delay()
         self.sftp.chmod(pathname, mode)
 
diff --git a/obnamlib/vfs.py b/obnamlib/vfs.py
index 207d4c35..680ca65e 100644
--- a/obnamlib/vfs.py
+++ b/obnamlib/vfs.py
@@ -160,8 +160,23 @@ class VirtualFileSystem(object):
     def lchown(self, pathname, uid, gid):
         '''Like os.lchown.'''
 
-    def lchmod(self, pathname, mode):
-        '''Like os.lchmod.'''
+    def chmod_symlink(self, pathname, mode):
+        '''Like os.lchmod, for symlinks only.
+
+        This may fail if the pathname is not a symlink (but it may
+        not).  If the target is a symlink, but the platform (e.g.,
+        Linux) does not allow setting the permissions of a symlink,
+        the method will silently do nothing.
+
+        '''
+
+    def chmod_not_symlink(self, pathname, mode):
+        '''Like os.chmod, for non-symlinks only.
+
+        This may fail if pathname is a symlink (but it may not). It
+        MUST NOT be called for a symlink; use chmod_symlink instead.
+
+        '''
 
     def lutimes(self, pathname, atime_sec, atime_nsec, mtime_sec, mtime_nsec):
         '''Like lutimes(2).
@@ -523,13 +538,16 @@ class VfsTests(object): # pragma: no cover
     def test_lstat_raises_oserror_for_nonexistent_entry(self):
         self.assertRaises(OSError, self.fs.lstat, 'notexists')
 
-    def test_lchmod_sets_permissions_correctly(self):
+    def test_chmod_not_symlink_sets_permissions_correctly(self):
         self.fs.mkdir('foo')
-        self.fs.lchmod('foo', 0777)
+        self.fs.chmod_not_symlink('foo', 0777)
         self.assertEqual(self.fs.lstat('foo').st_mode & 0777, 0777)
 
-    def test_lchmod_raises_oserror_for_nonexistent_entry(self):
-        self.assertRaises(OSError, self.fs.lchmod, 'notexists', 0)
+    def test_chmod_not_symlink_raises_oserror_for_nonexistent_entry(self):
+        self.assertRaises(OSError, self.fs.chmod_not_symlink, 'notexists', 0)
+
+    def test_chmod_symlink_raises_oserror_for_nonexistent_entry(self):
+        self.assertRaises(OSError, self.fs.chmod_symlink, 'notexists', 0)
 
     def test_lutimes_sets_times_correctly(self):
         self.fs.mkdir('foo')
diff --git a/obnamlib/vfs_local.py b/obnamlib/vfs_local.py
index 20b0dc7d..70a550f8 100644
--- a/obnamlib/vfs_local.py
+++ b/obnamlib/vfs_local.py
@@ -32,11 +32,6 @@ import obnamlib
 # O_NOATIME is Linux specific:
 EXTRA_OPEN_FLAGS = getattr(os, "O_NOATIME", 0)
 
-# On Linux, lchmod() is not available. We make sure not to call it
-# with symlinks (see obnamlib.metadata.set_metadata) so falling back
-# to os.chmod is fine.
-lchmod = getattr(os, "lchmod", os.chmod)
-
 
 class LocalFSFile(file):
 
@@ -76,6 +71,9 @@ class LocalFS(obnamlib.VirtualFileSystem):
         self.crash_limit = None
         self.crash_counter = 0
 
+        # Do we have lchmod?
+        self.got_lchmod = hasattr(os, 'lchmod')
+
     def maybe_crash(self): # pragma: no cover
         if self.crash_limit is not None:
             self.crash_counter += 1
@@ -200,9 +198,20 @@ class LocalFS(obnamlib.VirtualFileSystem):
         tracing.trace('lchown %s %d %d', pathname, uid, gid)
         os.lchown(self.join(pathname), uid, gid)
 
-    def lchmod(self, pathname, mode):
-        tracing.trace('lchmod %s %o', pathname, mode)
-        lchmod(self.join(pathname), mode)
+    # This method is excluded from test coverage because the platform
+    # either has lchmod or doesn't, and accordingly either branch of
+    # the if statement is taken, and the other branch shows up as not
+    # being tested by the unit tests.
+    def chmod_symlink(self, pathname, mode): # pragma: no cover
+        tracing.trace('chmod_symlink %s %o', pathname, mode)
+        if self.got_lchmod:
+            os.lchmod(self.join(pathname), mode)
+        else:
+            self.lstat(pathname)
+
+    def chmod_not_symlink(self, pathname, mode):
+        tracing.trace('chmod_not_symlink %s %o', pathname, mode)
+        os.chmod(self.join(pathname), mode)
 
     def lutimes(self, pathname, atime_sec, atime_nsec, mtime_sec, mtime_nsec):
         assert atime_sec is not None