pybind/cephfs: Add lchmod python binding

Fixes: https://tracker.ceph.com/issues/49882
Signed-off-by: Kotresh HR <khiremat@redhat.com>
(cherry picked from commit b2375adce085e98bef521422441b80a945e38c80)

Conflicts:
  src/pybind/cephfs/mock_cephfs.pxi : Not present in octopus
  src/pybind/cephfs/c_cephfs.pxd : Not present in octopus
  src/pybind/cephfs/cephfs.pyx : Few of the fops is not part of octopus
      which got pulled as part of this backport
  src/test/pybind/test_cephfs.py :  Few of the fops is not part of
      octopus, which got pulled as part of this backport. Added missing
      stat import.

diff --git a/src/pybind/cephfs/cephfs.pyx b/src/pybind/cephfs/cephfs.pyx
index 9fcf187ea82ff6f267ee7f5883a09c4dbc60801e..e19938aeea970bce9a90f7e3a5719a5776cedef9 100644 (file)
--- a/src/pybind/cephfs/cephfs.pyx
+++ b/src/pybind/cephfs/cephfs.pyx
@@ -185,6 +185,7 @@ cdef extern from "cephfs/libcephfs.h" nogil:
     int ceph_fsync(ceph_mount_info *cmount, int fd, int syncdataonly)
     int ceph_conf_parse_argv(ceph_mount_info *cmount, int argc, const char **argv)
     int ceph_chmod(ceph_mount_info *cmount, const char *path, mode_t mode)
+    int ceph_lchmod(ceph_mount_info *cmount, const char *path, mode_t mode)
     int ceph_chown(ceph_mount_info *cmount, const char *path, int uid, int gid)
     int ceph_lchown(ceph_mount_info *cmount, const char *path, int uid, int gid)
     int64_t ceph_lseek(ceph_mount_info *cmount, int fd, int64_t offset, int whence)
@@ -904,6 +905,26 @@ cdef class LibCephFS(object):
         if ret < 0:
             raise make_ex(ret, "error in chmod {}".format(path.decode('utf-8')))
 
+    def lchmod(self, path, mode) -> None:
+        """
+        Change file mode. If the path is a symbolic link, it won't be dereferenced.
+
+        :param path: the path of the file. This must be either an absolute path or
+                     a relative path off of the current working directory.
+        :param mode: the permissions to be set .
+        """
+        self.require_state("mounted")
+        path = cstr(path, 'path')
+        if not isinstance(mode, int):
+            raise TypeError('mode must be an int')
+        cdef:
+            char* _path = path
+            int _mode = mode
+        with nogil:
+            ret = ceph_lchmod(self.cluster, _path, _mode)
+        if ret < 0:
+            raise make_ex(ret, "error in chmod {}".format(path.decode('utf-8')))
+
     def chown(self, path, uid, gid, follow_symlink=True):
         """
         Change directory ownership

diff --git a/src/test/pybind/test_cephfs.py b/src/test/pybind/test_cephfs.py
index 93aa623e97f9cb38336efcc75fac780dd3e32c65..d234623c1522f9fd0ee31c27cabfbaf57e7f262b 100644 (file)
--- a/src/test/pybind/test_cephfs.py
+++ b/src/test/pybind/test_cephfs.py
@@ -1,8 +1,9 @@
 # vim: expandtab smarttab shiftwidth=4 softtabstop=4
-from nose.tools import assert_raises, assert_equal, assert_greater, with_setup
+from nose.tools import assert_raises, assert_equal, assert_not_equal, assert_greater, with_setup
 import cephfs as libcephfs
 import fcntl
 import os
+import stat
 import time
 from datetime import datetime
 
@@ -424,6 +425,31 @@ def test_futimens():
     cephfs.close(fd)
     cephfs.unlink(b'/file-1')
 
+@with_setup(setup_test)
+def test_lchmod():
+    fd = cephfs.open(b'/file-1', 'w', 0o755)
+    cephfs.write(fd, b'0000', 0)
+    cephfs.close(fd)
+
+    cephfs.symlink(b'/file-1', b'/file-2')
+
+    stx_pre_t = cephfs.statx(b'/file-1', libcephfs.CEPH_STATX_MODE, 0)
+    stx_pre_s = cephfs.statx(b'/file-2', libcephfs.CEPH_STATX_MODE, libcephfs.AT_SYMLINK_NOFOLLOW)
+
+    time.sleep(1)
+    cephfs.lchmod(b'/file-2', 0o400)
+
+    stx_post_t = cephfs.statx(b'/file-1', libcephfs.CEPH_STATX_MODE, 0)
+    stx_post_s = cephfs.statx(b'/file-2', libcephfs.CEPH_STATX_MODE, libcephfs.AT_SYMLINK_NOFOLLOW)
+
+    assert_equal(stx_post_t['mode'], stx_pre_t['mode'])
+    assert_not_equal(stx_post_s['mode'], stx_pre_s['mode'])
+    stx_post_s_perm_bits = stx_post_s['mode'] & ~stat.S_IFMT(stx_post_s["mode"])
+    assert_equal(stx_post_s_perm_bits, 0o400)
+
+    cephfs.unlink(b'/file-2')
+    cephfs.unlink(b'/file-1')
+
 @with_setup(setup_test)
 def test_disk_quota_exceeeded_error():
     cephfs.mkdir("/dir-1", 0o755)