test_cephfs: Add python binding cephfs tests

Signed-off-by: Haomai Wang <haomaiwang@gmail.com>

diff --git a/src/pybind/cephfs.py b/src/pybind/cephfs.py
index d9967dfebbeb56172dac31638d6fe8265939fd2d..0b613d922772df6083f112155462b4bcc824846d 100644 (file)
--- a/src/pybind/cephfs.py
+++ b/src/pybind/cephfs.py
@@ -36,6 +36,10 @@ class NoSpace(Error):
     pass
 
 
+class InvalidValue(Error):
+    pass
+
+
 class IncompleteWriteError(Error):
     pass
 
@@ -61,7 +65,8 @@ def make_ex(ret, msg):
         errno.EIO       : IOError,
         errno.ENOSPC    : NoSpace,
         errno.EEXIST    : ObjectExists,
-        errno.ENODATA   : NoData
+        errno.ENODATA   : NoData,
+        errno.EINVAL    : InvalidValue,
         }
     ret = abs(ret)
     if ret in errors:

diff --git a/src/test/pybind/test_cephfs.py b/src/test/pybind/test_cephfs.py
new file mode 100644 (file)
index 0000000..5dc2171
--- /dev/null
+++ b/src/test/pybind/test_cephfs.py
@@ -0,0 +1,61 @@
+# vim: expandtab smarttab shiftwidth=4 softtabstop=4
+from nose.tools import assert_raises
+import cephfs
+
+cephfs = None
+
+def setup_module():
+    global cephfs
+    cephfs = cephfs.LibCephfs(conffile='')
+    cephfs.mount()
+
+def teardown_module():
+    global cephfs
+    cephfs.shutdown()
+
+def test_mount():
+    cephfs.mount()
+    cephfs.shutdown()
+    assert_raise(cephfs.LibCephFSStateError, cephfs.statfs)
+
+def test_version():
+    cephfs.version()
+
+def test_statfs():
+    stat = cephfs.statfs()
+    assert(len(stat) == 11)
+
+def test_syncfs():
+    stat = cephfs.syncfs()
+
+def test_directory():
+    cephfs.mkdir("/temp-directory", 0755)
+    cephfs.chdir("/temp-directory")
+    assert_equal(cephfs.getcwd() == "/temp-directory")
+    cephfs.rmdir("/temp-directory")
+    cephfs.chdir("/temp-directory")
+    assert_raise(cephfs.ObjectNotFound, cephfs.chdir("/temp-directory"))
+
+def test_walk_dir():
+    dirs = ["/dir-1", "/dir-2", "dir-3"]
+    for i in dirs:
+        cephfs.mkdir(i, 0755)
+    handler = cephfs.opendir("/")
+    d = cephfs.readdir(handler)
+    while d:
+        assert(d['d_name'] in dirs)
+        dirs.remove(d['d_name'])
+        d = cephfs.readdir(handler)
+    assert(len(dirs) == 0)
+    cephfs.closedir(handler)
+
+def test_xattr():
+    assert_raise(InvalidValue, cephfs.setxattr, "/", "key", "value", 0)
+    cephfs.setxattr("/", "user.key", "value", 0)
+    assert_equal("value", cephfs.getxattr("/", "user.key"))
+
+def test_rename():
+    cephfs.mkdir("/a", 0755)
+    cephfs.mkdir("/a/b", 0755)
+    cephfs.rename("/a", "/b")
+    cephfs.stat("/a/b")