RepoMirrors/ceph
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2024-12-17 17:05:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

cephfs.pyx: implement python bindings for fstat

Browse Source 
Signed-off-by: Ramana Raja <rraja@redhat.com>
This commit is contained in:
Ramana Raja 2016-06-23 17:39:32 +05:30
parent 7f7d2a76ae
commit f58403f3d1
2 changed files with 31 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
src/pybind/cephfs/cephfs.pyx
View File

@ -108,6 +108,7 @@ cdef extern from "cephfs/libcephfs.h" nogil:
int ceph_conf_set(ceph_mount_info *cmount, const char *option, const char *value)
int ceph_mount(ceph_mount_info *cmount, const char *root)
int ceph_fstat(ceph_mount_info *cmount, int fd, stat *stbuf)
int ceph_stat(ceph_mount_info *cmount, const char *path, stat *stbuf)
int ceph_statfs(ceph_mount_info *cmount, const char *path, statvfs *stbuf)
@ -800,6 +801,29 @@ cdef class LibCephFS(object):
st_mtime=datetime.fromtimestamp(statbuf.st_mtime),
st_ctime=datetime.fromtimestamp(statbuf.st_ctime))
def fstat(self, fd):
self.require_state("mounted")
if not isinstance(fd, int):
raise TypeError('fd must be an int')
cdef:
int _fd = fd
stat statbuf
with nogil:
ret = ceph_fstat(self.cluster, _fd, &statbuf)
if ret < 0:
raise make_ex(ret, "error in fsat")
return StatResult(st_dev=statbuf.st_dev, st_ino=statbuf.st_ino,
st_mode=statbuf.st_mode, st_nlink=statbuf.st_nlink,
st_uid=statbuf.st_uid, st_gid=statbuf.st_gid,
st_rdev=statbuf.st_rdev, st_size=statbuf.st_size,
st_blksize=statbuf.st_blksize,
st_blocks=statbuf.st_blocks,
st_atime=datetime.fromtimestamp(statbuf.st_atime),
st_mtime=datetime.fromtimestamp(statbuf.st_mtime),
st_ctime=datetime.fromtimestamp(statbuf.st_ctime))
def symlink(self, existing, newname):
self.require_state("mounted")
existing = cstr(existing, 'existing')

7
src/test/pybind/test_cephfs.py
View File

@ -40,6 +40,13 @@ def test_conf_get():
def test_version():
cephfs.version()
@with_setup(setup_test)
def test_fstat():
fd = cephfs.open('file-1', 'w', 0755)
stat = cephfs.fstat(fd)
assert(len(stat) == 13)
cephfs.close(fd)
@with_setup(setup_test)
def test_statfs():
stat = cephfs.statfs('/')