Merge pull request #12384 from renhwztetecs/renhw-wip-libcephfs-addre…

…adlink

libcephfs: add readlink function in cephfs.pyx

Reviewed-by: John Spray <john.spray@redhat.com>

src/pybind/cephfs/cephfs.pyx

@@ -124,6 +124,7 @@ cdef extern from "cephfs/libcephfs.h" nogil:
     int ceph_link(ceph_mount_info *cmount, const char *existing, const char *newname)
     int ceph_unlink(ceph_mount_info *cmount, const char *path)
     int ceph_symlink(ceph_mount_info *cmount, const char *existing, const char *newname)
+    int ceph_readlink(ceph_mount_info *cmount, const char *path, char *buf, int64_t size)
     int ceph_setxattr(ceph_mount_info *cmount, const char *path, const char *name,
                       const void *value, size_t size, int flags)
     int ceph_getxattr(ceph_mount_info *cmount, const char *path, const char *name,
@@ -863,6 +864,25 @@ cdef class LibCephFS(object):
             ret = ceph_link(self.cluster, _existing, _newname)
         if ret < 0:
             raise make_ex(ret, "error in link")    
+
+    def readlink(self, path, size):
+        self.require_state("mounted")
+        path = cstr(path, 'path')
+
+        cdef:
+            char* _path = path
+            int64_t _size = size
+            char *buf = NULL
+
+        try:
+            buf = <char *>realloc_chk(buf, _size)
+            with nogil:
+                ret = ceph_readlink(self.cluster, _path, buf, _size)
+            if ret < 0:
+                raise make_ex(ret, "error in readlink")
+            return buf
+        finally:
+            free(buf)
 
     def unlink(self, path):
         self.require_state("mounted")

src/test/pybind/test_cephfs.py

@@ -181,6 +181,17 @@ def test_symlink():
     cephfs.close(fd)
     cephfs.unlink(b'file-2')
 
+@with_setup(setup_test)
+def test_readlink():
+    fd = cephfs.open(b'/file-1', 'w', 0o755)
+    cephfs.write(fd, b"1111", 0)
+    cephfs.close(fd)
+    cephfs.symlink(b'/file-1', b'/file-2')
+    d = cephfs.readlink(b"/file-2",100)
+    assert_equal(d, b"/file-1")
+    cephfs.unlink(b'/file-2')
+    cephfs.unlink(b'/file-1')
+
 @with_setup(setup_test)
 def test_delete_cwd():
     assert_equal(b"/", cephfs.getcwd())