libcephfs/pybind: Add truncate, ftruncate

Fixes: https://tracker.ceph.com/issues/44171
Signed-off-by: Kotresh HR <khiremat@redhat.com>

diff --git a/src/pybind/cephfs/cephfs.pyx b/src/pybind/cephfs/cephfs.pyx
index 9ec9fffd6a07353cfa2fe57d42b6c09c316ba3b9..e3b9c5c97a9cd90c056aad8fdd1ec723ceadf879 100644 (file)
--- a/src/pybind/cephfs/cephfs.pyx
+++ b/src/pybind/cephfs/cephfs.pyx
@@ -1522,6 +1522,38 @@ cdef class LibCephFS(object):
             raise make_ex(ret, "error in write")
         return ret
 
+    def truncate(self, path, size):
+        """
+        Truncate the file to the given size.  If this operation causes the
+        file to expand, the empty bytes will be filled in with zeros.
+
+        :param path: the path to the file to truncate.
+        :param size: the new size of the file.
+        """
+
+        if not isinstance(size, int):
+            raise TypeError('size must be a int')
+
+        statx_dict = dict()
+        statx_dict["size"] = size
+        self.setattrx(path, statx_dict, CEPH_SETATTR_SIZE, AT_SYMLINK_NOFOLLOW)
+
+    def ftruncate(self, fd, size):
+        """
+        Truncate the file to the given size.  If this operation causes the
+        file to expand, the empty bytes will be filled in with zeros.
+
+        :param path: the path to the file to truncate.
+        :param size: the new size of the file.
+        """
+
+        if not isinstance(size, int):
+            raise TypeError('size must be a int')
+
+        statx_dict = dict()
+        statx_dict["size"] = size
+        self.fsetattrx(fd, statx_dict, CEPH_SETATTR_SIZE)
+
     def mknod(self, path, mode, rdev=0):
         """
         Make a block or character special file.

diff --git a/src/test/pybind/test_cephfs.py b/src/test/pybind/test_cephfs.py
index d1ec73b9df1aa7cfd5637288ee3573619bdf757a..c06647f8aa0c6acdcfbbc609afcf4ffe5584ccab 100644 (file)
--- a/src/test/pybind/test_cephfs.py
+++ b/src/test/pybind/test_cephfs.py
@@ -524,6 +524,29 @@ def test_fchown():
     cephfs.close(fd)
     cephfs.unlink(b'/file-fchown')
 
+@with_setup(setup_test)
+def test_truncate():
+    fd = cephfs.open(b'/file-truncate', 'w', 0o755)
+    cephfs.write(fd, b"1111", 0)
+    cephfs.truncate(b'/file-truncate', 0)
+    stat = cephfs.fsync(fd, 0)
+    st = cephfs.statx(b'/file-truncate', libcephfs.CEPH_STATX_SIZE, 0)
+    assert_equal(st["size"], 0)
+    cephfs.close(fd)
+    cephfs.unlink(b'/file-truncate')
+
+@with_setup(setup_test)
+def test_ftruncate():
+    fd = cephfs.open(b'/file-ftruncate', 'w', 0o755)
+    cephfs.write(fd, b"1111", 0)
+    assert_raises(TypeError, cephfs.ftruncate, b'/file-ftruncate', 0)
+    cephfs.ftruncate(fd, 0)
+    stat = cephfs.fsync(fd, 0)
+    st = cephfs.fstat(fd)
+    assert_equal(st.st_size, 0)
+    cephfs.close(fd)
+    cephfs.unlink(b'/file-ftruncate')
+
 @with_setup(setup_test)
 def test_fallocate():
     fd = cephfs.open(b'/file-fallocate', 'w', 0o755)