Adding upstream version 14.2.21.upstream/14.2.21upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

15 files changed, 6275 insertions, 0 deletions

diff --git a/src/test/libcephfs/CMakeLists.txt b/src/test/libcephfs/CMakeLists.txt
new file mode 100644
index 00000000..4ede0eca
--- /dev/null
+++ b/src/test/libcephfs/CMakeLists.txt
@@ -0,0 +1,64 @@
+if(${WITH_CEPHFS})
+  add_executable(ceph_test_libcephfs
+    test.cc
+    readdir_r_cb.cc
+    caps.cc
+    multiclient.cc
+    flock.cc
+    recordlock.cc
+    acl.cc
+    main.cc
+    deleg.cc
+    monconfig.cc
+  )
+  target_link_libraries(ceph_test_libcephfs
+    ceph-common
+    cephfs
+    ${UNITTEST_LIBS}
+    ${EXTRALIBS}
+    ${CMAKE_DL_LIBS}
+    )
+  install(TARGETS ceph_test_libcephfs
+    DESTINATION ${CMAKE_INSTALL_BINDIR})
+
+  add_executable(ceph_test_libcephfs_reclaim
+    reclaim.cc
+  )
+  target_link_libraries(ceph_test_libcephfs_reclaim
+    cephfs
+    ${UNITTEST_LIBS}
+    ${EXTRALIBS}
+    ${CMAKE_DL_LIBS}
+    )
+  install(TARGETS ceph_test_libcephfs_reclaim
+    DESTINATION ${CMAKE_INSTALL_BINDIR})
+
+  add_executable(ceph_test_libcephfs_lazyio
+    lazyio.cc
+  )
+  target_link_libraries(ceph_test_libcephfs_lazyio
+    cephfs
+    librados
+    ${UNITTEST_LIBS}
+    ${EXTRALIBS}
+    ${CMAKE_DL_LIBS}
+    )   
+  install(TARGETS ceph_test_libcephfs_lazyio
+    DESTINATION ${CMAKE_INSTALL_BINDIR})
+
+  add_executable(ceph_test_libcephfs_access
+    test.cc
+    access.cc
+  )
+  target_link_libraries(ceph_test_libcephfs_access
+    ceph-common
+    cephfs
+    librados
+    ${UNITTEST_LIBS}
+    ${EXTRALIBS}
+    ${CMAKE_DL_LIBS}
+    )
+  install(TARGETS ceph_test_libcephfs_access
+    DESTINATION ${CMAKE_INSTALL_BINDIR})
+endif(${WITH_CEPHFS})  
+
diff --git a/src/test/libcephfs/access.cc b/src/test/libcephfs/access.cc
new file mode 100644
index 00000000..cac42f33
--- /dev/null
+++ b/src/test/libcephfs/access.cc
@@ -0,0 +1,395 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 New Dream Network
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "gtest/gtest.h"
+#include "common/ceph_argparse.h"
+#include "include/buffer.h"
+#include "include/stringify.h"
+#include "include/cephfs/libcephfs.h"
+#include "include/rados/librados.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <dirent.h>
+#include <sys/uio.h>
+#include <iostream>
+#include <vector>
+#include "json_spirit/json_spirit.h"
+
+#ifdef __linux__
+#include <limits.h>
+#include <sys/xattr.h>
+#endif
+
+
+rados_t cluster;
+
+string key;
+
+int do_mon_command(string s, string *key)
+{
+  char *outs, *outbuf;
+  size_t outs_len, outbuf_len;
+  const char *ss = s.c_str();
+  int r = rados_mon_command(cluster, (const char **)&ss, 1,
+			    0, 0,
+			    &outbuf, &outbuf_len,
+			    &outs, &outs_len);
+  if (outbuf_len) {
+    string s(outbuf, outbuf_len);
+    std::cout << "out: " << s << std::endl;
+
+    // parse out the key
+    json_spirit::mValue v, k;
+    json_spirit::read_or_throw(s, v);
+    k = v.get_array()[0].get_obj().find("key")->second;
+    *key = k.get_str();
+    std::cout << "key: " << *key << std::endl;
+    free(outbuf);
+  } else {
+    return -EINVAL;
+  }
+  if (outs_len) {
+    string s(outs, outs_len);
+    std::cout << "outs: " << s << std::endl;
+    free(outs);
+  }
+  return r;
+}
+
+string get_unique_dir()
+{
+  return string("/ceph_test_libcephfs_access.") + stringify(rand());
+}
+
+TEST(AccessTest, Foo) {
+  string dir = get_unique_dir();
+  string user = "libcephfs_foo_test." + stringify(rand());
+  // admin mount to set up test
+  struct ceph_mount_info *admin;
+  ASSERT_EQ(0, ceph_create(&admin, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(admin, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(admin, NULL));
+  ASSERT_EQ(0, ceph_mount(admin, "/"));
+  ASSERT_EQ(0, ceph_mkdir(admin, dir.c_str(), 0755));
+
+  // create access key
+  string key;
+  ASSERT_EQ(0, do_mon_command(
+      "{\"prefix\": \"auth get-or-create\", \"entity\": \"client." + user + "\", "
+      "\"caps\": [\"mon\", \"allow *\", \"osd\", \"allow rw\", "
+      "\"mds\", \"allow rw\""
+      "], \"format\": \"json\"}", &key));
+
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, user.c_str()));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_set(cmount, "key", key.c_str()));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  ceph_shutdown(cmount);
+
+  // clean up
+  ASSERT_EQ(0, ceph_rmdir(admin, dir.c_str()));
+  ceph_shutdown(admin);
+}
+
+TEST(AccessTest, Path) {
+  string good = get_unique_dir();
+  string bad = get_unique_dir();
+  string user = "libcephfs_path_test." + stringify(rand());
+  struct ceph_mount_info *admin;
+  ASSERT_EQ(0, ceph_create(&admin, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(admin, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(admin, NULL));
+  ASSERT_EQ(0, ceph_mount(admin, "/"));
+  ASSERT_EQ(0, ceph_mkdir(admin, good.c_str(), 0755));
+  ASSERT_EQ(0, ceph_mkdir(admin, string(good + "/p").c_str(), 0755));
+  ASSERT_EQ(0, ceph_mkdir(admin, bad.c_str(), 0755));
+  ASSERT_EQ(0, ceph_mkdir(admin, string(bad + "/p").c_str(), 0755));
+  int fd = ceph_open(admin, string(good + "/q").c_str(), O_CREAT|O_WRONLY, 0755);
+  ceph_close(admin, fd);
+  fd = ceph_open(admin, string(bad + "/q").c_str(), O_CREAT|O_WRONLY, 0755);
+  ceph_close(admin, fd);
+  fd = ceph_open(admin, string(bad + "/z").c_str(), O_CREAT|O_WRONLY, 0755);
+  ceph_write(admin, fd, "TEST FAILED", 11, 0);
+  ceph_close(admin, fd);
+
+  string key;
+  ASSERT_EQ(0, do_mon_command(
+      "{\"prefix\": \"auth get-or-create\", \"entity\": \"client." + user + "\", "
+      "\"caps\": [\"mon\", \"allow r\", \"osd\", \"allow rwx\", "
+      "\"mds\", \"allow r, allow rw path=" + good + "\""
+      "], \"format\": \"json\"}", &key));
+
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, user.c_str()));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_set(cmount, "key", key.c_str()));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  // allowed
+  ASSERT_GE(ceph_mkdir(cmount, string(good + "/x").c_str(), 0755), 0);
+  ASSERT_GE(ceph_rmdir(cmount, string(good + "/p").c_str()), 0);
+  ASSERT_GE(ceph_unlink(cmount, string(good + "/q").c_str()), 0);
+  fd = ceph_open(cmount, string(good + "/y").c_str(), O_CREAT|O_WRONLY, 0755);
+  ASSERT_GE(fd, 0);
+  ceph_write(cmount, fd, "bar", 3, 0);
+  ceph_close(cmount, fd);
+  ASSERT_GE(ceph_unlink(cmount, string(good + "/y").c_str()), 0);
+  ASSERT_GE(ceph_rmdir(cmount, string(good + "/x").c_str()), 0);
+
+  fd = ceph_open(cmount, string(bad + "/z").c_str(), O_RDONLY, 0644);
+  ASSERT_GE(fd, 0);
+  ceph_close(cmount, fd);
+
+  // not allowed
+  ASSERT_LT(ceph_mkdir(cmount, string(bad + "/x").c_str(), 0755), 0);
+  ASSERT_LT(ceph_rmdir(cmount, string(bad + "/p").c_str()), 0);
+  ASSERT_LT(ceph_unlink(cmount, string(bad + "/q").c_str()), 0);
+  fd = ceph_open(cmount, string(bad + "/y").c_str(), O_CREAT|O_WRONLY, 0755);
+  ASSERT_LT(fd, 0);
+
+  // unlink open file
+  fd = ceph_open(cmount, string(good + "/unlinkme").c_str(), O_CREAT|O_WRONLY, 0755);
+  ceph_unlink(cmount, string(good + "/unlinkme").c_str());
+  ASSERT_GE(ceph_write(cmount, fd, "foo", 3, 0), 0);
+  ASSERT_GE(ceph_fchmod(cmount, fd, 0777), 0);
+  ASSERT_GE(ceph_ftruncate(cmount, fd, 0), 0);
+  ASSERT_GE(ceph_fsetxattr(cmount, fd, "user.any", "bar", 3, 0), 0);
+  ceph_close(cmount, fd);
+
+  // rename open file
+  fd = ceph_open(cmount, string(good + "/renameme").c_str(), O_CREAT|O_WRONLY, 0755);
+  ASSERT_EQ(ceph_rename(admin, string(good + "/renameme").c_str(),
+			string(bad + "/asdf").c_str()), 0);
+  ASSERT_GE(ceph_write(cmount, fd, "foo", 3, 0), 0);
+  ASSERT_GE(ceph_fchmod(cmount, fd, 0777), -EACCES);
+  ASSERT_GE(ceph_ftruncate(cmount, fd, 0), -EACCES);
+  ASSERT_GE(ceph_fsetxattr(cmount, fd, "user.any", "bar", 3, 0), -EACCES);
+  ceph_close(cmount, fd);
+
+  ceph_shutdown(cmount);
+  ASSERT_EQ(0, ceph_unlink(admin, string(bad + "/q").c_str()));
+  ASSERT_EQ(0, ceph_unlink(admin, string(bad + "/z").c_str()));
+  ASSERT_EQ(0, ceph_rmdir(admin, string(bad + "/p").c_str()));
+  ASSERT_EQ(0, ceph_unlink(admin, string(bad + "/asdf").c_str()));
+  ASSERT_EQ(0, ceph_rmdir(admin, good.c_str()));
+  ASSERT_EQ(0, ceph_rmdir(admin, bad.c_str()));
+  ceph_shutdown(admin);
+}
+
+TEST(AccessTest, ReadOnly) {
+  string dir = get_unique_dir();
+  string dir2 = get_unique_dir();
+  string user = "libcephfs_readonly_test." + stringify(rand());
+  struct ceph_mount_info *admin;
+  ASSERT_EQ(0, ceph_create(&admin, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(admin, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(admin, NULL));
+  ASSERT_EQ(0, ceph_mount(admin, "/"));
+  ASSERT_EQ(0, ceph_mkdir(admin, dir.c_str(), 0755));
+  int fd = ceph_open(admin, string(dir + "/out").c_str(), O_CREAT|O_WRONLY, 0755);
+  ceph_write(admin, fd, "foo", 3, 0);
+  ceph_close(admin,fd);
+
+  string key;
+  ASSERT_EQ(0, do_mon_command(
+      "{\"prefix\": \"auth get-or-create\", \"entity\": \"client." + user + "\", "
+      "\"caps\": [\"mon\", \"allow r\", \"osd\", \"allow rw\", "
+      "\"mds\", \"allow r\""
+      "], \"format\": \"json\"}", &key));
+
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, user.c_str()));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_set(cmount, "key", key.c_str()));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  // allowed
+  fd = ceph_open(cmount, string(dir + "/out").c_str(), O_RDONLY, 0644);
+  ASSERT_GE(fd, 0);
+  ceph_close(cmount,fd);
+
+  // not allowed
+  fd = ceph_open(cmount, string(dir + "/bar").c_str(), O_CREAT|O_WRONLY, 0755);
+  ASSERT_LT(fd, 0);
+  ASSERT_LT(ceph_mkdir(cmount, dir2.c_str(), 0755), 0);
+
+  ceph_shutdown(cmount);
+  ASSERT_EQ(0, ceph_unlink(admin, string(dir + "/out").c_str()));
+  ASSERT_EQ(0, ceph_rmdir(admin, dir.c_str()));
+  ceph_shutdown(admin);
+}
+
+TEST(AccessTest, User) {
+  string dir = get_unique_dir();
+  string user = "libcephfs_user_test." + stringify(rand());
+
+  // admin mount to set up test
+  struct ceph_mount_info *admin;
+  ASSERT_EQ(0, ceph_create(&admin, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(admin, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(admin, NULL));
+  ASSERT_EQ(0, ceph_conf_set(admin, "client_permissions", "0"));
+  ASSERT_EQ(0, ceph_mount(admin, "/"));
+  ASSERT_EQ(0, ceph_mkdir(admin, dir.c_str(), 0755));
+
+  // create access key
+  string key;
+  ASSERT_EQ(0, do_mon_command(
+      "{\"prefix\": \"auth get-or-create\", \"entity\": \"client." + user + "\", "
+      "\"caps\": [\"mon\", \"allow *\", \"osd\", \"allow rw\", "
+      "\"mds\", \"allow rw uid=123 gids=456,789\""
+      "], \"format\": \"json\"}", &key));
+
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, user.c_str()));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_set(cmount, "key", key.c_str()));
+  ASSERT_EQ(-EACCES, ceph_mount(cmount, "/"));
+  ASSERT_EQ(0, ceph_init(cmount));
+
+  UserPerm *perms = ceph_userperm_new(123, 456, 0, NULL);
+  ASSERT_NE(nullptr, perms);
+  ASSERT_EQ(0, ceph_mount_perms_set(cmount, perms));
+  ceph_userperm_destroy(perms);
+
+  ASSERT_EQ(0, ceph_conf_set(cmount, "client_permissions", "0"));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  // user bits
+  ASSERT_EQ(0, ceph_chmod(admin, dir.c_str(), 0700));
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 123, 456));
+  ASSERT_EQ(0, ceph_mkdir(cmount, string(dir + "/u1").c_str(), 0755));
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 1, 456));
+  ASSERT_EQ(-EACCES, ceph_mkdir(cmount, string(dir + "/no").c_str(), 0755));
+
+  // group bits
+  ASSERT_EQ(0, ceph_chmod(admin, dir.c_str(), 0770));
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 1, 456));
+  ASSERT_EQ(0, ceph_mkdir(cmount, string(dir + "/u2").c_str(), 0755));
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 1, 2));
+  ASSERT_EQ(-EACCES, ceph_mkdir(cmount, string(dir + "/no").c_str(), 0755));
+
+  // user overrides group
+  ASSERT_EQ(0, ceph_chmod(admin, dir.c_str(), 0470));
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 123, 456));
+  ASSERT_EQ(-EACCES, ceph_mkdir(cmount, string(dir + "/no").c_str(), 0755));
+
+  // other
+  ASSERT_EQ(0, ceph_chmod(admin, dir.c_str(), 0777));
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 1, 1));
+  ASSERT_EQ(0, ceph_mkdir(cmount, string(dir + "/u3").c_str(), 0755));
+  ASSERT_EQ(0, ceph_chmod(admin, dir.c_str(), 0770));
+  ASSERT_EQ(-EACCES, ceph_mkdir(cmount, string(dir + "/no").c_str(), 0755));
+
+  // user and group overrides other
+  ASSERT_EQ(0, ceph_chmod(admin, dir.c_str(), 07));
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 1, 456));
+  ASSERT_EQ(-EACCES, ceph_mkdir(cmount, string(dir + "/no").c_str(), 0755));
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 123, 1));
+  ASSERT_EQ(-EACCES, ceph_mkdir(cmount, string(dir + "/no").c_str(), 0755));
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 123, 456));
+  ASSERT_EQ(-EACCES, ceph_mkdir(cmount, string(dir + "/no").c_str(), 0755));
+
+  // chown and chgrp
+  ASSERT_EQ(0, ceph_chmod(admin, dir.c_str(), 0700));
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 123, 456));
+  // FIXME: Re-enable these 789 tests once we can set multiple GIDs via libcephfs/config
+  // ASSERT_EQ(0, ceph_chown(cmount, dir.c_str(), 123, 789));
+  ASSERT_EQ(0, ceph_chown(cmount, dir.c_str(), 123, 456));
+  // ASSERT_EQ(0, ceph_chown(cmount, dir.c_str(), -1, 789));
+  ASSERT_EQ(0, ceph_chown(cmount, dir.c_str(), -1, 456));
+  ASSERT_EQ(-EACCES, ceph_chown(cmount, dir.c_str(), 123, 1));
+  ASSERT_EQ(-EACCES, ceph_chown(cmount, dir.c_str(), 1, 456));
+
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 1, 1));
+  ASSERT_EQ(-EACCES, ceph_chown(cmount, dir.c_str(), 123, 456));
+  ASSERT_EQ(-EACCES, ceph_chown(cmount, dir.c_str(), 123, -1));
+  ASSERT_EQ(-EACCES, ceph_chown(cmount, dir.c_str(), -1, 456));
+
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 1, 456));
+  ASSERT_EQ(-EACCES, ceph_chown(cmount, dir.c_str(), 123, 456));
+  ASSERT_EQ(-EACCES, ceph_chown(cmount, dir.c_str(), 123, -1));
+  ASSERT_EQ(-EACCES, ceph_chown(cmount, dir.c_str(), -1, 456));
+
+  ASSERT_EQ(0, ceph_chown(admin, dir.c_str(), 123, 1));
+  ASSERT_EQ(0, ceph_chown(cmount, dir.c_str(), -1, 456));
+  // ASSERT_EQ(0, ceph_chown(cmount, dir.c_str(), 123, 789));
+
+  ceph_shutdown(cmount);
+
+  // clean up
+  ASSERT_EQ(0, ceph_rmdir(admin, string(dir + "/u1").c_str()));
+  ASSERT_EQ(0, ceph_rmdir(admin, string(dir + "/u2").c_str()));
+  ASSERT_EQ(0, ceph_rmdir(admin, string(dir + "/u3").c_str()));
+  ASSERT_EQ(0, ceph_rmdir(admin, dir.c_str()));
+  ceph_shutdown(admin);
+}
+
+static int update_root_mode()
+{
+  struct ceph_mount_info *admin;
+  int r = ceph_create(&admin, NULL);
+  if (r < 0)
+    return r;
+  ceph_conf_read_file(admin, NULL);
+  ceph_conf_parse_env(admin, NULL);
+  ceph_conf_set(admin, "client_permissions", "false");
+  r = ceph_mount(admin, "/");
+  if (r < 0)
+    goto out;
+  r = ceph_chmod(admin, "/", 0777);
+out:
+  ceph_shutdown(admin);
+  return r;
+}
+
+
+int main(int argc, char **argv)
+{
+  int r = update_root_mode();
+  if (r < 0)
+    exit(1);
+
+  ::testing::InitGoogleTest(&argc, argv);
+
+  srand(getpid());
+
+  r = rados_create(&cluster, NULL);
+  if (r < 0)
+    exit(1);
+  
+  r = rados_conf_read_file(cluster, NULL);
+  if (r < 0)
+    exit(1);
+
+  rados_conf_parse_env(cluster, NULL);
+  r = rados_connect(cluster);
+  if (r < 0)
+    exit(1);
+
+  r = RUN_ALL_TESTS();
+
+  rados_shutdown(cluster);
+
+  return r;
+}
diff --git a/src/test/libcephfs/acl.cc b/src/test/libcephfs/acl.cc
new file mode 100644
index 00000000..a6c4a659
--- /dev/null
+++ b/src/test/libcephfs/acl.cc
@@ -0,0 +1,309 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 New Dream Network
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+#include "include/types.h"
+#include "gtest/gtest.h"
+#include "include/cephfs/libcephfs.h"
+#include "include/ceph_fs.h"
+#include "client/posix_acl.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#ifdef __linux__
+#include <sys/xattr.h>
+#endif
+
+static size_t acl_ea_size(int count)
+{
+  return sizeof(acl_ea_header) + count * sizeof(acl_ea_entry);
+}
+
+static int acl_ea_count(size_t size)
+{
+  if (size < sizeof(acl_ea_header))
+    return -1;
+  size -= sizeof(acl_ea_header);
+  if (size % sizeof(acl_ea_entry))
+    return -1;
+  return size / sizeof(acl_ea_entry);
+}
+
+static int check_acl_and_mode(const void *buf, size_t size, mode_t mode)
+{
+  const acl_ea_entry *group_entry = NULL, *mask_entry = NULL;
+  const acl_ea_header *header = reinterpret_cast<const acl_ea_header*>(buf);
+  const acl_ea_entry *entry = header->a_entries;
+  int count = (size - sizeof(*header)) / sizeof(*entry);
+  for (int i = 0; i < count; ++i) {
+    __u16 tag = entry->e_tag;
+    __u16 perm = entry->e_perm;
+    switch(tag) {
+      case ACL_USER_OBJ:
+	if (perm != ((mode >> 6) & 7))
+	  return -EINVAL;
+	break;
+      case ACL_USER:
+      case ACL_GROUP:
+	break;
+      case ACL_GROUP_OBJ:
+	group_entry = entry;
+	break;
+      case ACL_OTHER:
+	if (perm != (mode & 7))
+	  return -EINVAL;
+	break;
+      case ACL_MASK:
+	mask_entry = entry;
+	break;
+      default:
+	return -EIO;
+    }
+    ++entry;
+  }
+  if (mask_entry) {
+    __u16 perm = mask_entry->e_perm;
+    if (perm != ((mode >> 3) & 7))
+      return -EINVAL;
+  } else {
+    if (!group_entry)
+      return -EIO;
+    __u16 perm = group_entry->e_perm;
+    if (perm != ((mode >> 3) & 7))
+      return -EINVAL;
+  }
+  return 0;
+}
+
+static int generate_test_acl(void *buf, size_t size, mode_t mode)
+{
+  if (acl_ea_count(size) != 5)
+    return -1;
+  acl_ea_header *header = reinterpret_cast<acl_ea_header*>(buf);
+  header->a_version = (__u32)ACL_EA_VERSION;
+  acl_ea_entry *entry = header->a_entries;
+  entry->e_tag = ACL_USER_OBJ;
+  entry->e_perm = (mode >> 6) & 7;
+  ++entry;
+  entry->e_tag = ACL_USER;
+  entry->e_perm = 7;
+  entry->e_id = getuid();
+  ++entry;
+  entry->e_tag = ACL_GROUP_OBJ;
+  entry->e_perm = (mode >> 3) & 7;
+  ++entry;
+  entry->e_tag = ACL_MASK;
+  entry->e_perm = 7;
+  ++entry;
+  entry->e_tag = ACL_OTHER;
+  entry->e_perm = mode & 7;
+  return 0;
+}
+
+static int generate_empty_acl(void *buf, size_t size, mode_t mode)
+{
+
+ if (acl_ea_count(size) != 3)
+    return -1;
+  acl_ea_header *header = reinterpret_cast<acl_ea_header*>(buf);
+  header->a_version = (__u32)ACL_EA_VERSION;
+  acl_ea_entry *entry = header->a_entries;
+  entry->e_tag = ACL_USER_OBJ;
+  entry->e_perm = (mode >> 6) & 7;
+  ++entry;
+  entry->e_tag = ACL_GROUP_OBJ;
+  entry->e_perm = (mode >> 3) & 7;
+  ++entry;
+  entry->e_tag = ACL_OTHER;
+  entry->e_perm = mode & 7;
+  return 0;
+}
+
+TEST(ACL, SetACL) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+  ASSERT_EQ(0, ceph_conf_set(cmount, "client_acl_type", "posix_acl"));
+  ASSERT_EQ(0, ceph_conf_set(cmount, "client_permissions", "0"));
+
+  char test_file[256];
+  sprintf(test_file, "file1_setacl_%d", getpid());
+
+  int fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0600);
+  ASSERT_GT(fd, 0);
+  // change ownership to nobody -- we assume nobody exists and id is always 65534
+  ASSERT_EQ(ceph_fchown(cmount, fd, 65534, 65534), 0);
+
+  ASSERT_EQ(0, ceph_conf_set(cmount, "client_permissions", "1"));
+  ASSERT_EQ(ceph_open(cmount, test_file, O_RDWR, 0), -EACCES);
+  ASSERT_EQ(0, ceph_conf_set(cmount, "client_permissions", "0"));
+
+  size_t acl_buf_size = acl_ea_size(5);
+  void *acl_buf = malloc(acl_buf_size);
+  ASSERT_EQ(generate_test_acl(acl_buf, acl_buf_size, 0750), 0);
+
+  // can't set default acl for non-directory
+  ASSERT_EQ(ceph_fsetxattr(cmount, fd, ACL_EA_DEFAULT, acl_buf, acl_buf_size, 0), -EACCES);
+  ASSERT_EQ(ceph_fsetxattr(cmount, fd, ACL_EA_ACCESS, acl_buf, acl_buf_size, 0), 0);
+
+  int tmpfd = ceph_open(cmount, test_file, O_RDWR, 0);
+  ASSERT_GT(tmpfd, 0);
+  ceph_close(cmount, tmpfd);
+
+  struct ceph_statx stx;
+  ASSERT_EQ(ceph_fstatx(cmount, fd, &stx, CEPH_STATX_MODE, 0), 0);
+  // mode was modified according to ACL
+  ASSERT_EQ(stx.stx_mode & 0777u, 0770u);
+  ASSERT_EQ(check_acl_and_mode(acl_buf, acl_buf_size, stx.stx_mode), 0);
+
+  acl_buf_size = acl_ea_size(3);
+  // setting ACL that is equivalent to file mode
+  ASSERT_EQ(generate_empty_acl(acl_buf, acl_buf_size, 0600), 0);
+  ASSERT_EQ(ceph_fsetxattr(cmount, fd, ACL_EA_ACCESS, acl_buf, acl_buf_size, 0), 0);
+  // ACL was deleted
+  ASSERT_EQ(ceph_fgetxattr(cmount, fd, ACL_EA_ACCESS, NULL, 0), -ENODATA);
+
+  ASSERT_EQ(ceph_fstatx(cmount, fd, &stx, CEPH_STATX_MODE, 0), 0);
+  // mode was modified according to ACL
+  ASSERT_EQ(stx.stx_mode & 0777u, 0600u);
+
+  free(acl_buf);
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(ACL, Chmod) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+  ASSERT_EQ(0, ceph_conf_set(cmount, "client_acl_type", "posix_acl"));
+
+  char test_file[256];
+  sprintf(test_file, "file1_acl_chmod_%d", getpid());
+
+  int fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0600);
+  ASSERT_GT(fd, 0);
+
+  int acl_buf_size = acl_ea_size(5);
+  void *acl_buf = malloc(acl_buf_size);
+  ASSERT_EQ(generate_test_acl(acl_buf, acl_buf_size, 0775), 0);
+  ASSERT_EQ(ceph_fsetxattr(cmount, fd, ACL_EA_ACCESS, acl_buf, acl_buf_size, 0), 0);
+
+  struct ceph_statx stx;
+  ASSERT_EQ(ceph_fstatx(cmount, fd, &stx, CEPH_STATX_MODE, 0), 0);
+  // mode was updated according to ACL
+  ASSERT_EQ(stx.stx_mode & 0777u, 0775u);
+
+  // change mode
+  ASSERT_EQ(ceph_fchmod(cmount, fd, 0640), 0);
+
+  ASSERT_EQ(ceph_fstatx(cmount, fd, &stx, CEPH_STATX_MODE, 0), 0);
+  ASSERT_EQ(stx.stx_mode & 0777u, 0640u);
+
+  // ACL was updated according to mode
+  ASSERT_EQ(ceph_fgetxattr(cmount, fd, ACL_EA_ACCESS, acl_buf, acl_buf_size), acl_buf_size);
+  ASSERT_EQ(check_acl_and_mode(acl_buf, acl_buf_size, stx.stx_mode), 0);
+
+  free(acl_buf);
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(ACL, DefaultACL) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+  ASSERT_EQ(0, ceph_conf_set(cmount, "client_acl_type", "posix_acl"));
+
+  int acl_buf_size = acl_ea_size(5);
+  void *acl1_buf = malloc(acl_buf_size);
+  void *acl2_buf = malloc(acl_buf_size);
+
+  ASSERT_EQ(generate_test_acl(acl1_buf, acl_buf_size, 0750), 0);
+
+  char test_dir1[256];
+  sprintf(test_dir1, "dir1_acl_default_%d", getpid());
+  ASSERT_EQ(ceph_mkdir(cmount, test_dir1, 0750), 0);
+
+  // set default acl
+  ASSERT_EQ(ceph_setxattr(cmount, test_dir1, ACL_EA_DEFAULT, acl1_buf, acl_buf_size, 0), 0);
+
+  char test_dir2[262];
+  sprintf(test_dir2, "%s/dir2", test_dir1);
+  ASSERT_EQ(ceph_mkdir(cmount, test_dir2, 0755), 0);
+
+  // inherit default acl
+  ASSERT_EQ(ceph_getxattr(cmount, test_dir2, ACL_EA_DEFAULT, acl2_buf, acl_buf_size), acl_buf_size);
+  ASSERT_EQ(memcmp(acl1_buf, acl2_buf, acl_buf_size), 0);
+
+  // mode and ACL are updated
+  ASSERT_EQ(ceph_getxattr(cmount, test_dir2, ACL_EA_ACCESS, acl2_buf, acl_buf_size), acl_buf_size);
+  {
+    struct ceph_statx stx;
+    ASSERT_EQ(ceph_statx(cmount, test_dir2, &stx, CEPH_STATX_MODE, 0), 0);
+    // other bits of mode &= acl other perm
+    ASSERT_EQ(stx.stx_mode & 0777u, 0750u);
+    ASSERT_EQ(check_acl_and_mode(acl2_buf, acl_buf_size, stx.stx_mode), 0);
+  }
+
+  char test_file1[262];
+  sprintf(test_file1, "%s/file1", test_dir1);
+  int fd = ceph_open(cmount, test_file1, O_CREAT|O_RDWR, 0666);
+  ASSERT_GT(fd, 0);
+
+  // no default acl
+  ASSERT_EQ(ceph_fgetxattr(cmount, fd, ACL_EA_DEFAULT, NULL, 0), -ENODATA);
+
+  // mode and ACL are updated
+  ASSERT_EQ(ceph_fgetxattr(cmount, fd, ACL_EA_ACCESS, acl2_buf, acl_buf_size), acl_buf_size);
+  {
+    struct ceph_statx stx;
+    ASSERT_EQ(ceph_statx(cmount, test_file1, &stx, CEPH_STATX_MODE, 0), 0);
+    // other bits of mode &= acl other perm
+    ASSERT_EQ(stx.stx_mode & 0777u, 0660u);
+    ASSERT_EQ(check_acl_and_mode(acl2_buf, acl_buf_size, stx.stx_mode), 0);
+  }
+
+  free(acl1_buf);
+  free(acl2_buf);
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(ACL, Disabled) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+  ASSERT_EQ(0, ceph_conf_set(cmount, "client_acl_type", ""));
+
+  size_t acl_buf_size = acl_ea_size(3);
+  void *acl_buf = malloc(acl_buf_size);
+  ASSERT_EQ(generate_empty_acl(acl_buf, acl_buf_size, 0755), 0);
+
+  char test_dir[256];
+  sprintf(test_dir, "dir1_acl_disabled_%d", getpid());
+  ASSERT_EQ(ceph_mkdir(cmount, test_dir, 0750), 0);
+
+  ASSERT_EQ(ceph_setxattr(cmount, test_dir, ACL_EA_DEFAULT, acl_buf, acl_buf_size, 0), -EOPNOTSUPP);
+  ASSERT_EQ(ceph_setxattr(cmount, test_dir, ACL_EA_ACCESS, acl_buf, acl_buf_size, 0), -EOPNOTSUPP);
+  ASSERT_EQ(ceph_getxattr(cmount, test_dir, ACL_EA_DEFAULT, acl_buf, acl_buf_size), -EOPNOTSUPP);
+  ASSERT_EQ(ceph_getxattr(cmount, test_dir, ACL_EA_ACCESS, acl_buf, acl_buf_size), -EOPNOTSUPP);
+
+  free(acl_buf);
+  ceph_shutdown(cmount);
+}
diff --git a/src/test/libcephfs/caps.cc b/src/test/libcephfs/caps.cc
new file mode 100644
index 00000000..b4976c9c
--- /dev/null
+++ b/src/test/libcephfs/caps.cc
@@ -0,0 +1,94 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 New Dream Network
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+#include "include/int_types.h"
+
+#include "gtest/gtest.h"
+#include "include/ceph_fs.h"
+#include "include/cephfs/libcephfs.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <dirent.h>
+#ifdef __linux__
+#include <sys/xattr.h>
+#endif
+#include <signal.h>
+
+TEST(Caps, ReadZero) {
+
+  int mypid = getpid();
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  int i = 0;
+  for(; i < 30; ++i) {
+
+    char c_path[1024];
+    sprintf(c_path, "/caps_rzfile_%d_%d", mypid, i);
+    int fd = ceph_open(cmount, c_path, O_CREAT|O_TRUNC|O_WRONLY, 0644);
+    ASSERT_LT(0, fd);
+
+    int expect = CEPH_CAP_FILE_EXCL | CEPH_CAP_FILE_WR | CEPH_CAP_FILE_BUFFER;
+    int caps = ceph_debug_get_fd_caps(cmount, fd);
+
+    ASSERT_EQ(expect, caps & expect);
+    ASSERT_EQ(0, ceph_close(cmount, fd));
+
+    caps = ceph_debug_get_file_caps(cmount, c_path);
+    ASSERT_EQ(expect, caps & expect);
+
+    char cw_path[1024];
+    sprintf(cw_path, "/caps_wzfile_%d_%d", mypid, i);
+    int wfd = ceph_open(cmount, cw_path, O_CREAT|O_TRUNC|O_WRONLY, 0644);
+    ASSERT_LT(0, wfd);
+
+    char wbuf[4096];
+    ASSERT_EQ(4096, ceph_write(cmount, wfd, wbuf, 4096, 0));
+
+    ASSERT_EQ(0, ceph_close(cmount, wfd));
+
+    struct ceph_statx stx;
+    ASSERT_EQ(0, ceph_statx(cmount, c_path, &stx, CEPH_STATX_MTIME, 0));
+
+    caps = ceph_debug_get_file_caps(cmount, c_path);
+    ASSERT_EQ(expect, caps & expect);
+  }
+
+  ASSERT_EQ(0, ceph_conf_set(cmount, "client_debug_inject_tick_delay", "20"));
+
+  for(i = 0; i < 30; ++i) {
+
+    char c_path[1024];
+    sprintf(c_path, "/caps_rzfile_%d_%d", mypid, i);
+
+    int fd = ceph_open(cmount, c_path, O_RDONLY, 0);
+    ASSERT_LT(0, fd);
+    char buf[256];
+
+    int expect = CEPH_CAP_FILE_RD | CEPH_STAT_CAP_SIZE | CEPH_CAP_FILE_CACHE;
+    int caps = ceph_debug_get_fd_caps(cmount, fd);
+    ASSERT_EQ(expect, caps & expect);
+    ASSERT_EQ(0, ceph_read(cmount, fd, buf, 256, 0));
+
+    caps = ceph_debug_get_fd_caps(cmount, fd);
+    ASSERT_EQ(expect, caps & expect);
+    ASSERT_EQ(0, ceph_close(cmount, fd));
+
+  }
+  ceph_shutdown(cmount);
+}
diff --git a/src/test/libcephfs/ceph_pthread_self.h b/src/test/libcephfs/ceph_pthread_self.h
new file mode 100644
index 00000000..4c0c98f6
--- /dev/null
+++ b/src/test/libcephfs/ceph_pthread_self.h
@@ -0,0 +1,31 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_TEST_LIBCEPHFS_PTHREAD_SELF
+#define CEPH_TEST_LIBCEPHFS_PTHREAD_SELF
+
+#include <pthread.h>
+
+#include <type_traits>
+
+/*
+ * There is a  difference between libc shipped with FreeBSD and
+ * glibc shipped with GNU/Linux for the return type of pthread_self().
+ *
+ * Introduced a conversion function in include/compat.h
+ *     (uint64_t)ceph_pthread_self()
+ *
+ * libc returns an opague pthread_t that is not default convertable
+ * to a uint64_t, which is what gtest expects.
+ * And tests using gtest will not compile because of this difference.
+ * 
+ */
+static uint64_t ceph_pthread_self() {
+  auto me = pthread_self();
+  static_assert(std::is_convertible_v<decltype(me), uint64_t> ||
+                std::is_pointer_v<decltype(me)>,
+                "we need to use pthread_self() for the owner parameter");
+  return reinterpret_cast<uint64_t>(me);
+}
+
+#endif
diff --git a/src/test/libcephfs/deleg.cc b/src/test/libcephfs/deleg.cc
new file mode 100644
index 00000000..1a6376ee
--- /dev/null
+++ b/src/test/libcephfs/deleg.cc
@@ -0,0 +1,385 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Tests for Ceph delegation handling
+ *
+ * (c) 2017, Jeff Layton <jlayton@redhat.com>
+ */
+
+#include "gtest/gtest.h"
+#include "include/cephfs/libcephfs.h"
+#include "include/stat.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <dirent.h>
+#include <sys/uio.h>
+
+#ifdef __linux__
+#include <limits.h>
+#include <sys/xattr.h>
+#endif
+
+#include <map>
+#include <vector>
+#include <thread>
+#include <atomic>
+
+#include "include/ceph_assert.h"
+
+/* in ms -- 1 minute */
+#define MAX_WAIT	(60 * 1000)
+
+static void wait_for_atomic_bool(std::atomic_bool &recalled)
+{
+  int i = 0;
+
+  while (!recalled.load()) {
+    ASSERT_LT(i++, MAX_WAIT);
+    usleep(1000);
+  }
+}
+
+static int set_default_deleg_timeout(struct ceph_mount_info *cmount)
+{
+  uint32_t session_timeout = ceph_get_cap_return_timeout(cmount);
+  return ceph_set_deleg_timeout(cmount, session_timeout - 1);
+}
+
+static void dummy_deleg_cb(Fh *fh, void *priv)
+{
+  std::atomic_bool *recalled = (std::atomic_bool *)priv;
+  recalled->store(true);
+}
+
+static void open_breaker_func(struct ceph_mount_info *cmount, const char *filename, int flags, std::atomic_bool *opened)
+{
+  bool do_shutdown = false;
+
+  if (!cmount) {
+    ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+    ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+    ASSERT_EQ(ceph_conf_parse_env(cmount, NULL), 0);
+    ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+    ASSERT_EQ(set_default_deleg_timeout(cmount), 0);
+    do_shutdown = true;
+  }
+
+  Inode *root, *file;
+  ASSERT_EQ(ceph_ll_lookup_root(cmount, &root), 0);
+
+  Fh *fh;
+  struct ceph_statx stx;
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  ASSERT_EQ(ceph_ll_lookup(cmount, root, filename, &file, &stx, CEPH_STATX_ALL_STATS, 0, perms), 0);
+  int ret, i = 0;
+  for (;;) {
+    ASSERT_EQ(ceph_ll_getattr(cmount, file, &stx, CEPH_STATX_ALL_STATS, 0, perms), 0);
+    ret = ceph_ll_open(cmount, file, flags, &fh, perms);
+    if (ret != -EAGAIN)
+      break;
+    ASSERT_LT(i++, MAX_WAIT);
+    usleep(1000);
+  }
+  ASSERT_EQ(ret, 0);
+  opened->store(true);
+  ASSERT_EQ(ceph_ll_close(cmount, fh), 0);
+
+  if (do_shutdown)
+    ceph_shutdown(cmount);
+}
+
+enum {
+  DelegTestLink,
+  DelegTestRename,
+  DelegTestUnlink
+};
+
+static void namespace_breaker_func(struct ceph_mount_info *cmount, int cmd, const char *oldname, const char *newname)
+{
+  bool do_shutdown = false;
+
+  if (!cmount) {
+    ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+    ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+    ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+    ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+    ASSERT_EQ(set_default_deleg_timeout(cmount), 0);
+    do_shutdown = true;
+  }
+
+  Inode *root, *file = nullptr;
+  ASSERT_EQ(ceph_ll_lookup_root(cmount, &root), 0);
+
+  struct ceph_statx stx;
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  int ret, i = 0;
+  for (;;) {
+    switch (cmd) {
+    case DelegTestRename:
+      ret = ceph_ll_rename(cmount, root, oldname, root, newname, perms);
+      break;
+    case DelegTestLink:
+      if (!file) {
+	ASSERT_EQ(ceph_ll_lookup(cmount, root, oldname, &file, &stx, 0, 0, perms), 0);
+      }
+      ret = ceph_ll_link(cmount, file, root, newname, perms);
+      break;
+    case DelegTestUnlink:
+      ret = ceph_ll_unlink(cmount, root, oldname, perms);
+      break;
+    default:
+      // Bad command
+      ceph_abort();
+    }
+    if (ret != -EAGAIN)
+      break;
+    ASSERT_LT(i++, MAX_WAIT);
+    usleep(1000);
+  }
+  ASSERT_EQ(ret, 0);
+
+  if (do_shutdown)
+    ceph_shutdown(cmount);
+}
+
+static void simple_deleg_test(struct ceph_mount_info *cmount, struct ceph_mount_info *tcmount)
+{
+  Inode *root, *file;
+
+  ASSERT_EQ(ceph_ll_lookup_root(cmount, &root), 0);
+
+  char filename[32];
+
+  Fh *fh;
+  struct ceph_statx stx;
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  std::atomic_bool recalled(false);
+  std::atomic_bool opened(false);
+
+  // ensure r/w open breaks a r/w delegation
+  sprintf(filename, "deleg.rwrw.%x", getpid());
+  ASSERT_EQ(ceph_ll_create(cmount, root, filename, 0666,
+		    O_RDWR|O_CREAT|O_EXCL, &file, &fh, &stx, 0, 0, perms), 0);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_WR, dummy_deleg_cb, &recalled), 0);
+  std::thread breaker1(open_breaker_func, tcmount, filename, O_RDWR, &opened);
+
+  wait_for_atomic_bool(recalled);
+  ASSERT_EQ(opened.load(), false);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_NONE, dummy_deleg_cb, &recalled), 0);
+  breaker1.join();
+  ASSERT_EQ(ceph_ll_close(cmount, fh), 0);
+  ASSERT_EQ(ceph_ll_unlink(cmount, root, filename, perms), 0);
+
+  // ensure r/o open breaks a r/w delegation
+  recalled.store(false);
+  opened.store(false);
+  sprintf(filename, "deleg.rorw.%x", getpid());
+  ASSERT_EQ(ceph_ll_create(cmount, root, filename, 0666,
+		    O_RDWR|O_CREAT|O_EXCL, &file, &fh, &stx, 0, 0, perms), 0);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_WR, dummy_deleg_cb, &recalled), 0);
+  std::thread breaker2(open_breaker_func, tcmount, filename, O_RDONLY, &opened);
+  wait_for_atomic_bool(recalled);
+  ASSERT_EQ(opened.load(), false);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_NONE, dummy_deleg_cb, &recalled), 0);
+  breaker2.join();
+  ASSERT_EQ(ceph_ll_close(cmount, fh), 0);
+  ASSERT_EQ(ceph_ll_unlink(cmount, root, filename, perms), 0);
+
+  // ensure r/o open does not break a r/o delegation
+  sprintf(filename, "deleg.rwro.%x", getpid());
+  ASSERT_EQ(ceph_ll_create(cmount, root, filename, 0666,
+		    O_RDONLY|O_CREAT|O_EXCL, &file, &fh, &stx, 0, 0, perms), 0);
+  recalled.store(false);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_RD, dummy_deleg_cb, &recalled), 0);
+  std::thread breaker3(open_breaker_func, tcmount, filename, O_RDONLY, &opened);
+  breaker3.join();
+  ASSERT_EQ(recalled.load(), false);
+
+  // ensure that r/w open breaks r/o delegation
+  opened.store(false);
+  std::thread breaker4(open_breaker_func, tcmount, filename, O_WRONLY, &opened);
+  wait_for_atomic_bool(recalled);
+  usleep(1000);
+  ASSERT_EQ(opened.load(), false);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_NONE, dummy_deleg_cb, &recalled), 0);
+  breaker4.join();
+  ASSERT_EQ(ceph_ll_close(cmount, fh), 0);
+  ASSERT_EQ(ceph_ll_unlink(cmount, root, filename, perms), 0);
+
+  // ensure hardlinking breaks a r/w delegation
+  recalled.store(false);
+  char newname[32];
+  sprintf(filename, "deleg.old.%x", getpid());
+  sprintf(newname, "deleg.new.%x", getpid());
+  ASSERT_EQ(ceph_ll_create(cmount, root, filename, 0666,
+		    O_RDWR|O_CREAT|O_EXCL, &file, &fh, &stx, 0, 0, perms), 0);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_WR, dummy_deleg_cb, &recalled), 0);
+  std::thread breaker5(namespace_breaker_func, tcmount, DelegTestLink, filename, newname);
+  wait_for_atomic_bool(recalled);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_NONE, dummy_deleg_cb, &recalled), 0);
+  breaker5.join();
+  ASSERT_EQ(ceph_ll_close(cmount, fh), 0);
+  ASSERT_EQ(ceph_ll_unlink(cmount, root, filename, perms), 0);
+  ASSERT_EQ(ceph_ll_unlink(cmount, root, newname, perms), 0);
+
+  // ensure renaming breaks a r/w delegation
+  recalled.store(false);
+  ASSERT_EQ(ceph_ll_create(cmount, root, filename, 0666,
+		    O_RDWR|O_CREAT|O_EXCL, &file, &fh, &stx, 0, 0, perms), 0);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_WR, dummy_deleg_cb, &recalled), 0);
+  std::thread breaker6(namespace_breaker_func, tcmount, DelegTestRename, filename, newname);
+  wait_for_atomic_bool(recalled);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_NONE, dummy_deleg_cb, &recalled), 0);
+  breaker6.join();
+  ASSERT_EQ(ceph_ll_close(cmount, fh), 0);
+  ASSERT_EQ(ceph_ll_unlink(cmount, root, newname, perms), 0);
+
+  // ensure unlinking breaks a r/w delegation
+  recalled.store(false);
+  ASSERT_EQ(ceph_ll_create(cmount, root, filename, 0666,
+		    O_RDWR|O_CREAT|O_EXCL, &file, &fh, &stx, 0, 0, perms), 0);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_WR, dummy_deleg_cb, &recalled), 0);
+  std::thread breaker7(namespace_breaker_func, tcmount, DelegTestUnlink, filename, nullptr);
+  wait_for_atomic_bool(recalled);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_NONE, dummy_deleg_cb, &recalled), 0);
+  breaker7.join();
+  ASSERT_EQ(ceph_ll_close(cmount, fh), 0);
+}
+
+TEST(LibCephFS, DelegMultiClient) {
+  struct ceph_mount_info *cmount;
+
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+  ASSERT_EQ(set_default_deleg_timeout(cmount), 0);
+
+  simple_deleg_test(cmount, nullptr);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, DelegSingleClient) {
+  struct ceph_mount_info *cmount;
+
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+  ASSERT_EQ(set_default_deleg_timeout(cmount), 0);
+
+  simple_deleg_test(cmount, cmount);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, DelegTimeout) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+  // tweak timeout to run quickly, since we don't plan to return it anyway
+  ASSERT_EQ(ceph_set_deleg_timeout(cmount, 2), 0);
+
+  Inode *root, *file;
+  ASSERT_EQ(ceph_ll_lookup_root(cmount, &root), 0);
+
+  char filename[32];
+  sprintf(filename, "delegtimeo%x", getpid());
+
+  Fh *fh;
+  struct ceph_statx stx;
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  ASSERT_EQ(ceph_ll_create(cmount, root, filename, 0666,
+		    O_RDWR|O_CREAT|O_EXCL, &file, &fh, &stx, 0, 0, perms), 0);
+
+  /* Reopen read-only */
+  ASSERT_EQ(ceph_ll_close(cmount, fh), 0);
+  ASSERT_EQ(ceph_ll_open(cmount, file, O_RDONLY, &fh, perms), 0);
+
+  std::atomic_bool recalled(false);
+  ASSERT_EQ(ceph_ll_delegation(cmount, fh, CEPH_DELEGATION_RD, dummy_deleg_cb, &recalled), 0);
+  std::atomic_bool opened(false);
+  std::thread breaker1(open_breaker_func, nullptr, filename, O_RDWR, &opened);
+  breaker1.join();
+  ASSERT_EQ(recalled.load(), true);
+  ASSERT_EQ(ceph_ll_getattr(cmount, root, &stx, 0, 0, perms), -ENOTCONN);
+  ceph_release(cmount);
+}
+
+TEST(LibCephFS, RecalledGetattr) {
+  struct ceph_mount_info *cmount1;
+  ASSERT_EQ(ceph_create(&cmount1, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount1, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount1, NULL));
+  ASSERT_EQ(ceph_mount(cmount1, "/"), 0);
+  ASSERT_EQ(set_default_deleg_timeout(cmount1), 0);
+
+  Inode *root, *file;
+  ASSERT_EQ(ceph_ll_lookup_root(cmount1, &root), 0);
+
+  char filename[32];
+  sprintf(filename, "recalledgetattr%x", getpid());
+
+  Fh *fh;
+  struct ceph_statx stx;
+  UserPerm *perms = ceph_mount_perms(cmount1);
+
+  ASSERT_EQ(ceph_ll_create(cmount1, root, filename, 0666,
+		    O_RDWR|O_CREAT|O_EXCL, &file, &fh, &stx, 0, 0, perms), 0);
+  ASSERT_EQ(ceph_ll_write(cmount1, fh, 0, sizeof(filename), filename),
+	    static_cast<int>(sizeof(filename)));
+  ASSERT_EQ(ceph_ll_close(cmount1, fh), 0);
+
+  /* New mount for read delegation */
+  struct ceph_mount_info *cmount2;
+  ASSERT_EQ(ceph_create(&cmount2, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount2, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount2, NULL));
+  ASSERT_EQ(ceph_mount(cmount2, "/"), 0);
+  ASSERT_EQ(set_default_deleg_timeout(cmount2), 0);
+
+  ASSERT_EQ(ceph_ll_lookup_root(cmount2, &root), 0);
+  perms = ceph_mount_perms(cmount2);
+  ASSERT_EQ(ceph_ll_lookup(cmount2, root, filename, &file, &stx, 0, 0, perms), 0);
+
+  ASSERT_EQ(ceph_ll_open(cmount2, file, O_WRONLY, &fh, perms), 0);
+  ASSERT_EQ(ceph_ll_write(cmount2, fh, 0, sizeof(filename), filename),
+	    static_cast<int>(sizeof(filename)));
+  ASSERT_EQ(ceph_ll_close(cmount2, fh), 0);
+
+  ASSERT_EQ(ceph_ll_open(cmount2, file, O_RDONLY, &fh, perms), 0);
+
+  /* Break delegation */
+  std::atomic_bool recalled(false);
+  ASSERT_EQ(ceph_ll_delegation(cmount2, fh, CEPH_DELEGATION_RD, dummy_deleg_cb, &recalled), 0);
+  ASSERT_EQ(ceph_ll_read(cmount2, fh, 0, sizeof(filename), filename),
+	    static_cast<int>(sizeof(filename)));
+  ASSERT_EQ(ceph_ll_getattr(cmount2, file, &stx, CEPH_STATX_ALL_STATS, 0, perms), 0);
+  std::atomic_bool opened(false);
+  std::thread breaker1(open_breaker_func, cmount1, filename, O_WRONLY, &opened);
+  int i = 0;
+  do {
+    ASSERT_EQ(ceph_ll_getattr(cmount2, file, &stx, CEPH_STATX_ALL_STATS, 0, perms), 0);
+    ASSERT_LT(i++, MAX_WAIT);
+    usleep(1000);
+  } while (!recalled.load());
+  ASSERT_EQ(opened.load(), false);
+  ASSERT_EQ(ceph_ll_getattr(cmount2, file, &stx, CEPH_STATX_ALL_STATS, 0, perms), 0);
+  ASSERT_EQ(ceph_ll_delegation(cmount2, fh, CEPH_DELEGATION_NONE, dummy_deleg_cb, nullptr), 0);
+  breaker1.join();
+  ASSERT_EQ(ceph_ll_close(cmount2, fh), 0);
+  ceph_unmount(cmount2);
+  ceph_release(cmount2);
+  ceph_unmount(cmount1);
+  ceph_release(cmount1);
+}
diff --git a/src/test/libcephfs/flock.cc b/src/test/libcephfs/flock.cc
new file mode 100644
index 00000000..9add019d
--- /dev/null
+++ b/src/test/libcephfs/flock.cc
@@ -0,0 +1,645 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 New Dream Network
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include <pthread.h>
+#include "gtest/gtest.h"
+#ifndef GTEST_IS_THREADSAFE
+#error "!GTEST_IS_THREADSAFE"
+#endif
+
+#include "include/cephfs/libcephfs.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/file.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <dirent.h>
+#include <stdlib.h>
+#include <semaphore.h>
+#include <time.h>
+#include <sys/mman.h>
+
+#ifdef __linux__
+#include <limits.h>
+#include <sys/xattr.h>
+#elif __FreeBSD__
+#include <sys/types.h>
+#include <sys/wait.h>
+#endif
+
+#include "include/ceph_assert.h"
+#include "ceph_pthread_self.h"
+
+// Startup common: create and mount ceph fs
+#define STARTUP_CEPH() do {				\
+    ASSERT_EQ(0, ceph_create(&cmount, NULL));		\
+    ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));	\
+    ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));	\
+    ASSERT_EQ(0, ceph_mount(cmount, NULL));		\
+  } while(0)
+
+// Cleanup common: unmount and release ceph fs
+#define CLEANUP_CEPH() do {			\
+    ASSERT_EQ(0, ceph_unmount(cmount));		\
+    ASSERT_EQ(0, ceph_release(cmount));		\
+  } while(0)
+
+static const mode_t fileMode = S_IRWXU | S_IRWXG | S_IRWXO;
+
+// Default wait time for normal and "slow" operations
+// (5" should be enough in case of network congestion)
+static const long waitMs = 10;
+static const long waitSlowMs = 5000;
+
+// Get the absolute struct timespec reference from now + 'ms' milliseconds
+static const struct timespec* abstime(struct timespec &ts, long ms) {
+  if (clock_gettime(CLOCK_REALTIME, &ts) == -1) {
+    ceph_abort();
+  }
+  ts.tv_nsec += ms * 1000000;
+  ts.tv_sec += ts.tv_nsec / 1000000000;
+  ts.tv_nsec %= 1000000000;
+  return &ts;
+}
+
+/* Basic locking */
+TEST(LibCephFS, BasicLocking) {
+  struct ceph_mount_info *cmount = NULL;
+  STARTUP_CEPH();
+
+  char c_file[1024];
+  sprintf(c_file, "/flock_test_%d", getpid());
+  const int fd = ceph_open(cmount, c_file, O_RDWR | O_CREAT, fileMode);
+  ASSERT_GE(fd, 0); 
+
+  // Lock exclusively twice
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, 42));
+  ASSERT_EQ(-EWOULDBLOCK, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, 43));
+  ASSERT_EQ(-EWOULDBLOCK, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, 44));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, 42));
+
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, 43));
+  ASSERT_EQ(-EWOULDBLOCK, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, 44));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, 43));
+
+  // Lock shared three times
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH, 42));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH, 43));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH, 44));
+  // And then attempt to lock exclusively
+  ASSERT_EQ(-EWOULDBLOCK, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, 45));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, 42));
+  ASSERT_EQ(-EWOULDBLOCK, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, 45));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, 44));
+  ASSERT_EQ(-EWOULDBLOCK, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, 45));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, 43));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, 45));
+  ASSERT_EQ(-EWOULDBLOCK, ceph_flock(cmount, fd, LOCK_SH | LOCK_NB, 42));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, 45));
+
+  // Lock shared with upgrade to exclusive (POSIX) 
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH, 42));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, 42));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, 42));
+
+  // Lock exclusive with downgrade to shared (POSIX) 
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, 42));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH, 42));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, 42));
+
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+  ASSERT_EQ(0, ceph_unlink(cmount, c_file));
+  CLEANUP_CEPH();
+}
+
+/* Locking in different threads */
+
+// Used by ConcurrentLocking test
+struct str_ConcurrentLocking {
+  const char *file;
+  struct ceph_mount_info *cmount;  // !NULL if shared
+  sem_t sem[2];
+  sem_t semReply[2];
+  void sem_init(int pshared) {
+    ASSERT_EQ(0, ::sem_init(&sem[0], pshared, 0));
+    ASSERT_EQ(0, ::sem_init(&sem[1], pshared, 0));
+    ASSERT_EQ(0, ::sem_init(&semReply[0], pshared, 0));
+    ASSERT_EQ(0, ::sem_init(&semReply[1], pshared, 0));
+  }
+  void sem_destroy() {
+    ASSERT_EQ(0, ::sem_destroy(&sem[0]));
+    ASSERT_EQ(0, ::sem_destroy(&sem[1]));
+    ASSERT_EQ(0, ::sem_destroy(&semReply[0]));
+    ASSERT_EQ(0, ::sem_destroy(&semReply[1]));
+  }
+};
+
+// Wakeup main (for (N) steps)
+#define PING_MAIN(n) ASSERT_EQ(0, sem_post(&s.sem[n%2]))
+// Wait for main to wake us up (for (RN) steps)
+#define WAIT_MAIN(n) \
+  ASSERT_EQ(0, sem_timedwait(&s.semReply[n%2], abstime(ts, waitSlowMs)))
+
+// Wakeup worker (for (RN) steps)
+#define PING_WORKER(n) ASSERT_EQ(0, sem_post(&s.semReply[n%2]))
+// Wait for worker to wake us up (for (N) steps)
+#define WAIT_WORKER(n) \
+  ASSERT_EQ(0, sem_timedwait(&s.sem[n%2], abstime(ts, waitSlowMs)))
+// Worker shall not wake us up (for (N) steps)
+#define NOT_WAIT_WORKER(n) \
+  ASSERT_EQ(-1, sem_timedwait(&s.sem[n%2], abstime(ts, waitMs)))
+
+// Do twice an operation
+#define TWICE(EXPR) do {			\
+    EXPR;					\
+    EXPR;					\
+  } while(0)
+
+/* Locking in different threads */
+
+// Used by ConcurrentLocking test
+static void thread_ConcurrentLocking(str_ConcurrentLocking& s) {
+  struct ceph_mount_info *const cmount = s.cmount;
+  struct timespec ts;
+
+  const int fd = ceph_open(cmount, s.file, O_RDWR | O_CREAT, fileMode);
+  ASSERT_GE(fd, 0); 
+
+  ASSERT_EQ(-EWOULDBLOCK,
+	    ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, ceph_pthread_self()));
+  PING_MAIN(1); // (1)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, ceph_pthread_self()));
+  PING_MAIN(2); // (2)
+
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, ceph_pthread_self()));
+  PING_MAIN(3); // (3)
+
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH, ceph_pthread_self()));
+  PING_MAIN(4); // (4)
+
+  WAIT_MAIN(1); // (R1)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, ceph_pthread_self()));
+  PING_MAIN(5); // (5)
+
+  WAIT_MAIN(2); // (R2)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, ceph_pthread_self()));
+  PING_MAIN(6); // (6)
+
+  WAIT_MAIN(3); // (R3)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, ceph_pthread_self()));
+  PING_MAIN(7); // (7)
+}
+
+// Used by ConcurrentLocking test
+static void* thread_ConcurrentLocking_(void *arg) {
+  str_ConcurrentLocking *const s =
+    reinterpret_cast<str_ConcurrentLocking*>(arg);
+  thread_ConcurrentLocking(*s);
+  return NULL;
+}
+
+TEST(LibCephFS, ConcurrentLocking) {
+  const pid_t mypid = getpid();
+  struct ceph_mount_info *cmount;
+  STARTUP_CEPH();
+
+  char c_file[1024];
+  sprintf(c_file, "/flock_test_%d", mypid);
+  const int fd = ceph_open(cmount, c_file, O_RDWR | O_CREAT, fileMode);
+  ASSERT_GE(fd, 0); 
+
+  // Lock
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, ceph_pthread_self()));
+
+  // Start locker thread
+  pthread_t thread;
+  struct timespec ts;
+  str_ConcurrentLocking s = { c_file, cmount };
+  s.sem_init(0);
+  ASSERT_EQ(0, pthread_create(&thread, NULL, thread_ConcurrentLocking_, &s));
+  // Synchronization point with thread (failure: thread is dead)
+  WAIT_WORKER(1); // (1)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(2); // (2)
+
+  // Unlock
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, ceph_pthread_self()));
+
+  // Shall have lock
+  // Synchronization point with thread (failure: thread is dead)
+  WAIT_WORKER(2); // (2)
+
+  // Synchronization point with thread (failure: thread is dead)
+  WAIT_WORKER(3); // (3)
+
+  // Wait for thread to share lock
+  WAIT_WORKER(4); // (4)
+  ASSERT_EQ(-EWOULDBLOCK,
+	    ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, ceph_pthread_self()));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH | LOCK_NB, ceph_pthread_self()));
+
+  // Wake up thread to unlock shared lock
+  PING_WORKER(1); // (R1)
+  WAIT_WORKER(5); // (5)
+
+  // Now we can lock exclusively
+  // Upgrade to exclusive lock (as per POSIX)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, ceph_pthread_self()));
+
+  // Wake up thread to lock shared lock
+  PING_WORKER(2); // (R2)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(6); // (6)
+
+  // Release lock ; thread will get it
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, ceph_pthread_self()));
+  WAIT_WORKER(6); // (6)
+
+  // We no longer have the lock
+  ASSERT_EQ(-EWOULDBLOCK,
+	    ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, ceph_pthread_self()));
+  ASSERT_EQ(-EWOULDBLOCK,
+	    ceph_flock(cmount, fd, LOCK_SH | LOCK_NB, ceph_pthread_self()));
+
+  // Wake up thread to unlock exclusive lock
+  PING_WORKER(3); // (R3)
+  WAIT_WORKER(7); // (7)
+
+  // We can lock it again
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, ceph_pthread_self()));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, ceph_pthread_self()));
+
+  // Cleanup
+  void *retval = (void*) (uintptr_t) -1;
+  ASSERT_EQ(0, pthread_join(thread, &retval));
+  ASSERT_EQ(NULL, retval);
+  s.sem_destroy();
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+  ASSERT_EQ(0, ceph_unlink(cmount, c_file));
+  CLEANUP_CEPH();
+}
+
+TEST(LibCephFS, ThreesomeLocking) {
+  const pid_t mypid = getpid();
+  struct ceph_mount_info *cmount;
+  STARTUP_CEPH();
+
+  char c_file[1024];
+  sprintf(c_file, "/flock_test_%d", mypid);
+  const int fd = ceph_open(cmount, c_file, O_RDWR | O_CREAT, fileMode);
+  ASSERT_GE(fd, 0); 
+
+  // Lock
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, ceph_pthread_self()));
+
+  // Start locker thread
+  pthread_t thread[2];
+  struct timespec ts;
+  str_ConcurrentLocking s = { c_file, cmount };
+  s.sem_init(0);
+  ASSERT_EQ(0, pthread_create(&thread[0], NULL, thread_ConcurrentLocking_, &s));
+  ASSERT_EQ(0, pthread_create(&thread[1], NULL, thread_ConcurrentLocking_, &s));
+  // Synchronization point with thread (failure: thread is dead)
+  TWICE(WAIT_WORKER(1)); // (1)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(2); // (2)
+
+  // Unlock
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, ceph_pthread_self()));
+
+  // Shall have lock
+  TWICE(// Synchronization point with thread (failure: thread is dead)
+	WAIT_WORKER(2); // (2)
+	
+	// Synchronization point with thread (failure: thread is dead)
+	WAIT_WORKER(3)); // (3)
+  
+  // Wait for thread to share lock
+  TWICE(WAIT_WORKER(4)); // (4)
+  ASSERT_EQ(-EWOULDBLOCK,
+	    ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, ceph_pthread_self()));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH | LOCK_NB, ceph_pthread_self()));
+
+  // Wake up thread to unlock shared lock
+  TWICE(PING_WORKER(1); // (R1)
+	WAIT_WORKER(5)); // (5)
+
+  // Now we can lock exclusively
+  // Upgrade to exclusive lock (as per POSIX)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, ceph_pthread_self()));
+
+  TWICE(  // Wake up thread to lock shared lock
+	PING_WORKER(2); // (R2)
+	
+	// Shall not have lock immediately
+	NOT_WAIT_WORKER(6)); // (6)
+  
+  // Release lock ; thread will get it
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, ceph_pthread_self()));
+  TWICE(WAIT_WORKER(6); // (6)
+	
+	// We no longer have the lock
+	ASSERT_EQ(-EWOULDBLOCK,
+		  ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, ceph_pthread_self()));
+	ASSERT_EQ(-EWOULDBLOCK,
+		  ceph_flock(cmount, fd, LOCK_SH | LOCK_NB, ceph_pthread_self()));
+	
+	// Wake up thread to unlock exclusive lock
+	PING_WORKER(3); // (R3)
+	WAIT_WORKER(7); // (7)
+	);
+  
+  // We can lock it again
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, ceph_pthread_self()));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, ceph_pthread_self()));
+
+  // Cleanup
+  void *retval = (void*) (uintptr_t) -1;
+  ASSERT_EQ(0, pthread_join(thread[0], &retval));
+  ASSERT_EQ(NULL, retval);
+  ASSERT_EQ(0, pthread_join(thread[1], &retval));
+  ASSERT_EQ(NULL, retval);
+  s.sem_destroy();
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+  ASSERT_EQ(0, ceph_unlink(cmount, c_file));
+  CLEANUP_CEPH();
+}
+
+/* Locking in different processes */
+
+#define PROCESS_SLOW_MS() \
+  static const long waitMs = 100; \
+  (void) waitMs
+
+// Used by ConcurrentLocking test
+static void process_ConcurrentLocking(str_ConcurrentLocking& s) {
+  const pid_t mypid = getpid();
+  PROCESS_SLOW_MS();
+
+  struct ceph_mount_info *cmount = NULL;
+  struct timespec ts;
+
+  STARTUP_CEPH();
+  s.cmount = cmount;
+
+  const int fd = ceph_open(cmount, s.file, O_RDWR | O_CREAT, fileMode);
+  ASSERT_GE(fd, 0); 
+  WAIT_MAIN(1); // (R1)
+
+  ASSERT_EQ(-EWOULDBLOCK,
+	    ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, mypid));
+  PING_MAIN(1); // (1)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, mypid));
+  PING_MAIN(2); // (2)
+
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, mypid));
+  PING_MAIN(3); // (3)
+
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH, mypid));
+  PING_MAIN(4); // (4)
+
+  WAIT_MAIN(2); // (R2)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, mypid));
+  PING_MAIN(5); // (5)
+
+  WAIT_MAIN(3); // (R3)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, mypid));
+  PING_MAIN(6); // (6)
+
+  WAIT_MAIN(4); // (R4)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, mypid));
+  PING_MAIN(7); // (7)
+
+  CLEANUP_CEPH();
+
+  s.sem_destroy();
+  exit(EXIT_SUCCESS);
+}
+
+// Disabled because of fork() issues (http://tracker.ceph.com/issues/16556)
+TEST(LibCephFS, DISABLED_InterProcessLocking) {
+  PROCESS_SLOW_MS();
+  // Process synchronization
+  char c_file[1024];
+  const pid_t mypid = getpid();
+  sprintf(c_file, "/flock_test_%d", mypid);
+
+  // Note: the semaphores MUST be on a shared memory segment
+  str_ConcurrentLocking *const shs =
+    reinterpret_cast<str_ConcurrentLocking*>
+    (mmap(0, sizeof(*shs), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS,
+	  -1, 0));
+  str_ConcurrentLocking &s = *shs;
+  s.file = c_file;
+  s.sem_init(1);
+
+  // Start locker process
+  const pid_t pid = fork();
+  ASSERT_GE(pid, 0);
+  if (pid == 0) {
+    process_ConcurrentLocking(s);
+    exit(EXIT_FAILURE);
+  }
+
+  struct timespec ts;
+  struct ceph_mount_info *cmount;
+  STARTUP_CEPH();
+
+  const int fd = ceph_open(cmount, c_file, O_RDWR | O_CREAT, fileMode);
+  ASSERT_GE(fd, 0); 
+
+  // Lock
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, mypid));
+
+  // Synchronization point with process (failure: process is dead)
+  PING_WORKER(1); // (R1)
+  WAIT_WORKER(1); // (1)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(2); // (2)
+
+  // Unlock
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, mypid));
+
+  // Shall have lock
+  // Synchronization point with process (failure: process is dead)
+  WAIT_WORKER(2); // (2)
+
+  // Synchronization point with process (failure: process is dead)
+  WAIT_WORKER(3); // (3)
+
+  // Wait for process to share lock
+  WAIT_WORKER(4); // (4)
+  ASSERT_EQ(-EWOULDBLOCK, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, mypid));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH | LOCK_NB, mypid));
+
+  // Wake up process to unlock shared lock
+  PING_WORKER(2); // (R2)
+  WAIT_WORKER(5); // (5)
+
+  // Now we can lock exclusively
+  // Upgrade to exclusive lock (as per POSIX)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, mypid));
+
+  // Wake up process to lock shared lock
+  PING_WORKER(3); // (R3)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(6); // (6)
+
+  // Release lock ; process will get it
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, mypid));
+  WAIT_WORKER(6); // (6)
+
+  // We no longer have the lock
+  ASSERT_EQ(-EWOULDBLOCK, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, mypid));
+  ASSERT_EQ(-EWOULDBLOCK, ceph_flock(cmount, fd, LOCK_SH | LOCK_NB, mypid));
+
+  // Wake up process to unlock exclusive lock
+  PING_WORKER(4); // (R4)
+  WAIT_WORKER(7); // (7)
+
+  // We can lock it again
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, mypid));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, mypid));
+
+  // Wait pid
+  int status;
+  ASSERT_EQ(pid, waitpid(pid, &status, 0));
+  ASSERT_EQ(EXIT_SUCCESS, status);
+
+  // Cleanup
+  s.sem_destroy();
+  ASSERT_EQ(0, munmap(shs, sizeof(*shs)));
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+  ASSERT_EQ(0, ceph_unlink(cmount, c_file));
+  CLEANUP_CEPH();
+}
+
+// Disabled because of fork() issues (http://tracker.ceph.com/issues/16556)
+TEST(LibCephFS, DISABLED_ThreesomeInterProcessLocking) {
+  PROCESS_SLOW_MS();
+  // Process synchronization
+  char c_file[1024];
+  const pid_t mypid = getpid();
+  sprintf(c_file, "/flock_test_%d", mypid);
+
+  // Note: the semaphores MUST be on a shared memory segment
+  str_ConcurrentLocking *const shs =
+    reinterpret_cast<str_ConcurrentLocking*>
+    (mmap(0, sizeof(*shs), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS,
+	  -1, 0));
+  str_ConcurrentLocking &s = *shs;
+  s.file = c_file;
+  s.sem_init(1);
+
+  // Start locker processes
+  pid_t pid[2];
+  pid[0] = fork();
+  ASSERT_GE(pid[0], 0);
+  if (pid[0] == 0) {
+    process_ConcurrentLocking(s);
+    exit(EXIT_FAILURE);
+  }
+  pid[1] = fork();
+  ASSERT_GE(pid[1], 0);
+  if (pid[1] == 0) {
+    process_ConcurrentLocking(s);
+    exit(EXIT_FAILURE);
+  }
+
+  struct timespec ts;
+  struct ceph_mount_info *cmount;
+  STARTUP_CEPH();
+
+  const int fd = ceph_open(cmount, c_file, O_RDWR | O_CREAT, fileMode);
+  ASSERT_GE(fd, 0); 
+
+  // Lock
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, mypid));
+
+  // Synchronization point with process (failure: process is dead)
+  TWICE(PING_WORKER(1)); // (R1)
+  TWICE(WAIT_WORKER(1)); // (1)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(2); // (2)
+
+  // Unlock
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, mypid));
+
+  // Shall have lock
+  TWICE(// Synchronization point with process (failure: process is dead)
+	WAIT_WORKER(2); // (2)
+	
+	// Synchronization point with process (failure: process is dead)
+	WAIT_WORKER(3)); // (3)
+  
+  // Wait for process to share lock
+  TWICE(WAIT_WORKER(4)); // (4)
+  ASSERT_EQ(-EWOULDBLOCK,
+	    ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, mypid));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_SH | LOCK_NB, mypid));
+
+  // Wake up process to unlock shared lock
+  TWICE(PING_WORKER(2); // (R2)
+	WAIT_WORKER(5)); // (5)
+
+  // Now we can lock exclusively
+  // Upgrade to exclusive lock (as per POSIX)
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX, mypid));
+
+  TWICE(  // Wake up process to lock shared lock
+	PING_WORKER(3); // (R3)
+	
+	// Shall not have lock immediately
+	NOT_WAIT_WORKER(6)); // (6)
+  
+  // Release lock ; process will get it
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, mypid));
+  TWICE(WAIT_WORKER(6); // (6)
+	
+	// We no longer have the lock
+	ASSERT_EQ(-EWOULDBLOCK,
+		  ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, mypid));
+	ASSERT_EQ(-EWOULDBLOCK,
+		  ceph_flock(cmount, fd, LOCK_SH | LOCK_NB, mypid));
+	
+	// Wake up process to unlock exclusive lock
+	PING_WORKER(4); // (R4)
+	WAIT_WORKER(7); // (7)
+	);
+  
+  // We can lock it again
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_EX | LOCK_NB, mypid));
+  ASSERT_EQ(0, ceph_flock(cmount, fd, LOCK_UN, mypid));
+
+  // Wait pids
+  int status;
+  ASSERT_EQ(pid[0], waitpid(pid[0], &status, 0));
+  ASSERT_EQ(EXIT_SUCCESS, status);
+  ASSERT_EQ(pid[1], waitpid(pid[1], &status, 0));
+  ASSERT_EQ(EXIT_SUCCESS, status);
+
+  // Cleanup
+  s.sem_destroy();
+  ASSERT_EQ(0, munmap(shs, sizeof(*shs)));
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+  ASSERT_EQ(0, ceph_unlink(cmount, c_file));
+  CLEANUP_CEPH();
+}
diff --git a/src/test/libcephfs/lazyio.cc b/src/test/libcephfs/lazyio.cc
new file mode 100644
index 00000000..b4847304
--- /dev/null
+++ b/src/test/libcephfs/lazyio.cc
@@ -0,0 +1,355 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2019 Red Hat Ltd
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software 
+ * Foundation.  See file COPYING.
+ * 
+ */
+
+#include "gtest/gtest.h"
+#include "include/cephfs/libcephfs.h"
+#include "include/rados/librados.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <dirent.h>
+#if defined(__linux__)
+#include <sys/xattr.h>
+#endif
+
+rados_t cluster;
+
+TEST(LibCephFS, LazyIOOneWriterMulipleReaders) {
+  struct ceph_mount_info *ca, *cb;
+  ASSERT_EQ(ceph_create(&ca, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(ca, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(ca, NULL));
+  ASSERT_EQ(ceph_mount(ca, NULL), 0);
+
+  ASSERT_EQ(ceph_create(&cb, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cb, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cb, NULL));
+  ASSERT_EQ(ceph_mount(cb, NULL), 0);
+
+  char name[20];
+  snprintf(name, sizeof(name), "foo.%d", getpid());
+
+  int fda = ceph_open(ca, name, O_CREAT|O_RDWR, 0644);
+  ASSERT_LE(0, fda);
+
+  int fdb = ceph_open(cb, name, O_RDONLY, 0644);
+  ASSERT_LE(0, fdb);
+
+  ASSERT_EQ(0, ceph_lazyio(ca, fda, 1));
+  ASSERT_EQ(0, ceph_lazyio(cb, fdb, 1));
+ 
+  char out_buf[] = "fooooooooo";
+
+  /* Client a issues a write and propagates/flushes the buffer */
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(ca, fda, out_buf, sizeof(out_buf), 0));
+  ASSERT_EQ(0, ceph_lazyio_propagate(ca, fda, 0, 0));
+
+  /* Client a issues a write and propagates/flushes the buffer */
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(ca, fda, out_buf, sizeof(out_buf), 10));
+  ASSERT_EQ(0, ceph_lazyio_propagate(ca, fda, 0, 0));
+
+  char in_buf[40];
+  /* Calling ceph_lazyio_synchronize here will invalidate client b's cache and hence enable client a to fetch the propagated write of client a in the subsequent read */
+  ASSERT_EQ(0, ceph_lazyio_synchronize(cb, fdb, 0, 0));
+  ASSERT_EQ(ceph_read(cb, fdb, in_buf, sizeof(in_buf), 0), 2*strlen(out_buf)+1);
+  ASSERT_STREQ(in_buf, "fooooooooofooooooooo");
+
+  /* Client a does not need to call ceph_lazyio_synchronize here because it is the latest writer and fda holds the updated inode*/
+  ASSERT_EQ(ceph_read(ca, fda, in_buf, sizeof(in_buf), 0), 2*strlen(out_buf)+1);
+  ASSERT_STREQ(in_buf, "fooooooooofooooooooo");
+
+  ceph_close(ca, fda);
+  ceph_close(cb, fdb);
+
+  ceph_shutdown(ca);
+  ceph_shutdown(cb);
+}
+
+TEST(LibCephFS, LazyIOMultipleWritersMulipleReaders) {
+  struct ceph_mount_info *ca, *cb;
+  ASSERT_EQ(ceph_create(&ca, NULL), 0); 
+  ASSERT_EQ(ceph_conf_read_file(ca, NULL), 0); 
+  ASSERT_EQ(0, ceph_conf_parse_env(ca, NULL));
+  ASSERT_EQ(ceph_mount(ca, NULL), 0); 
+
+  ASSERT_EQ(ceph_create(&cb, NULL), 0); 
+  ASSERT_EQ(ceph_conf_read_file(cb, NULL), 0); 
+  ASSERT_EQ(0, ceph_conf_parse_env(cb, NULL));
+  ASSERT_EQ(ceph_mount(cb, NULL), 0); 
+
+  char name[20];
+  snprintf(name, sizeof(name), "foo2.%d", getpid());
+
+  int fda = ceph_open(ca, name, O_CREAT|O_RDWR, 0644);
+  ASSERT_LE(0, fda);
+
+  int fdb = ceph_open(cb, name, O_RDWR, 0644);
+  ASSERT_LE(0, fdb);
+
+  ASSERT_EQ(0, ceph_lazyio(ca, fda, 1));
+  ASSERT_EQ(0, ceph_lazyio(cb, fdb, 1));
+
+  char out_buf[] = "fooooooooo";
+  /* Client a issues a write and propagates/flushes the buffer */
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(ca, fda, out_buf, sizeof(out_buf), 0));
+  ASSERT_EQ(0, ceph_lazyio_propagate(ca, fda, 0, 0));
+  
+  /* Client b issues a write and propagates/flushes the buffer*/
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(cb, fdb, out_buf, sizeof(out_buf), 10));
+  ASSERT_EQ(0, ceph_lazyio_propagate(cb, fdb, 0, 0));
+
+  char in_buf[40];
+  /* Calling ceph_lazyio_synchronize here will invalidate client a's cache and hence enable client a to fetch the propagated writes of client b in the subsequent read */
+  ASSERT_EQ(0, ceph_lazyio_synchronize(ca, fda, 0, 0));
+  ASSERT_EQ(ceph_read(ca, fda, in_buf, sizeof(in_buf), 0), 2*strlen(out_buf)+1);
+  ASSERT_STREQ(in_buf, "fooooooooofooooooooo");
+  
+  /* Client b does not need to call ceph_lazyio_synchronize here because it is the latest writer and the writes before it have already been propagated*/
+  ASSERT_EQ(ceph_read(cb, fdb, in_buf, sizeof(in_buf), 0), 2*strlen(out_buf)+1);
+  ASSERT_STREQ(in_buf, "fooooooooofooooooooo");
+
+  /* Client a issues a write */
+  char wait_out_buf[] = "foobarbars";
+  ASSERT_EQ((int)sizeof(wait_out_buf), ceph_write(ca, fda, wait_out_buf, sizeof(wait_out_buf), 20));
+  ASSERT_EQ(0, ceph_lazyio_propagate(ca, fda, 0, 0));
+
+  /* Client a does not need to call ceph_lazyio_synchronize here because it is the latest writer and the writes before it have already been propagated*/
+  ASSERT_EQ(ceph_read(ca, fda, in_buf, sizeof(in_buf), 0), (2*(strlen(out_buf)))+strlen(wait_out_buf)+1);
+  ASSERT_STREQ(in_buf, "fooooooooofooooooooofoobarbars");
+
+  /* Calling ceph_lazyio_synchronize here will invalidate client b's cache and hence enable client a to fetch the propagated write of client a in the subsequent read */
+  ASSERT_EQ(0, ceph_lazyio_synchronize(cb, fdb, 0, 0));
+  ASSERT_EQ(ceph_read(cb, fdb, in_buf, sizeof(in_buf), 0), (2*(strlen(out_buf)))+strlen(wait_out_buf)+1);
+  ASSERT_STREQ(in_buf, "fooooooooofooooooooofoobarbars");
+
+  ceph_close(ca, fda);
+  ceph_close(cb, fdb);
+
+  ceph_shutdown(ca);
+  ceph_shutdown(cb);
+}
+
+TEST(LibCephFS, LazyIOMultipleWritersOneReader) {
+  struct ceph_mount_info *ca, *cb;
+  ASSERT_EQ(ceph_create(&ca, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(ca, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(ca, NULL));
+  ASSERT_EQ(ceph_mount(ca, NULL), 0);
+
+  ASSERT_EQ(ceph_create(&cb, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cb, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cb, NULL));
+  ASSERT_EQ(ceph_mount(cb, NULL), 0);
+
+  char name[20];
+  snprintf(name, sizeof(name), "foo3.%d", getpid());
+
+  int fda = ceph_open(ca, name, O_CREAT|O_RDWR, 0644);
+  ASSERT_LE(0, fda);
+
+  int fdb = ceph_open(cb, name, O_RDWR, 0644);
+  ASSERT_LE(0, fdb);
+ 
+  ASSERT_EQ(0, ceph_lazyio(ca, fda, 1));
+  ASSERT_EQ(0, ceph_lazyio(cb, fdb, 1));
+
+  char out_buf[] = "fooooooooo";
+  /* Client a issues a write and propagates/flushes the buffer */
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(ca, fda, out_buf, sizeof(out_buf), 0));
+  ASSERT_EQ(0, ceph_lazyio_propagate(ca, fda, 0, 0));
+  
+  /* Client b issues a write and propagates/flushes the buffer*/
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(cb, fdb, out_buf, sizeof(out_buf), 10));
+  ASSERT_EQ(0, ceph_lazyio_propagate(cb, fdb, 0, 0));
+
+  char in_buf[40];
+  /* Client a reads the file and verifies that it only reads it's propagated writes and not Client b's*/
+  ASSERT_EQ(ceph_read(ca, fda, in_buf, sizeof(in_buf), 0), strlen(out_buf)+1);
+  ASSERT_STREQ(in_buf, "fooooooooo");
+  
+  /* Client a reads the file again, this time with a lazyio_synchronize to check if the cache gets invalidated and data is refetched i.e all the propagated writes are being read*/
+  ASSERT_EQ(0, ceph_lazyio_synchronize(ca, fda, 0, 0));
+  ASSERT_EQ(ceph_read(ca, fda, in_buf, sizeof(in_buf), 0), 2*strlen(out_buf)+1);
+  ASSERT_STREQ(in_buf, "fooooooooofooooooooo");
+
+  ceph_close(ca, fda);
+  ceph_close(cb, fdb);
+
+  ceph_shutdown(ca);
+  ceph_shutdown(cb);
+}
+
+TEST(LibCephFS, LazyIOSynchronizeFlush) {
+  /* Test to make sure lazyio_synchronize flushes dirty buffers */
+  struct ceph_mount_info *ca, *cb;
+  ASSERT_EQ(ceph_create(&ca, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(ca, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(ca, NULL));
+  ASSERT_EQ(ceph_mount(ca, NULL), 0);
+
+  ASSERT_EQ(ceph_create(&cb, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cb, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cb, NULL));
+  ASSERT_EQ(ceph_mount(cb, NULL), 0);
+
+  char name[20];
+  snprintf(name, sizeof(name), "foo4.%d", getpid());
+
+  int fda = ceph_open(ca, name, O_CREAT|O_RDWR, 0644);
+  ASSERT_LE(0, fda);
+
+  int fdb = ceph_open(cb, name, O_RDWR, 0644);
+  ASSERT_LE(0, fdb);
+
+  ASSERT_EQ(0, ceph_lazyio(ca, fda, 1));
+  ASSERT_EQ(0, ceph_lazyio(cb, fdb, 1));
+
+  char out_buf[] = "fooooooooo";
+
+  /* Client a issues a write and propagates it*/
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(ca, fda, out_buf, sizeof(out_buf), 0));
+  ASSERT_EQ(0, ceph_lazyio_propagate(ca, fda, 0, 0));
+
+  /* Client b issues writes and without lazyio_propagate*/
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(cb, fdb, out_buf, sizeof(out_buf), 10));
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(cb, fdb, out_buf, sizeof(out_buf), 20));
+  
+  char in_buf[40];
+  /* Calling ceph_lazyio_synchronize here will first flush the possibly pending buffered write of client b and invalidate client b's cache and hence enable client b to fetch all the propagated writes */
+  ASSERT_EQ(0, ceph_lazyio_synchronize(cb, fdb, 0, 0));
+  ASSERT_EQ(ceph_read(cb, fdb, in_buf, sizeof(in_buf), 0), 3*strlen(out_buf)+1);
+  ASSERT_STREQ(in_buf, "fooooooooofooooooooofooooooooo");
+
+  /* Required to call ceph_lazyio_synchronize here since client b is the latest writer and client a is out of sync with updated file*/ 
+  ASSERT_EQ(0, ceph_lazyio_synchronize(ca, fda, 0, 0));
+  ASSERT_EQ(ceph_read(ca, fda, in_buf, sizeof(in_buf), 0), 3*strlen(out_buf)+1);
+  ASSERT_STREQ(in_buf, "fooooooooofooooooooofooooooooo");
+
+  ceph_close(ca, fda);
+  ceph_close(cb, fdb);
+
+  ceph_shutdown(ca);
+  ceph_shutdown(cb);
+}
+
+TEST(LibCephFS, WithoutandWithLazyIO) {
+  struct ceph_mount_info *ca, *cb;
+  ASSERT_EQ(ceph_create(&ca, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(ca, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(ca, NULL));
+  ASSERT_EQ(ceph_mount(ca, NULL), 0);
+
+  ASSERT_EQ(ceph_create(&cb, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cb, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cb, NULL));
+  ASSERT_EQ(ceph_mount(cb, NULL), 0);
+
+  char name[20];
+  snprintf(name, sizeof(name), "foo5.%d", getpid());
+
+  int fda = ceph_open(ca, name, O_CREAT|O_RDWR, 0644);
+  ASSERT_LE(0, fda);
+
+  int fdb = ceph_open(cb, name, O_RDWR, 0644);
+  ASSERT_LE(0, fdb);
+
+  char out_buf_w[] = "1234567890";
+  /* Doing some non lazyio writes and read*/
+  ASSERT_EQ((int)sizeof(out_buf_w), ceph_write(ca, fda, out_buf_w, sizeof(out_buf_w), 0));
+
+  ASSERT_EQ((int)sizeof(out_buf_w), ceph_write(cb, fdb, out_buf_w, sizeof(out_buf_w), 10));
+
+  char in_buf_w[30];
+  ASSERT_EQ(ceph_read(ca, fda, in_buf_w, sizeof(in_buf_w), 0), 2*strlen(out_buf_w)+1);
+
+  /* Enable lazyio*/
+  ASSERT_EQ(0, ceph_lazyio(ca, fda, 1));
+  ASSERT_EQ(0, ceph_lazyio(cb, fdb, 1));
+
+  char out_buf[] = "fooooooooo";
+
+  /* Client a issues a write and propagates/flushes the buffer*/
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(ca, fda, out_buf, sizeof(out_buf), 20));
+  ASSERT_EQ(0, ceph_lazyio_propagate(ca, fda, 0, 0));
+  
+  /* Client b issues a write and propagates/flushes the buffer*/
+  ASSERT_EQ((int)sizeof(out_buf), ceph_write(cb, fdb, out_buf, sizeof(out_buf), 30));
+  ASSERT_EQ(0, ceph_lazyio_propagate(cb, fdb, 0, 0));
+
+  char in_buf[50];
+  /* Calling ceph_lazyio_synchronize here will invalidate client a's cache and hence enable client a to fetch the propagated writes of client b in the subsequent read */
+  ASSERT_EQ(0, ceph_lazyio_synchronize(ca, fda, 0, 0));
+  ASSERT_EQ(ceph_read(ca, fda, in_buf, sizeof(in_buf), 0), (2*(strlen(out_buf)))+(2*(strlen(out_buf_w)))+1);
+  ASSERT_STREQ(in_buf, "12345678901234567890fooooooooofooooooooo");
+
+  /* Client b does not need to call ceph_lazyio_synchronize here because it is the latest writer and the writes before it have already been propagated*/
+  ASSERT_EQ(ceph_read(cb, fdb, in_buf, sizeof(in_buf), 0), (2*(strlen(out_buf)))+(2*(strlen(out_buf_w)))+1);
+  ASSERT_STREQ(in_buf, "12345678901234567890fooooooooofooooooooo");
+
+  ceph_close(ca, fda);
+  ceph_close(cb, fdb);
+
+  ceph_shutdown(ca);
+  ceph_shutdown(cb);
+}
+
+static int update_root_mode()
+{
+  struct ceph_mount_info *admin;
+  int r = ceph_create(&admin, NULL);
+  if (r < 0)
+    return r;
+  ceph_conf_read_file(admin, NULL);
+  ceph_conf_parse_env(admin, NULL);
+  ceph_conf_set(admin, "client_permissions", "false");
+  r = ceph_mount(admin, "/");
+  if (r < 0)
+    goto out;
+  r = ceph_chmod(admin, "/", 0777);
+out:
+  ceph_shutdown(admin);
+  return r;
+}
+
+int main(int argc, char **argv)
+{
+  int r = update_root_mode();
+  if (r < 0)
+    exit(1);
+
+  ::testing::InitGoogleTest(&argc, argv);
+
+  srand(getpid());
+
+  r = rados_create(&cluster, NULL);
+  if (r < 0)
+    exit(1);
+
+  r = rados_conf_read_file(cluster, NULL);
+  if (r < 0)
+    exit(1);
+
+  rados_conf_parse_env(cluster, NULL);
+  r = rados_connect(cluster);
+  if (r < 0)
+    exit(1);
+
+  r = RUN_ALL_TESTS();
+
+  rados_shutdown(cluster);
+
+  return r;
+}
diff --git a/src/test/libcephfs/main.cc b/src/test/libcephfs/main.cc
new file mode 100644
index 00000000..99a72b2a
--- /dev/null
+++ b/src/test/libcephfs/main.cc
@@ -0,0 +1,49 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 New Dream Network
+ * Copyright (C) 2016 Red Hat
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "gtest/gtest.h"
+#include "include/cephfs/libcephfs.h"
+
+static int update_root_mode()
+{
+  struct ceph_mount_info *admin;
+  int r = ceph_create(&admin, NULL);
+  if (r < 0)
+    return r;
+  ceph_conf_read_file(admin, NULL);
+  ceph_conf_parse_env(admin, NULL);
+  ceph_conf_set(admin, "client_permissions", "false");
+  r = ceph_mount(admin, "/");
+  if (r < 0)
+    goto out;
+  r = ceph_chmod(admin, "/", 01777);
+out:
+  ceph_shutdown(admin);
+  return r;
+}
+
+
+int main(int argc, char **argv)
+{
+  int r = update_root_mode();
+  if (r < 0)
+    exit(1);
+
+  ::testing::InitGoogleTest(&argc, argv);
+
+  srand(getpid());
+
+  return RUN_ALL_TESTS();
+}
diff --git a/src/test/libcephfs/monconfig.cc b/src/test/libcephfs/monconfig.cc
new file mode 100644
index 00000000..c962c4f3
--- /dev/null
+++ b/src/test/libcephfs/monconfig.cc
@@ -0,0 +1,99 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2020 Red Hat
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "gtest/gtest.h"
+#include "include/cephfs/libcephfs.h"
+#include "common/ceph_context.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+
+class MonConfig : public ::testing::Test
+{
+  protected:
+    struct ceph_mount_info *ca;
+
+    void SetUp() override {
+      ASSERT_EQ(0, ceph_create(&ca, NULL));
+      ASSERT_EQ(0, ceph_conf_read_file(ca, NULL));
+      ASSERT_EQ(0, ceph_conf_parse_env(ca, NULL));
+    }
+
+    void TearDown() override {
+      ceph_shutdown(ca);
+    }
+
+    // Helper to remove/unset all possible mon information from ConfigProxy
+    void clear_mon_config(CephContext *cct) {
+      auto& conf = cct->_conf;
+      // Clear safe_to_start_threads, allowing updates to config values
+      conf._clear_safe_to_start_threads();
+      ASSERT_EQ(0, conf.set_val("monmap", "", nullptr));
+      ASSERT_EQ(0, conf.set_val("mon_host", "", nullptr));
+      ASSERT_EQ(0, conf.set_val("mon_dns_srv_name", "", nullptr));
+      conf.set_safe_to_start_threads();
+    }
+
+    // Helper to test basic operation on a mount
+    void use_mount(struct ceph_mount_info *mnt, string name_prefix) {
+      char name[20];
+      snprintf(name, sizeof(name), "%s.%d", name_prefix.c_str(), getpid());
+      int fd = ceph_open(mnt, name, O_CREAT|O_RDWR, 0644);
+      ASSERT_LE(0, fd);
+
+      ceph_close(mnt, fd);
+    }
+};
+
+TEST_F(MonConfig, MonAddrsMissing) {
+  CephContext *cct;
+
+  // Test mount failure when there is no known mon config source
+  cct = ceph_get_mount_context(ca);
+  ASSERT_NE(nullptr, cct);
+  clear_mon_config(cct);
+
+  ASSERT_EQ(-ENOENT, ceph_mount(ca, NULL));
+}
+
+TEST_F(MonConfig, MonAddrsInConfigProxy) {
+  // Test a successful mount with default mon config source in ConfigProxy
+  ASSERT_EQ(0, ceph_mount(ca, NULL));
+
+  use_mount(ca, "foo");
+}
+
+TEST_F(MonConfig, MonAddrsInCct) {
+  struct ceph_mount_info *cb;
+  CephContext *cct;
+
+  // Perform mount to bootstrap mon addrs in CephContext
+  ASSERT_EQ(0, ceph_mount(ca, NULL));
+
+  // Reuse bootstrapped CephContext, clearing ConfigProxy mon addr sources
+  cct = ceph_get_mount_context(ca);
+  ASSERT_NE(nullptr, cct);
+  clear_mon_config(cct);
+  ASSERT_EQ(0, ceph_create_with_context(&cb, cct));
+
+  // Test a successful mount with only mon values in CephContext
+  ASSERT_EQ(0, ceph_mount(cb, NULL));
+
+  use_mount(ca, "bar");
+  use_mount(cb, "bar");
+
+  ceph_shutdown(cb);
+}
diff --git a/src/test/libcephfs/multiclient.cc b/src/test/libcephfs/multiclient.cc
new file mode 100644
index 00000000..465bbd8f
--- /dev/null
+++ b/src/test/libcephfs/multiclient.cc
@@ -0,0 +1,99 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 New Dream Network
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "gtest/gtest.h"
+#include "include/cephfs/libcephfs.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <dirent.h>
+#ifdef __linux__
+#include <sys/xattr.h>
+#endif
+
+TEST(LibCephFS, MulticlientSimple) {
+  struct ceph_mount_info *ca, *cb;
+  ASSERT_EQ(ceph_create(&ca, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(ca, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(ca, NULL));
+  ASSERT_EQ(ceph_mount(ca, NULL), 0);
+
+  ASSERT_EQ(ceph_create(&cb, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cb, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cb, NULL));
+  ASSERT_EQ(ceph_mount(cb, NULL), 0);
+
+  char name[20];
+  snprintf(name, sizeof(name), "foo.%d", getpid());
+  int fda = ceph_open(ca, name, O_CREAT|O_RDWR, 0644);
+  ASSERT_LE(0, fda);
+  int fdb = ceph_open(cb, name, O_CREAT|O_RDWR, 0644);
+  ASSERT_LE(0, fdb);
+
+  char bufa[4] = "foo";
+  char bufb[4];
+
+  for (int i=0; i<10; i++) {
+    strcpy(bufa, "foo");
+    ASSERT_EQ((int)sizeof(bufa), ceph_write(ca, fda, bufa, sizeof(bufa), i*6));
+    ASSERT_EQ((int)sizeof(bufa), ceph_read(cb, fdb, bufb, sizeof(bufa), i*6));
+    ASSERT_EQ(0, memcmp(bufa, bufb, sizeof(bufa)));
+    strcpy(bufb, "bar");
+    ASSERT_EQ((int)sizeof(bufb), ceph_write(cb, fdb, bufb, sizeof(bufb), i*6+3));
+    ASSERT_EQ((int)sizeof(bufb), ceph_read(ca, fda, bufa, sizeof(bufb), i*6+3));
+    ASSERT_EQ(0, memcmp(bufa, bufb, sizeof(bufa)));
+  }
+
+  ceph_close(ca, fda);
+  ceph_close(cb, fdb);
+
+  ceph_shutdown(ca);
+  ceph_shutdown(cb);
+}
+
+TEST(LibCephFS, MulticlientHoleEOF) {
+  struct ceph_mount_info *ca, *cb;
+  ASSERT_EQ(ceph_create(&ca, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(ca, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(ca, NULL));
+  ASSERT_EQ(ceph_mount(ca, NULL), 0);
+
+  ASSERT_EQ(ceph_create(&cb, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cb, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cb, NULL));
+  ASSERT_EQ(ceph_mount(cb, NULL), 0);
+
+  char name[20];
+  snprintf(name, sizeof(name), "foo.%d", getpid());
+  int fda = ceph_open(ca, name, O_CREAT|O_RDWR, 0644);
+  ASSERT_LE(0, fda);
+  int fdb = ceph_open(cb, name, O_CREAT|O_RDWR, 0644);
+  ASSERT_LE(0, fdb);
+
+  ASSERT_EQ(3, ceph_write(ca, fda, "foo", 3, 0));
+  ASSERT_EQ(0, ceph_ftruncate(ca, fda, 1000000));
+
+  char buf[4];
+  ASSERT_EQ(2, ceph_read(cb, fdb, buf, sizeof(buf), 1000000-2));
+  ASSERT_EQ(0, buf[0]);
+  ASSERT_EQ(0, buf[1]);
+
+  ceph_close(ca, fda);
+  ceph_close(cb, fdb);
+
+  ceph_shutdown(ca);
+  ceph_shutdown(cb);
+}
diff --git a/src/test/libcephfs/readdir_r_cb.cc b/src/test/libcephfs/readdir_r_cb.cc
new file mode 100644
index 00000000..b8556947
--- /dev/null
+++ b/src/test/libcephfs/readdir_r_cb.cc
@@ -0,0 +1,63 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 New Dream Network
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "gtest/gtest.h"
+#include "include/cephfs/libcephfs.h"
+#include <errno.h>
+#include <fcntl.h>
+
+TEST(LibCephFS, ReaddirRCB) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  char c_dir[256];
+  sprintf(c_dir, "/readdir_r_cb_tests_%d", getpid());
+  struct ceph_dir_result *dirp;
+  ASSERT_EQ(0, ceph_mkdirs(cmount, c_dir, 0777));
+  ASSERT_LE(0, ceph_opendir(cmount, c_dir, &dirp));
+
+  // dir is empty, check that it only contains . and ..
+  int buflen = 100;
+  char *buf = new char[buflen];
+  // . is 2, .. is 3 (for null terminators)
+  ASSERT_EQ(5, ceph_getdnames(cmount, dirp, buf, buflen));
+  char c_file[256];
+  sprintf(c_file, "/readdir_r_cb_tests_%d/foo", getpid());
+  int fd = ceph_open(cmount, c_file, O_CREAT, 0777);
+  ASSERT_LT(0, fd);
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+
+  // check correctness with one entry
+  ASSERT_LE(0, ceph_closedir(cmount, dirp));
+  ASSERT_LE(0, ceph_opendir(cmount, c_dir, &dirp));
+  ASSERT_EQ(9, ceph_getdnames(cmount, dirp, buf, buflen)); // ., .., foo
+
+  // check correctness if buffer is too small
+  ASSERT_LE(0, ceph_closedir(cmount, dirp));
+  ASSERT_GE(0, ceph_opendir(cmount, c_dir, &dirp));
+  ASSERT_EQ(-ERANGE, ceph_getdnames(cmount, dirp, buf, 1));
+
+  //check correctness if it needs to split listing
+  ASSERT_LE(0, ceph_closedir(cmount, dirp));
+  ASSERT_LE(0, ceph_opendir(cmount, c_dir, &dirp));
+  ASSERT_EQ(5, ceph_getdnames(cmount, dirp, buf, 6));
+  ASSERT_EQ(4, ceph_getdnames(cmount, dirp, buf, 6));
+
+  // free cmount after finishing testing
+  ASSERT_LE(0, ceph_closedir(cmount, dirp));
+  ASSERT_EQ(0, ceph_unmount(cmount));
+  ASSERT_EQ(0, ceph_release(cmount));
+}
diff --git a/src/test/libcephfs/reclaim.cc b/src/test/libcephfs/reclaim.cc
new file mode 100644
index 00000000..7b2f7b76
--- /dev/null
+++ b/src/test/libcephfs/reclaim.cc
@@ -0,0 +1,161 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Tests for Ceph delegation handling
+ *
+ * (c) 2017, Jeff Layton <jlayton@redhat.com>
+ */
+
+#include "gtest/gtest.h"
+#include "include/cephfs/libcephfs.h"
+#include "include/stat.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <dirent.h>
+#include <sys/uio.h>
+#include <libgen.h>
+#include <stdlib.h>
+
+#ifdef __linux__
+#include <sys/xattr.h>
+#include <limits.h>
+#endif
+
+#ifdef __FreeBSD__
+#include <sys/wait.h>
+#endif
+
+
+#include <map>
+#include <vector>
+#include <thread>
+#include <atomic>
+
+#define	CEPHFS_RECLAIM_TIMEOUT		60
+
+static int dying_client(int argc, char **argv)
+{
+  struct ceph_mount_info *cmount;
+
+  /* Caller must pass in the uuid */
+  if (argc < 2)
+    return 1;
+
+  if (ceph_create(&cmount, nullptr) != 0)
+    return 1;
+
+  if (ceph_conf_read_file(cmount, nullptr) != 0)
+    return 1;
+
+  if (ceph_conf_parse_env(cmount, nullptr) != 0)
+    return 1;
+
+  if (ceph_init(cmount) != 0)
+    return 1;
+
+  ceph_set_session_timeout(cmount, CEPHFS_RECLAIM_TIMEOUT);
+
+  if (ceph_start_reclaim(cmount, argv[1], CEPH_RECLAIM_RESET) != -ENOENT)
+    return 1;
+
+  ceph_set_uuid(cmount, argv[1]);
+
+  if (ceph_mount(cmount, "/") != 0)
+    return 1;
+
+  Inode *root, *file;
+  if (ceph_ll_lookup_root(cmount, &root) != 0)
+    return 1;
+
+  Fh *fh;
+  struct ceph_statx stx;
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  if (ceph_ll_create(cmount, root, argv[1], 0666, O_RDWR|O_CREAT|O_EXCL,
+		      &file, &fh, &stx, 0, 0, perms) != 0)
+    return 1;
+
+  return 0;
+}
+
+TEST(LibCephFS, ReclaimReset) {
+  pid_t		pid;
+  char		uuid[256];
+  const char	*exe = "/proc/self/exe";
+
+  sprintf(uuid, "simplereclaim:%x", getpid());
+
+  pid = fork();
+  ASSERT_GE(pid, 0);
+  if (pid == 0) {
+    errno = 0;
+    execl(exe, exe, uuid, nullptr);
+    /* It won't be zero of course, which is sort of the point... */
+    ASSERT_EQ(errno, 0);
+  }
+
+  /* parent - wait for child to exit */
+  int ret;
+  pid_t wp = wait(&ret);
+  ASSERT_GE(wp, 0);
+  ASSERT_EQ(WIFEXITED(ret), true);
+  ASSERT_EQ(WEXITSTATUS(ret), 0);
+
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, nullptr), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, nullptr), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, nullptr));
+  ASSERT_EQ(ceph_init(cmount), 0);
+  ceph_set_session_timeout(cmount, CEPHFS_RECLAIM_TIMEOUT);
+  ASSERT_EQ(ceph_start_reclaim(cmount, uuid, CEPH_RECLAIM_RESET), 0);
+  ceph_set_uuid(cmount, uuid);
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  Inode *root, *file;
+  ASSERT_EQ(ceph_ll_lookup_root(cmount, &root), 0);
+  UserPerm *perms = ceph_mount_perms(cmount);
+  struct ceph_statx stx;
+  ASSERT_EQ(ceph_ll_lookup(cmount, root, uuid, &file, &stx, 0, 0, perms), 0);
+  Fh *fh;
+  ASSERT_EQ(ceph_ll_open(cmount, file, O_WRONLY, &fh, perms), 0);
+
+  ceph_unmount(cmount);
+  ceph_release(cmount);
+}
+
+static int update_root_mode()
+{
+  struct ceph_mount_info *admin;
+  int r = ceph_create(&admin, nullptr);
+  if (r < 0)
+    return r;
+  ceph_conf_read_file(admin, nullptr);
+  ceph_conf_parse_env(admin, nullptr);
+  ceph_conf_set(admin, "client_permissions", "false");
+  r = ceph_mount(admin, "/");
+  if (r < 0)
+    goto out;
+  r = ceph_chmod(admin, "/", 01777);
+out:
+  ceph_shutdown(admin);
+  return r;
+}
+
+int main(int argc, char **argv)
+{
+  int r = update_root_mode();
+  if (r < 0)
+    exit(1);
+
+  ::testing::InitGoogleTest(&argc, argv);
+
+  if (argc > 1)
+    return dying_client(argc, argv);
+
+  srand(getpid());
+
+  return RUN_ALL_TESTS();
+}
diff --git a/src/test/libcephfs/recordlock.cc b/src/test/libcephfs/recordlock.cc
new file mode 100644
index 00000000..efcfc6aa
--- /dev/null
+++ b/src/test/libcephfs/recordlock.cc
@@ -0,0 +1,1096 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 New Dream Network
+ *               2016 Red Hat
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include <pthread.h>
+#include "gtest/gtest.h"
+#ifndef GTEST_IS_THREADSAFE
+#error "!GTEST_IS_THREADSAFE"
+#endif
+
+#include "include/cephfs/libcephfs.h"
+#include <errno.h>
+#include <sys/fcntl.h>
+#include <unistd.h>
+#include <sys/file.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <dirent.h>
+
+#include <stdlib.h>
+#include <semaphore.h>
+#include <time.h>
+#include <sys/mman.h>
+
+#ifdef __linux__
+#include <limits.h>
+#include <sys/xattr.h>
+#elif __FreeBSD__
+#include <sys/types.h>
+#include <sys/wait.h>
+#endif
+
+#include "include/ceph_assert.h"
+#include "ceph_pthread_self.h"
+
+// Startup common: create and mount ceph fs
+#define STARTUP_CEPH() do {				\
+    ASSERT_EQ(0, ceph_create(&cmount, NULL));		\
+    ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));	\
+    ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));	\
+    ASSERT_EQ(0, ceph_mount(cmount, NULL));		\
+  } while(0)
+
+// Cleanup common: unmount and release ceph fs
+#define CLEANUP_CEPH() do {			\
+    ASSERT_EQ(0, ceph_unmount(cmount));		\
+    ASSERT_EQ(0, ceph_release(cmount));		\
+  } while(0)
+
+static const mode_t fileMode = S_IRWXU | S_IRWXG | S_IRWXO;
+
+// Default wait time for normal and "slow" operations
+// (5" should be enough in case of network congestion)
+static const long waitMs = 10;
+static const long waitSlowMs = 5000;
+
+// Get the absolute struct timespec reference from now + 'ms' milliseconds
+static const struct timespec* abstime(struct timespec &ts, long ms) {
+  if (clock_gettime(CLOCK_REALTIME, &ts) == -1) {
+    ceph_abort();
+  }
+  ts.tv_nsec += ms * 1000000;
+  ts.tv_sec += ts.tv_nsec / 1000000000;
+  ts.tv_nsec %= 1000000000;
+  return &ts;
+}
+
+/* Basic locking */
+
+TEST(LibCephFS, BasicRecordLocking) {
+  struct ceph_mount_info *cmount = NULL;
+  STARTUP_CEPH();
+
+  char c_file[1024];
+  sprintf(c_file, "recordlock_test_%d", getpid());
+  Fh *fh = NULL;
+  Inode *root = NULL, *inode = NULL;
+  struct ceph_statx stx;
+  int rc;
+  struct flock lock1, lock2;
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  // Get the root inode
+  rc = ceph_ll_lookup_root(cmount, &root);
+  ASSERT_EQ(rc, 0); 
+
+  // Get the inode and Fh corresponding to c_file
+  rc = ceph_ll_create(cmount, root, c_file, fileMode, O_RDWR | O_CREAT,
+		      &inode, &fh, &stx, 0, 0, perms);
+  ASSERT_EQ(rc, 0); 
+
+  // write lock twice
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, 42, false));
+
+  lock2.l_type = F_WRLCK;
+  lock2.l_whence = SEEK_SET;
+  lock2.l_start = 0;
+  lock2.l_len = 1024;
+  lock2.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock2, 43, false));
+
+  // Now try a conflicting read lock
+  lock2.l_type = F_RDLCK;
+  lock2.l_whence = SEEK_SET;
+  lock2.l_start = 100;
+  lock2.l_len = 100;
+  lock2.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock2, 43, false));
+
+  // Now do a getlk
+  ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
+  ASSERT_EQ(lock2.l_type, F_WRLCK);
+  ASSERT_EQ(lock2.l_start, 0);
+  ASSERT_EQ(lock2.l_len, 1024);
+  ASSERT_EQ(lock2.l_pid, getpid());
+
+  // Extend the range of the write lock
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 1024;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, 42, false));
+
+  // Now do a getlk
+  lock2.l_type = F_RDLCK;
+  lock2.l_whence = SEEK_SET;
+  lock2.l_start = 100;
+  lock2.l_len = 100;
+  lock2.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
+  ASSERT_EQ(lock2.l_type, F_WRLCK);
+  ASSERT_EQ(lock2.l_start, 0);
+  ASSERT_EQ(lock2.l_len, 2048);
+  ASSERT_EQ(lock2.l_pid, getpid());
+
+  // Now release part of the range
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 512;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, 42, false));
+
+  // Now do a getlk to check 1st part
+  lock2.l_type = F_RDLCK;
+  lock2.l_whence = SEEK_SET;
+  lock2.l_start = 100;
+  lock2.l_len = 100;
+  lock2.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
+  ASSERT_EQ(lock2.l_type, F_WRLCK);
+  ASSERT_EQ(lock2.l_start, 0);
+  ASSERT_EQ(lock2.l_len, 512);
+  ASSERT_EQ(lock2.l_pid, getpid());
+
+  // Now do a getlk to check 2nd part
+  lock2.l_type = F_RDLCK;
+  lock2.l_whence = SEEK_SET;
+  lock2.l_start = 2000;
+  lock2.l_len = 100;
+  lock2.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
+  ASSERT_EQ(lock2.l_type, F_WRLCK);
+  ASSERT_EQ(lock2.l_start, 1536);
+  ASSERT_EQ(lock2.l_len, 512);
+  ASSERT_EQ(lock2.l_pid, getpid());
+
+  // Now do a getlk to check released part
+  lock2.l_type = F_RDLCK;
+  lock2.l_whence = SEEK_SET;
+  lock2.l_start = 512;
+  lock2.l_len = 1024;
+  lock2.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
+  ASSERT_EQ(lock2.l_type, F_UNLCK);
+  ASSERT_EQ(lock2.l_start, 512);
+  ASSERT_EQ(lock2.l_len, 1024);
+  ASSERT_EQ(lock2.l_pid, getpid());
+
+  // Now downgrade the 1st part of the lock
+  lock1.l_type = F_RDLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 512;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, 42, false));
+
+  // Now do a getlk to check 1st part
+  lock2.l_type = F_WRLCK;
+  lock2.l_whence = SEEK_SET;
+  lock2.l_start = 100;
+  lock2.l_len = 100;
+  lock2.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
+  ASSERT_EQ(lock2.l_type, F_RDLCK);
+  ASSERT_EQ(lock2.l_start, 0);
+  ASSERT_EQ(lock2.l_len, 512);
+  ASSERT_EQ(lock2.l_pid, getpid());
+
+  // Now upgrade the 1st part of the lock
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 512;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, 42, false));
+
+  // Now do a getlk to check 1st part
+  lock2.l_type = F_WRLCK;
+  lock2.l_whence = SEEK_SET;
+  lock2.l_start = 100;
+  lock2.l_len = 100;
+  lock2.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
+  ASSERT_EQ(lock2.l_type, F_WRLCK);
+  ASSERT_EQ(lock2.l_start, 0);
+  ASSERT_EQ(lock2.l_len, 512);
+  ASSERT_EQ(lock2.l_pid, getpid());
+
+  ASSERT_EQ(0, ceph_ll_close(cmount, fh));
+  ASSERT_EQ(0, ceph_ll_unlink(cmount, root, c_file, perms));
+  CLEANUP_CEPH();
+}
+
+/* Locking in different threads */
+
+// Used by ConcurrentLocking test
+struct str_ConcurrentRecordLocking {
+  const char *file;
+  struct ceph_mount_info *cmount;  // !NULL if shared
+  sem_t sem[2];
+  sem_t semReply[2];
+  void sem_init(int pshared) {
+    ASSERT_EQ(0, ::sem_init(&sem[0], pshared, 0));
+    ASSERT_EQ(0, ::sem_init(&sem[1], pshared, 0));
+    ASSERT_EQ(0, ::sem_init(&semReply[0], pshared, 0));
+    ASSERT_EQ(0, ::sem_init(&semReply[1], pshared, 0));
+  }
+  void sem_destroy() {
+    ASSERT_EQ(0, ::sem_destroy(&sem[0]));
+    ASSERT_EQ(0, ::sem_destroy(&sem[1]));
+    ASSERT_EQ(0, ::sem_destroy(&semReply[0]));
+    ASSERT_EQ(0, ::sem_destroy(&semReply[1]));
+  }
+};
+
+// Wakeup main (for (N) steps)
+#define PING_MAIN(n) ASSERT_EQ(0, sem_post(&s.sem[n%2]))
+// Wait for main to wake us up (for (RN) steps)
+#define WAIT_MAIN(n) \
+  ASSERT_EQ(0, sem_timedwait(&s.semReply[n%2], abstime(ts, waitSlowMs)))
+
+// Wakeup worker (for (RN) steps)
+#define PING_WORKER(n) ASSERT_EQ(0, sem_post(&s.semReply[n%2]))
+// Wait for worker to wake us up (for (N) steps)
+#define WAIT_WORKER(n) \
+  ASSERT_EQ(0, sem_timedwait(&s.sem[n%2], abstime(ts, waitSlowMs)))
+// Worker shall not wake us up (for (N) steps)
+#define NOT_WAIT_WORKER(n) \
+  ASSERT_EQ(-1, sem_timedwait(&s.sem[n%2], abstime(ts, waitMs)))
+
+// Do twice an operation
+#define TWICE(EXPR) do {			\
+    EXPR;					\
+    EXPR;					\
+  } while(0)
+
+/* Locking in different threads */
+
+// Used by ConcurrentLocking test
+static void thread_ConcurrentRecordLocking(str_ConcurrentRecordLocking& s) {
+  struct ceph_mount_info *const cmount = s.cmount;
+  Fh *fh = NULL;
+  Inode *root = NULL, *inode = NULL;
+  struct ceph_statx stx;
+  struct flock lock1;
+  int rc;
+  struct timespec ts;
+
+  // Get the root inode
+  rc = ceph_ll_lookup_root(cmount, &root);
+  ASSERT_EQ(rc, 0); 
+
+  // Get the inode and Fh corresponding to c_file
+  rc = ceph_ll_create(cmount, root, s.file, fileMode, O_RDWR | O_CREAT,
+		      &inode, &fh, &stx, 0, 0, ceph_mount_perms(cmount));
+  ASSERT_EQ(rc, 0); 
+
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+
+  PING_MAIN(1); // (1)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), true));
+  PING_MAIN(2); // (2)
+
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+  PING_MAIN(3); // (3)
+
+  lock1.l_type = F_RDLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), true));
+  PING_MAIN(4); // (4)
+
+  WAIT_MAIN(1); // (R1)
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+  PING_MAIN(5); // (5)
+
+  WAIT_MAIN(2); // (R2)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), true));
+  PING_MAIN(6); // (6)
+
+  WAIT_MAIN(3); // (R3)
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+  PING_MAIN(7); // (7)
+
+  ASSERT_EQ(0, ceph_ll_close(cmount, fh));
+}
+
+// Used by ConcurrentRecordLocking test
+static void* thread_ConcurrentRecordLocking_(void *arg) {
+  str_ConcurrentRecordLocking *const s =
+    reinterpret_cast<str_ConcurrentRecordLocking*>(arg);
+  thread_ConcurrentRecordLocking(*s);
+  return NULL;
+}
+
+TEST(LibCephFS, ConcurrentRecordLocking) {
+  const pid_t mypid = getpid();
+  struct ceph_mount_info *cmount;
+  STARTUP_CEPH();
+
+  char c_file[1024];
+  sprintf(c_file, "recordlock_test_%d", mypid);
+  Fh *fh = NULL;
+  Inode *root = NULL, *inode = NULL;
+  struct ceph_statx stx;
+  struct flock lock1;
+  int rc;
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  // Get the root inode
+  rc = ceph_ll_lookup_root(cmount, &root);
+  ASSERT_EQ(rc, 0); 
+
+  // Get the inode and Fh corresponding to c_file
+  rc = ceph_ll_create(cmount, root, c_file, fileMode, O_RDWR | O_CREAT,
+		      &inode, &fh, &stx, 0, 0, perms);
+  ASSERT_EQ(rc, 0); 
+
+  // Lock
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), true));
+
+  // Start locker thread
+  pthread_t thread;
+  struct timespec ts;
+  str_ConcurrentRecordLocking s = { c_file, cmount };
+  s.sem_init(0);
+  ASSERT_EQ(0, pthread_create(&thread, NULL, thread_ConcurrentRecordLocking_, &s));
+  // Synchronization point with thread (failure: thread is dead)
+  WAIT_WORKER(1); // (1)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(2); // (2)
+
+  // Unlock
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+
+  // Shall have lock
+  // Synchronization point with thread (failure: thread is dead)
+  WAIT_WORKER(2); // (2)
+
+  // Synchronization point with thread (failure: thread is dead)
+  WAIT_WORKER(3); // (3)
+
+  // Wait for thread to share lock
+  WAIT_WORKER(4); // (4)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+  lock1.l_type = F_RDLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+
+  // Wake up thread to unlock shared lock
+  PING_WORKER(1); // (R1)
+  WAIT_WORKER(5); // (5)
+
+  // Now we can lock exclusively
+  // Upgrade to exclusive lock (as per POSIX)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), true));
+
+  // Wake up thread to lock shared lock
+  PING_WORKER(2); // (R2)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(6); // (6)
+
+  // Release lock ; thread will get it
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+  WAIT_WORKER(6); // (6)
+
+  // We no longer have the lock
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+  lock1.l_type = F_RDLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+
+  // Wake up thread to unlock exclusive lock
+  PING_WORKER(3); // (R3)
+  WAIT_WORKER(7); // (7)
+
+  // We can lock it again
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+
+  // Cleanup
+  void *retval = (void*) (uintptr_t) -1;
+  ASSERT_EQ(0, pthread_join(thread, &retval));
+  ASSERT_EQ(NULL, retval);
+  s.sem_destroy();
+  ASSERT_EQ(0, ceph_ll_close(cmount, fh));
+  ASSERT_EQ(0, ceph_ll_unlink(cmount, root, c_file, perms));
+  CLEANUP_CEPH();
+}
+
+TEST(LibCephFS, ThreesomeRecordLocking) {
+  const pid_t mypid = getpid();
+  struct ceph_mount_info *cmount;
+  STARTUP_CEPH();
+
+  char c_file[1024];
+  sprintf(c_file, "recordlock_test_%d", mypid);
+  Fh *fh = NULL;
+  Inode *root = NULL, *inode = NULL;
+  struct ceph_statx stx;
+  struct flock lock1;
+  int rc;
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  // Get the root inode
+  rc = ceph_ll_lookup_root(cmount, &root);
+  ASSERT_EQ(rc, 0); 
+
+  // Get the inode and Fh corresponding to c_file
+  rc = ceph_ll_create(cmount, root, c_file, fileMode, O_RDWR | O_CREAT,
+		      &inode, &fh, &stx, 0, 0, perms);
+  ASSERT_EQ(rc, 0); 
+
+  // Lock
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), true));
+
+  // Start locker thread
+  pthread_t thread[2];
+  struct timespec ts;
+  str_ConcurrentRecordLocking s = { c_file, cmount };
+  s.sem_init(0);
+  ASSERT_EQ(0, pthread_create(&thread[0], NULL, thread_ConcurrentRecordLocking_, &s));
+  ASSERT_EQ(0, pthread_create(&thread[1], NULL, thread_ConcurrentRecordLocking_, &s));
+  // Synchronization point with thread (failure: thread is dead)
+  TWICE(WAIT_WORKER(1)); // (1)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(2); // (2)
+
+  // Unlock
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+
+  // Shall have lock
+  TWICE(// Synchronization point with thread (failure: thread is dead)
+	WAIT_WORKER(2); // (2)
+	
+	// Synchronization point with thread (failure: thread is dead)
+	WAIT_WORKER(3)); // (3)
+  
+  // Wait for thread to share lock
+  TWICE(WAIT_WORKER(4)); // (4)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+  lock1.l_type = F_RDLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+
+  // Wake up thread to unlock shared lock
+  TWICE(PING_WORKER(1); // (R1)
+	WAIT_WORKER(5)); // (5)
+
+  // Now we can lock exclusively
+  // Upgrade to exclusive lock (as per POSIX)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), true));
+
+  TWICE(  // Wake up thread to lock shared lock
+	PING_WORKER(2); // (R2)
+	
+	// Shall not have lock immediately
+	NOT_WAIT_WORKER(6)); // (6)
+  
+  // Release lock ; thread will get it
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+  TWICE(WAIT_WORKER(6); // (6)
+	
+	// We no longer have the lock
+	lock1.l_type = F_WRLCK;
+	lock1.l_whence = SEEK_SET;
+	lock1.l_start = 0;
+	lock1.l_len = 1024;
+	lock1.l_pid = getpid();
+	ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+	lock1.l_type = F_RDLCK;
+	lock1.l_whence = SEEK_SET;
+	lock1.l_start = 0;
+	lock1.l_len = 1024;
+	lock1.l_pid = getpid();
+	ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+	
+	// Wake up thread to unlock exclusive lock
+	PING_WORKER(3); // (R3)
+	WAIT_WORKER(7); // (7)
+	);
+  
+  // We can lock it again
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+
+  // Cleanup
+  void *retval = (void*) (uintptr_t) -1;
+  ASSERT_EQ(0, pthread_join(thread[0], &retval));
+  ASSERT_EQ(NULL, retval);
+  ASSERT_EQ(0, pthread_join(thread[1], &retval));
+  ASSERT_EQ(NULL, retval);
+  s.sem_destroy();
+  ASSERT_EQ(0, ceph_ll_close(cmount, fh));
+  ASSERT_EQ(0, ceph_ll_unlink(cmount, root, c_file, perms));
+  CLEANUP_CEPH();
+}
+
+/* Locking in different processes */
+
+#define PROCESS_SLOW_MS() \
+  static const long waitMs = 100; \
+  (void) waitMs
+
+// Used by ConcurrentLocking test
+static void process_ConcurrentRecordLocking(str_ConcurrentRecordLocking& s) {
+  const pid_t mypid = getpid();
+  PROCESS_SLOW_MS();
+
+  struct ceph_mount_info *cmount = NULL;
+  struct timespec ts;
+  Fh *fh = NULL;
+  Inode *root = NULL, *inode = NULL;
+  struct ceph_statx stx;
+  int rc;
+  struct flock lock1;
+
+  STARTUP_CEPH();
+  s.cmount = cmount;
+
+  // Get the root inode
+  rc = ceph_ll_lookup_root(cmount, &root);
+  ASSERT_EQ(rc, 0); 
+
+  // Get the inode and Fh corresponding to c_file
+  rc = ceph_ll_create(cmount, root, s.file, fileMode, O_RDWR | O_CREAT,
+		      &inode, &fh, &stx, 0, 0, ceph_mount_perms(cmount));
+  ASSERT_EQ(rc, 0); 
+
+  WAIT_MAIN(1); // (R1)
+
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  PING_MAIN(1); // (1)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
+  PING_MAIN(2); // (2)
+
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  PING_MAIN(3); // (3)
+
+  lock1.l_type = F_RDLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
+  PING_MAIN(4); // (4)
+
+  WAIT_MAIN(2); // (R2)
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  PING_MAIN(5); // (5)
+
+  WAIT_MAIN(3); // (R3)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
+  PING_MAIN(6); // (6)
+
+  WAIT_MAIN(4); // (R4)
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  PING_MAIN(7); // (7)
+
+  ASSERT_EQ(0, ceph_ll_close(cmount, fh));
+  CLEANUP_CEPH();
+
+  s.sem_destroy();
+  exit(EXIT_SUCCESS);
+}
+
+// Disabled because of fork() issues (http://tracker.ceph.com/issues/16556)
+TEST(LibCephFS, DISABLED_InterProcessRecordLocking) {
+  PROCESS_SLOW_MS();
+  // Process synchronization
+  char c_file[1024];
+  const pid_t mypid = getpid();
+  sprintf(c_file, "recordlock_test_%d", mypid);
+  Fh *fh = NULL;
+  Inode *root = NULL, *inode = NULL;
+  struct ceph_statx stx;
+  struct flock lock1;
+  int rc;
+
+  // Note: the semaphores MUST be on a shared memory segment
+  str_ConcurrentRecordLocking *const shs =
+    reinterpret_cast<str_ConcurrentRecordLocking*>
+    (mmap(0, sizeof(*shs), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS,
+	  -1, 0));
+  str_ConcurrentRecordLocking &s = *shs;
+  s.file = c_file;
+  s.sem_init(1);
+
+  // Start locker process
+  const pid_t pid = fork();
+  ASSERT_GE(pid, 0);
+  if (pid == 0) {
+    process_ConcurrentRecordLocking(s);
+    exit(EXIT_FAILURE);
+  }
+
+  struct timespec ts;
+  struct ceph_mount_info *cmount;
+  STARTUP_CEPH();
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  // Get the root inode
+  rc = ceph_ll_lookup_root(cmount, &root);
+  ASSERT_EQ(rc, 0); 
+
+  // Get the inode and Fh corresponding to c_file
+  rc = ceph_ll_create(cmount, root, c_file, fileMode, O_RDWR | O_CREAT,
+		      &inode, &fh, &stx, 0, 0, perms);
+  ASSERT_EQ(rc, 0); 
+
+  // Lock
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
+
+  // Synchronization point with process (failure: process is dead)
+  PING_WORKER(1); // (R1)
+  WAIT_WORKER(1); // (1)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(2); // (2)
+
+  // Unlock
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+
+  // Shall have lock
+  // Synchronization point with process (failure: process is dead)
+  WAIT_WORKER(2); // (2)
+
+  // Synchronization point with process (failure: process is dead)
+  WAIT_WORKER(3); // (3)
+
+  // Wait for process to share lock
+  WAIT_WORKER(4); // (4)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  lock1.l_type = F_RDLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+
+  // Wake up process to unlock shared lock
+  PING_WORKER(2); // (R2)
+  WAIT_WORKER(5); // (5)
+
+  // Now we can lock exclusively
+  // Upgrade to exclusive lock (as per POSIX)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
+
+  // Wake up process to lock shared lock
+  PING_WORKER(3); // (R3)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(6); // (6)
+
+  // Release lock ; process will get it
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  WAIT_WORKER(6); // (6)
+
+  // We no longer have the lock
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  lock1.l_type = F_RDLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+
+  // Wake up process to unlock exclusive lock
+  PING_WORKER(4); // (R4)
+  WAIT_WORKER(7); // (7)
+
+  // We can lock it again
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+
+  // Wait pid
+  int status;
+  ASSERT_EQ(pid, waitpid(pid, &status, 0));
+  ASSERT_EQ(EXIT_SUCCESS, status);
+
+  // Cleanup
+  s.sem_destroy();
+  ASSERT_EQ(0, munmap(shs, sizeof(*shs)));
+  ASSERT_EQ(0, ceph_ll_close(cmount, fh));
+  ASSERT_EQ(0, ceph_ll_unlink(cmount, root, c_file, perms));
+  CLEANUP_CEPH();
+}
+
+// Disabled because of fork() issues (http://tracker.ceph.com/issues/16556)
+TEST(LibCephFS, DISABLED_ThreesomeInterProcessRecordLocking) {
+  PROCESS_SLOW_MS();
+  // Process synchronization
+  char c_file[1024];
+  const pid_t mypid = getpid();
+  sprintf(c_file, "recordlock_test_%d", mypid);
+  Fh *fh = NULL;
+  Inode *root = NULL, *inode = NULL;
+  struct ceph_statx stx;
+  struct flock lock1;
+  int rc;
+
+  // Note: the semaphores MUST be on a shared memory segment
+  str_ConcurrentRecordLocking *const shs =
+    reinterpret_cast<str_ConcurrentRecordLocking*>
+    (mmap(0, sizeof(*shs), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS,
+	  -1, 0));
+  str_ConcurrentRecordLocking &s = *shs;
+  s.file = c_file;
+  s.sem_init(1);
+
+  // Start locker processes
+  pid_t pid[2];
+  pid[0] = fork();
+  ASSERT_GE(pid[0], 0);
+  if (pid[0] == 0) {
+    process_ConcurrentRecordLocking(s);
+    exit(EXIT_FAILURE);
+  }
+  pid[1] = fork();
+  ASSERT_GE(pid[1], 0);
+  if (pid[1] == 0) {
+    process_ConcurrentRecordLocking(s);
+    exit(EXIT_FAILURE);
+  }
+
+  struct timespec ts;
+  struct ceph_mount_info *cmount;
+  STARTUP_CEPH();
+
+  // Get the root inode
+  rc = ceph_ll_lookup_root(cmount, &root);
+  ASSERT_EQ(rc, 0); 
+
+  // Get the inode and Fh corresponding to c_file
+  UserPerm *perms = ceph_mount_perms(cmount);
+  rc = ceph_ll_create(cmount, root, c_file, fileMode, O_RDWR | O_CREAT,
+		      &inode, &fh, &stx, 0, 0, perms);
+  ASSERT_EQ(rc, 0); 
+
+  // Lock
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
+
+  // Synchronization point with process (failure: process is dead)
+  TWICE(PING_WORKER(1)); // (R1)
+  TWICE(WAIT_WORKER(1)); // (1)
+
+  // Shall not have lock immediately
+  NOT_WAIT_WORKER(2); // (2)
+
+  // Unlock
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+
+  // Shall have lock
+  TWICE(// Synchronization point with process (failure: process is dead)
+	WAIT_WORKER(2); // (2)
+	
+	// Synchronization point with process (failure: process is dead)
+	WAIT_WORKER(3)); // (3)
+  
+  // Wait for process to share lock
+  TWICE(WAIT_WORKER(4)); // (4)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  lock1.l_type = F_RDLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+
+  // Wake up process to unlock shared lock
+  TWICE(PING_WORKER(2); // (R2)
+	WAIT_WORKER(5)); // (5)
+
+  // Now we can lock exclusively
+  // Upgrade to exclusive lock (as per POSIX)
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
+
+  TWICE(  // Wake up process to lock shared lock
+	PING_WORKER(3); // (R3)
+	
+	// Shall not have lock immediately
+	NOT_WAIT_WORKER(6)); // (6)
+  
+  // Release lock ; process will get it
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  TWICE(WAIT_WORKER(6); // (6)
+	
+	// We no longer have the lock
+	lock1.l_type = F_WRLCK;
+	lock1.l_whence = SEEK_SET;
+	lock1.l_start = 0;
+	lock1.l_len = 1024;
+	lock1.l_pid = getpid();
+	ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+	lock1.l_type = F_RDLCK;
+	lock1.l_whence = SEEK_SET;
+	lock1.l_start = 0;
+	lock1.l_len = 1024;
+	lock1.l_pid = getpid();
+	ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, ceph_pthread_self(), false));
+	
+	// Wake up process to unlock exclusive lock
+	PING_WORKER(4); // (R4)
+	WAIT_WORKER(7); // (7)
+	);
+  
+  // We can lock it again
+  lock1.l_type = F_WRLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+  lock1.l_type = F_UNLCK;
+  lock1.l_whence = SEEK_SET;
+  lock1.l_start = 0;
+  lock1.l_len = 1024;
+  lock1.l_pid = getpid();
+  ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
+
+  // Wait pids
+  int status;
+  ASSERT_EQ(pid[0], waitpid(pid[0], &status, 0));
+  ASSERT_EQ(EXIT_SUCCESS, status);
+  ASSERT_EQ(pid[1], waitpid(pid[1], &status, 0));
+  ASSERT_EQ(EXIT_SUCCESS, status);
+
+  // Cleanup
+  s.sem_destroy();
+  ASSERT_EQ(0, munmap(shs, sizeof(*shs)));
+  ASSERT_EQ(0, ceph_ll_close(cmount, fh));
+  ASSERT_EQ(0, ceph_ll_unlink(cmount, root, c_file, perms));
+  CLEANUP_CEPH();
+}
diff --git a/src/test/libcephfs/test.cc b/src/test/libcephfs/test.cc
new file mode 100644
index 00000000..fb7292e3
--- /dev/null
+++ b/src/test/libcephfs/test.cc
@@ -0,0 +1,2430 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2011 New Dream Network
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#include "include/compat.h"
+#include "gtest/gtest.h"
+#include "include/cephfs/libcephfs.h"
+#include "include/stat.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <dirent.h>
+#include <sys/uio.h>
+#include <sys/time.h>
+#include <sys/resource.h>
+
+#include "common/Clock.h"
+
+#ifdef __linux__
+#include <limits.h>
+#include <sys/xattr.h>
+#endif
+
+#include <map>
+#include <vector>
+#include <thread>
+
+TEST(LibCephFS, OpenEmptyComponent) {
+
+  pid_t mypid = getpid();
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  char c_dir[1024];
+  sprintf(c_dir, "/open_test_%d", mypid);
+  struct ceph_dir_result *dirp;
+
+  ASSERT_EQ(0, ceph_mkdirs(cmount, c_dir, 0777));
+
+  ASSERT_EQ(0, ceph_opendir(cmount, c_dir, &dirp));
+
+  char c_path[1024];
+  sprintf(c_path, "/open_test_%d//created_file_%d", mypid, mypid);
+  int fd = ceph_open(cmount, c_path, O_RDONLY|O_CREAT, 0666);
+  ASSERT_LT(0, fd);
+
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+  ASSERT_EQ(0, ceph_closedir(cmount, dirp));
+  ceph_shutdown(cmount);
+
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  fd = ceph_open(cmount, c_path, O_RDONLY, 0666);
+  ASSERT_LT(0, fd);
+
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, OpenReadTruncate) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  auto path = "test_open_rdt_" + std::to_string(getpid());
+  int fd = ceph_open(cmount, path.c_str(), O_WRONLY|O_CREAT, 0666);
+  ASSERT_LE(0, fd);
+
+  auto data = std::string("hello world");
+  ASSERT_EQ(ceph_write(cmount, fd, data.c_str(), data.size(), 0), (int)data.size());
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+
+  fd = ceph_open(cmount, path.c_str(), O_RDONLY, 0);
+  ASSERT_LE(0, fd);
+  ASSERT_EQ(ceph_ftruncate(cmount, fd, 0), -EBADF);
+  ASSERT_EQ(ceph_ftruncate(cmount, fd, 1), -EBADF);
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, OpenReadWrite) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  char c_path[1024];
+  sprintf(c_path, "test_open_rdwr_%d", getpid());
+  int fd = ceph_open(cmount, c_path, O_WRONLY|O_CREAT, 0666);
+  ASSERT_LT(0, fd);
+
+  const char *out_buf = "hello world";
+  size_t size = strlen(out_buf);
+  char in_buf[100];
+  ASSERT_EQ(ceph_write(cmount, fd, out_buf, size, 0), (int)size);
+  ASSERT_EQ(ceph_read(cmount, fd, in_buf, sizeof(in_buf), 0), -EBADF);
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+
+  fd = ceph_open(cmount, c_path, O_RDONLY, 0);
+  ASSERT_LT(0, fd);
+  ASSERT_EQ(ceph_write(cmount, fd, out_buf, size, 0), -EBADF);
+  ASSERT_EQ(ceph_read(cmount, fd, in_buf, sizeof(in_buf), 0), (int)size);
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+
+  fd = ceph_open(cmount, c_path, O_RDWR, 0);
+  ASSERT_LT(0, fd);
+  ASSERT_EQ(ceph_write(cmount, fd, out_buf, size, 0), (int)size);
+  ASSERT_EQ(ceph_read(cmount, fd, in_buf, sizeof(in_buf), 0), (int)size);
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, MountNonExist) {
+
+  struct ceph_mount_info *cmount;
+
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_NE(0, ceph_mount(cmount, "/non-exist"));
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, MountDouble) {
+
+  struct ceph_mount_info *cmount;
+
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+  ASSERT_EQ(-EISCONN, ceph_mount(cmount, "/"));
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, MountRemount) {
+
+  struct ceph_mount_info *cmount;
+
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+
+  CephContext *cct = ceph_get_mount_context(cmount);
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+  ASSERT_EQ(0, ceph_unmount(cmount));
+
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+  ASSERT_EQ(cct, ceph_get_mount_context(cmount));
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, UnmountUnmounted) {
+
+  struct ceph_mount_info *cmount;
+
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(-ENOTCONN, ceph_unmount(cmount));
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, ReleaseUnmounted) {
+
+  struct ceph_mount_info *cmount;
+
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_release(cmount));
+}
+
+TEST(LibCephFS, ReleaseMounted) {
+
+  struct ceph_mount_info *cmount;
+
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+  ASSERT_EQ(-EISCONN, ceph_release(cmount));
+  ASSERT_EQ(0, ceph_unmount(cmount));
+  ASSERT_EQ(0, ceph_release(cmount));
+}
+
+TEST(LibCephFS, UnmountRelease) {
+
+  struct ceph_mount_info *cmount;
+
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+  ASSERT_EQ(0, ceph_unmount(cmount));
+  ASSERT_EQ(0, ceph_release(cmount));
+}
+
+TEST(LibCephFS, Mount) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+  ceph_shutdown(cmount);
+
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, OpenLayout) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  /* valid layout */
+  char test_layout_file[256];
+  sprintf(test_layout_file, "test_layout_%d_b", getpid());
+  int fd = ceph_open_layout(cmount, test_layout_file, O_CREAT|O_WRONLY, 0666, (1<<20), 7, (1<<20), NULL);
+  ASSERT_GT(fd, 0);
+  char poolname[80];
+  ASSERT_LT(0, ceph_get_file_pool_name(cmount, fd, poolname, sizeof(poolname)));
+  ASSERT_LT(0, ceph_get_file_pool_name(cmount, fd, poolname, 0));
+
+  /* on already-written file (ENOTEMPTY) */
+  ceph_write(cmount, fd, "hello world", 11, 0);
+  ceph_close(cmount, fd);
+
+  char xattrk[128];
+  char xattrv[128];
+  sprintf(xattrk, "ceph.file.layout.stripe_unit");
+  sprintf(xattrv, "65536");
+  ASSERT_EQ(-ENOTEMPTY, ceph_setxattr(cmount, test_layout_file, xattrk, (void *)xattrv, 5, 0));
+
+  /* invalid layout */
+  sprintf(test_layout_file, "test_layout_%d_c", getpid());
+  fd = ceph_open_layout(cmount, test_layout_file, O_CREAT, 0666, (1<<20), 1, 19, NULL);
+  ASSERT_EQ(fd, -EINVAL);
+
+  /* with data pool */
+  sprintf(test_layout_file, "test_layout_%d_d", getpid());
+  fd = ceph_open_layout(cmount, test_layout_file, O_CREAT, 0666, (1<<20), 7, (1<<20), poolname);
+  ASSERT_GT(fd, 0);
+  ceph_close(cmount, fd);
+
+  /* with metadata pool (invalid) */
+  sprintf(test_layout_file, "test_layout_%d_e", getpid());
+  fd = ceph_open_layout(cmount, test_layout_file, O_CREAT, 0666, (1<<20), 7, (1<<20), "metadata");
+  ASSERT_EQ(fd, -EINVAL);
+
+  /* with metadata pool (does not exist) */
+  sprintf(test_layout_file, "test_layout_%d_f", getpid());
+  fd = ceph_open_layout(cmount, test_layout_file, O_CREAT, 0666, (1<<20), 7, (1<<20), "asdfjasdfjasdf");
+  ASSERT_EQ(fd, -EINVAL);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, DirLs) {
+
+  pid_t mypid = getpid();
+
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  struct ceph_dir_result *ls_dir = NULL;
+  char foostr[256];
+  sprintf(foostr, "dir_ls%d", mypid);
+  ASSERT_EQ(ceph_opendir(cmount, foostr, &ls_dir), -ENOENT);
+
+  ASSERT_EQ(ceph_mkdir(cmount, foostr, 0777), 0);
+  struct ceph_statx stx;
+  ASSERT_EQ(ceph_statx(cmount, foostr, &stx, 0, 0), 0);
+  ASSERT_NE(S_ISDIR(stx.stx_mode), 0);
+
+  char barstr[256];
+  sprintf(barstr, "dir_ls2%d", mypid);
+  ASSERT_EQ(ceph_statx(cmount, barstr, &stx, 0, AT_SYMLINK_NOFOLLOW), -ENOENT);
+
+  // insert files into directory and test open
+  char bazstr[256];
+  int i = 0, r = rand() % 4096;
+  if (getenv("LIBCEPHFS_RAND")) {
+    r = atoi(getenv("LIBCEPHFS_RAND"));
+  }
+  printf("rand: %d\n", r);
+  for(; i < r; ++i) {
+
+    sprintf(bazstr, "dir_ls%d/dirf%d", mypid, i);
+    int fd  = ceph_open(cmount, bazstr, O_CREAT|O_RDONLY, 0666);
+    ASSERT_GT(fd, 0);
+    ASSERT_EQ(ceph_close(cmount, fd), 0);
+
+    // set file sizes for readdirplus
+    ceph_truncate(cmount, bazstr, i);
+  }
+
+  ASSERT_EQ(ceph_opendir(cmount, foostr, &ls_dir), 0);
+
+  // not guaranteed to get . and .. first, but its a safe assumption in this case
+  struct dirent *result = ceph_readdir(cmount, ls_dir);
+  ASSERT_TRUE(result != NULL);
+  ASSERT_STREQ(result->d_name, ".");
+  result = ceph_readdir(cmount, ls_dir);
+  ASSERT_TRUE(result != NULL);
+  ASSERT_STREQ(result->d_name, "..");
+
+  std::vector<std::string> entries;
+  std::map<std::string, int64_t> offset_map;
+  int64_t offset = ceph_telldir(cmount, ls_dir);
+  for(i = 0; i < r; ++i) {
+    result = ceph_readdir(cmount, ls_dir);
+    ASSERT_TRUE(result != NULL);
+    entries.push_back(result->d_name);
+    offset_map[result->d_name] = offset;
+    offset = ceph_telldir(cmount, ls_dir);
+  }
+
+  ASSERT_TRUE(ceph_readdir(cmount, ls_dir) == NULL);
+  offset = ceph_telldir(cmount, ls_dir);
+
+  ASSERT_EQ(offset_map.size(), entries.size());
+  for(i = 0; i < r; ++i) {
+    sprintf(bazstr, "dirf%d", i);
+    ASSERT_TRUE(offset_map.count(bazstr) == 1);
+  }
+
+  // test seekdir
+  ceph_seekdir(cmount, ls_dir, offset);
+  ASSERT_TRUE(ceph_readdir(cmount, ls_dir) == NULL);
+
+  for (auto p = offset_map.begin(); p != offset_map.end(); ++p) {
+    ceph_seekdir(cmount, ls_dir, p->second);
+    result = ceph_readdir(cmount, ls_dir);
+    ASSERT_TRUE(result != NULL);
+    std::string d_name(result->d_name);
+    ASSERT_EQ(p->first, d_name);
+  }
+
+  // test rewinddir
+  ceph_rewinddir(cmount, ls_dir);
+
+  result = ceph_readdir(cmount, ls_dir);
+  ASSERT_TRUE(result != NULL);
+  ASSERT_STREQ(result->d_name, ".");
+  result = ceph_readdir(cmount, ls_dir);
+  ASSERT_TRUE(result != NULL);
+  ASSERT_STREQ(result->d_name, "..");
+
+  ceph_rewinddir(cmount, ls_dir);
+
+  int t = ceph_telldir(cmount, ls_dir);
+  ASSERT_GT(t, -1);
+
+  ASSERT_TRUE(ceph_readdir(cmount, ls_dir) != NULL);
+
+  // test seekdir - move back to the beginning
+  ceph_seekdir(cmount, ls_dir, t);
+
+  // test getdents
+  struct dirent *getdents_entries;
+  size_t getdents_entries_len = (r + 2) * sizeof(*getdents_entries);
+  getdents_entries = (struct dirent *)malloc(getdents_entries_len);
+
+  int count = 0;
+  std::vector<std::string> found;
+  while (true) {
+    int len = ceph_getdents(cmount, ls_dir, (char *)getdents_entries, getdents_entries_len);
+    if (len == 0)
+      break;
+    ASSERT_GT(len, 0);
+    ASSERT_TRUE((len % sizeof(*getdents_entries)) == 0);
+    int n = len / sizeof(*getdents_entries);
+    int j;
+    if (count == 0) {
+      ASSERT_STREQ(getdents_entries[0].d_name, ".");
+      ASSERT_STREQ(getdents_entries[1].d_name, "..");
+      j = 2;
+    } else {
+      j = 0;
+    }
+    count += n;
+    for(; j < n; ++i, ++j) {
+      const char *name = getdents_entries[j].d_name;
+      found.push_back(name);
+    }
+  }
+  ASSERT_EQ(found, entries);
+  free(getdents_entries);
+
+  // test readdir_r
+  ceph_rewinddir(cmount, ls_dir);
+
+  result = ceph_readdir(cmount, ls_dir);
+  ASSERT_TRUE(result != NULL);
+  ASSERT_STREQ(result->d_name, ".");
+  result = ceph_readdir(cmount, ls_dir);
+  ASSERT_TRUE(result != NULL);
+  ASSERT_STREQ(result->d_name, "..");
+
+  found.clear();
+  while (true) {
+    struct dirent rdent;
+    int len = ceph_readdir_r(cmount, ls_dir, &rdent);
+    if (len == 0)
+      break;
+    ASSERT_EQ(len, 1);
+    found.push_back(rdent.d_name);
+  }
+  ASSERT_EQ(found, entries);
+
+  // test readdirplus
+  ceph_rewinddir(cmount, ls_dir);
+
+  result = ceph_readdir(cmount, ls_dir);
+  ASSERT_TRUE(result != NULL);
+  ASSERT_STREQ(result->d_name, ".");
+  result = ceph_readdir(cmount, ls_dir);
+  ASSERT_TRUE(result != NULL);
+  ASSERT_STREQ(result->d_name, "..");
+
+  found.clear();
+  while (true) {
+    struct dirent rdent;
+    struct ceph_statx stx;
+    int len = ceph_readdirplus_r(cmount, ls_dir, &rdent, &stx,
+				 CEPH_STATX_SIZE, AT_NO_ATTR_SYNC, NULL);
+    if (len == 0)
+      break;
+    ASSERT_EQ(len, 1);
+    const char *name = rdent.d_name;
+    found.push_back(name);
+    int size;
+    sscanf(name, "dirf%d", &size);
+    ASSERT_TRUE(stx.stx_mask & CEPH_STATX_SIZE);
+    ASSERT_EQ(stx.stx_size, (size_t)size);
+    ASSERT_EQ(stx.stx_ino, rdent.d_ino);
+    //ASSERT_EQ(st.st_mode, (mode_t)0666);
+  }
+  ASSERT_EQ(found, entries);
+
+  ASSERT_EQ(ceph_closedir(cmount, ls_dir), 0);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, ManyNestedDirs) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  const char *many_path = "a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a";
+  ASSERT_EQ(ceph_mkdirs(cmount, many_path, 0755), 0);
+
+  int i = 0;
+
+  for(; i < 39; ++i) {
+    ASSERT_EQ(ceph_chdir(cmount, "a"), 0);
+
+    struct ceph_dir_result *dirp;
+    ASSERT_EQ(ceph_opendir(cmount, "a", &dirp), 0);
+    struct dirent *dent = ceph_readdir(cmount, dirp);
+    ASSERT_TRUE(dent != NULL);
+    ASSERT_STREQ(dent->d_name, ".");
+    dent = ceph_readdir(cmount, dirp);
+    ASSERT_TRUE(dent != NULL);
+    ASSERT_STREQ(dent->d_name, "..");
+    dent = ceph_readdir(cmount, dirp);
+    ASSERT_TRUE(dent != NULL);
+    ASSERT_STREQ(dent->d_name, "a");
+    ASSERT_EQ(ceph_closedir(cmount, dirp), 0);
+  }
+
+  ASSERT_STREQ(ceph_getcwd(cmount), "/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a/a");
+
+  ASSERT_EQ(ceph_chdir(cmount, "a/a/a"), 0);
+
+  for(i = 0; i < 39; ++i) {
+    ASSERT_EQ(ceph_chdir(cmount, ".."), 0);
+    ASSERT_EQ(ceph_rmdir(cmount, "a"), 0);
+  }
+
+  ASSERT_EQ(ceph_chdir(cmount, "/"), 0);
+
+  ASSERT_EQ(ceph_rmdir(cmount, "a/a/a"), 0);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, Xattrs) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_xattr_file[256];
+  sprintf(test_xattr_file, "test_xattr_%d", getpid());
+  int fd = ceph_open(cmount, test_xattr_file, O_CREAT, 0666);
+  ASSERT_GT(fd, 0);
+
+  char i = 'a';
+  char xattrk[128];
+  char xattrv[128];
+  for(; i < 'a'+26; ++i) {
+    sprintf(xattrk, "user.test_xattr_%c", i);
+    int len = sprintf(xattrv, "testxattr%c", i);
+    ASSERT_EQ(ceph_setxattr(cmount, test_xattr_file, xattrk, (void *) xattrv, len, XATTR_CREATE), 0);
+  }
+
+  // zero size should return required buffer length
+  int len_needed = ceph_listxattr(cmount, test_xattr_file, NULL, 0);
+  ASSERT_GT(len_needed, 0);
+
+  // buffer size smaller than needed should fail
+  char xattrlist[128*26];
+  ASSERT_GT(sizeof(xattrlist), (size_t)len_needed);
+  int len = ceph_listxattr(cmount, test_xattr_file, xattrlist, len_needed - 1);
+  ASSERT_EQ(-ERANGE, len);
+
+  len = ceph_listxattr(cmount, test_xattr_file, xattrlist, sizeof(xattrlist));
+  ASSERT_EQ(len, len_needed);
+  char *p = xattrlist;
+  char *n;
+  i = 'a';
+  while (len > 0) {
+    // skip/ignore the dir layout
+    if (strcmp(p, "ceph.dir.layout") == 0 ||
+	strcmp(p, "ceph.file.layout") == 0) {
+      len -= strlen(p) + 1;
+      p += strlen(p) + 1;
+      continue;
+    }
+
+    sprintf(xattrk, "user.test_xattr_%c", i);
+    ASSERT_STREQ(p, xattrk);
+
+    char gxattrv[128];
+    std::cout << "getting attr " << p << std::endl;
+    int alen = ceph_getxattr(cmount, test_xattr_file, p, (void *) gxattrv, 128);
+    ASSERT_GT(alen, 0);
+    sprintf(xattrv, "testxattr%c", i);
+    ASSERT_TRUE(!strncmp(xattrv, gxattrv, alen));
+
+    n = index(p, '\0');
+    n++;
+    len -= (n - p);
+    p = n;
+    ++i;
+  }
+
+  i = 'a';
+  for(i = 'a'; i < 'a'+26; ++i) {
+    sprintf(xattrk, "user.test_xattr_%c", i);
+    ASSERT_EQ(ceph_removexattr(cmount, test_xattr_file, xattrk), 0);
+  }
+
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+
+}
+
+TEST(LibCephFS, Xattrs_ll) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_xattr_file[256];
+  sprintf(test_xattr_file, "test_xattr_%d", getpid());
+  int fd = ceph_open(cmount, test_xattr_file, O_CREAT, 0666);
+  ASSERT_GT(fd, 0);
+  ceph_close(cmount, fd);
+
+  Inode *root = NULL;
+  Inode *existent_file_handle = NULL;
+
+  int res = ceph_ll_lookup_root(cmount, &root);
+  ASSERT_EQ(res, 0);
+
+  UserPerm *perms = ceph_mount_perms(cmount);
+  struct ceph_statx stx;
+
+  res = ceph_ll_lookup(cmount, root, test_xattr_file, &existent_file_handle,
+		       &stx, 0, 0, perms);
+  ASSERT_EQ(res, 0);
+
+  const char *valid_name = "user.attrname";
+  const char *value = "attrvalue";
+  char value_buf[256] = { 0 };
+
+  res = ceph_ll_setxattr(cmount, existent_file_handle, valid_name, value, strlen(value), 0, perms);
+  ASSERT_EQ(res, 0);
+
+  res = ceph_ll_getxattr(cmount, existent_file_handle, valid_name, value_buf, 256, perms);
+  ASSERT_EQ(res, (int)strlen(value));
+
+  value_buf[res] = '\0';
+  ASSERT_STREQ(value_buf, value);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, LstatSlashdot) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  struct ceph_statx stx;
+  ASSERT_EQ(ceph_statx(cmount, "/.", &stx, 0, AT_SYMLINK_NOFOLLOW), 0);
+  ASSERT_EQ(ceph_statx(cmount, ".", &stx, 0, AT_SYMLINK_NOFOLLOW), 0);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, StatDirNlink) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_dir1[256];
+  sprintf(test_dir1, "dir1_symlinks_%d", getpid());
+  ASSERT_EQ(ceph_mkdir(cmount, test_dir1, 0700), 0);
+
+  int fd = ceph_open(cmount, test_dir1, O_DIRECTORY|O_RDONLY, 0);
+  ASSERT_GT(fd, 0);
+  struct ceph_statx stx;
+  ASSERT_EQ(ceph_fstatx(cmount, fd, &stx, CEPH_STATX_NLINK, AT_SYMLINK_NOFOLLOW), 0);
+  ASSERT_EQ(stx.stx_nlink, 2u);
+
+  {
+    char test_dir2[296];
+    sprintf(test_dir2, "%s/.", test_dir1);
+    ASSERT_EQ(ceph_statx(cmount, test_dir2, &stx, CEPH_STATX_NLINK, AT_SYMLINK_NOFOLLOW), 0);
+    ASSERT_EQ(stx.stx_nlink, 2u);
+  }
+
+  {
+    char test_dir2[296];
+    sprintf(test_dir2, "%s/1", test_dir1);
+    ASSERT_EQ(ceph_mkdir(cmount, test_dir2, 0700), 0);
+    ASSERT_EQ(ceph_statx(cmount, test_dir2, &stx, CEPH_STATX_NLINK, AT_SYMLINK_NOFOLLOW), 0);
+    ASSERT_EQ(stx.stx_nlink, 2u);
+      ASSERT_EQ(ceph_statx(cmount, test_dir1, &stx, CEPH_STATX_NLINK, AT_SYMLINK_NOFOLLOW), 0);
+      ASSERT_EQ(stx.stx_nlink, 3u);
+    sprintf(test_dir2, "%s/2", test_dir1);
+    ASSERT_EQ(ceph_mkdir(cmount, test_dir2, 0700), 0);
+      ASSERT_EQ(ceph_statx(cmount, test_dir1, &stx, CEPH_STATX_NLINK, AT_SYMLINK_NOFOLLOW), 0);
+      ASSERT_EQ(stx.stx_nlink, 4u);
+    sprintf(test_dir2, "%s/1/1", test_dir1);
+    ASSERT_EQ(ceph_mkdir(cmount, test_dir2, 0700), 0);
+      ASSERT_EQ(ceph_statx(cmount, test_dir1, &stx, CEPH_STATX_NLINK, AT_SYMLINK_NOFOLLOW), 0);
+      ASSERT_EQ(stx.stx_nlink, 4u);
+    ASSERT_EQ(ceph_rmdir(cmount, test_dir2), 0);
+      ASSERT_EQ(ceph_statx(cmount, test_dir1, &stx, CEPH_STATX_NLINK, AT_SYMLINK_NOFOLLOW), 0);
+      ASSERT_EQ(stx.stx_nlink, 4u);
+    sprintf(test_dir2, "%s/1", test_dir1);
+    ASSERT_EQ(ceph_rmdir(cmount, test_dir2), 0);
+      ASSERT_EQ(ceph_statx(cmount, test_dir1, &stx, CEPH_STATX_NLINK, AT_SYMLINK_NOFOLLOW), 0);
+      ASSERT_EQ(stx.stx_nlink, 3u);
+    sprintf(test_dir2, "%s/2", test_dir1);
+    ASSERT_EQ(ceph_rmdir(cmount, test_dir2), 0);
+      ASSERT_EQ(ceph_statx(cmount, test_dir1, &stx, CEPH_STATX_NLINK, AT_SYMLINK_NOFOLLOW), 0);
+      ASSERT_EQ(stx.stx_nlink, 2u);
+  }
+
+  ASSERT_EQ(ceph_rmdir(cmount, test_dir1), 0);
+  ASSERT_EQ(ceph_fstatx(cmount, fd, &stx, CEPH_STATX_NLINK, AT_SYMLINK_NOFOLLOW), 0);
+  ASSERT_EQ(stx.stx_nlink, 0u);
+
+  ceph_close(cmount, fd);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, DoubleChmod) {
+
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_file[256];
+  sprintf(test_file, "test_perms_%d", getpid());
+
+  int fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0666);
+  ASSERT_GT(fd, 0);
+
+  // write some stuff
+  const char *bytes = "foobarbaz";
+  ASSERT_EQ(ceph_write(cmount, fd, bytes, strlen(bytes), 0), (int)strlen(bytes));
+
+  ceph_close(cmount, fd);
+
+  // set perms to read but can't write
+  ASSERT_EQ(ceph_chmod(cmount, test_file, 0400), 0);
+
+  fd = ceph_open(cmount, test_file, O_RDWR, 0);
+  ASSERT_EQ(fd, -EACCES);
+
+  fd = ceph_open(cmount, test_file, O_RDONLY, 0);
+  ASSERT_GT(fd, -1);
+
+  char buf[100];
+  int ret = ceph_read(cmount, fd, buf, 100, 0);
+  ASSERT_EQ(ret, (int)strlen(bytes));
+  buf[ret] = '\0';
+  ASSERT_STREQ(buf, bytes);
+
+  ASSERT_EQ(ceph_write(cmount, fd, bytes, strlen(bytes), 0), -EBADF);
+
+  ceph_close(cmount, fd);
+
+  // reset back to writeable
+  ASSERT_EQ(ceph_chmod(cmount, test_file, 0600), 0);
+
+  // ensure perms are correct
+  struct ceph_statx stx;
+  ASSERT_EQ(ceph_statx(cmount, test_file, &stx, CEPH_STATX_MODE, AT_SYMLINK_NOFOLLOW), 0);
+  ASSERT_EQ(stx.stx_mode, 0100600U);
+
+  fd = ceph_open(cmount, test_file, O_RDWR, 0);
+  ASSERT_GT(fd, 0);
+
+  ASSERT_EQ(ceph_write(cmount, fd, bytes, strlen(bytes), 0), (int)strlen(bytes));
+  ceph_close(cmount, fd);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, Fchmod) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_file[256];
+  sprintf(test_file, "test_perms_%d", getpid());
+
+  int fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0666);
+  ASSERT_GT(fd, 0);
+
+  // write some stuff
+  const char *bytes = "foobarbaz";
+  ASSERT_EQ(ceph_write(cmount, fd, bytes, strlen(bytes), 0), (int)strlen(bytes));
+
+  // set perms to read but can't write
+  ASSERT_EQ(ceph_fchmod(cmount, fd, 0400), 0);
+
+  char buf[100];
+  int ret = ceph_read(cmount, fd, buf, 100, 0);
+  ASSERT_EQ(ret, (int)strlen(bytes));
+  buf[ret] = '\0';
+  ASSERT_STREQ(buf, bytes);
+
+  ASSERT_EQ(ceph_write(cmount, fd, bytes, strlen(bytes), 0), (int)strlen(bytes));
+
+  ceph_close(cmount, fd);
+
+  ASSERT_EQ(ceph_open(cmount, test_file, O_RDWR, 0), -EACCES);
+
+  // reset back to writeable
+  ASSERT_EQ(ceph_chmod(cmount, test_file, 0600), 0);
+
+  fd = ceph_open(cmount, test_file, O_RDWR, 0);
+  ASSERT_GT(fd, 0);
+
+  ASSERT_EQ(ceph_write(cmount, fd, bytes, strlen(bytes), 0), (int)strlen(bytes));
+  ceph_close(cmount, fd);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, Fchown) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_file[256];
+  sprintf(test_file, "test_fchown_%d", getpid());
+
+  int fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0666);
+  ASSERT_GT(fd, 0);
+
+  // set perms to readable and writeable only by owner
+  ASSERT_EQ(ceph_fchmod(cmount, fd, 0600), 0);
+
+  // change ownership to nobody -- we assume nobody exists and id is always 65534
+  ASSERT_EQ(ceph_conf_set(cmount, "client_permissions", "0"), 0);
+  ASSERT_EQ(ceph_fchown(cmount, fd, 65534, 65534), 0);
+  ASSERT_EQ(ceph_conf_set(cmount, "client_permissions", "1"), 0);
+
+  ceph_close(cmount, fd);
+
+  fd = ceph_open(cmount, test_file, O_RDWR, 0);
+  ASSERT_EQ(fd, -EACCES);
+
+  ceph_shutdown(cmount);
+}
+
+#if defined(__linux__) && defined(O_PATH)
+TEST(LibCephFS, FlagO_PATH) {
+  struct ceph_mount_info *cmount;
+
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, NULL));
+
+  char test_file[PATH_MAX];
+  sprintf(test_file, "test_oflag_%d", getpid());
+
+  int fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR|O_PATH, 0666);
+  ASSERT_EQ(-ENOENT, fd);
+
+  fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0666);
+  ASSERT_GT(fd, 0);
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+
+  // ok, the file has been created. perform real checks now
+  fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR|O_PATH, 0666);
+  ASSERT_GT(fd, 0);
+
+  char buf[128];
+  ASSERT_EQ(-EBADF, ceph_read(cmount, fd, buf, sizeof(buf), 0));
+  ASSERT_EQ(-EBADF, ceph_write(cmount, fd, buf, sizeof(buf), 0));
+
+  // set perms to readable and writeable only by owner
+  ASSERT_EQ(-EBADF, ceph_fchmod(cmount, fd, 0600));
+
+  // change ownership to nobody -- we assume nobody exists and id is always 65534
+  ASSERT_EQ(-EBADF, ceph_fchown(cmount, fd, 65534, 65534));
+
+  // try to sync
+  ASSERT_EQ(-EBADF, ceph_fsync(cmount, fd, false));
+
+  struct ceph_statx stx;
+  ASSERT_EQ(0, ceph_fstatx(cmount, fd, &stx, 0, 0));
+
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+  ceph_shutdown(cmount);
+}
+#endif /* __linux */
+
+TEST(LibCephFS, Symlinks) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_file[256];
+  sprintf(test_file, "test_symlinks_%d", getpid());
+
+  int fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0666);
+  ASSERT_GT(fd, 0);
+
+  ceph_close(cmount, fd);
+
+  char test_symlink[256];
+  sprintf(test_symlink, "test_symlinks_sym_%d", getpid());
+
+  ASSERT_EQ(ceph_symlink(cmount, test_file, test_symlink), 0);
+
+  // test the O_NOFOLLOW case
+  fd = ceph_open(cmount, test_symlink, O_NOFOLLOW, 0);
+  ASSERT_EQ(fd, -ELOOP);
+
+  // stat the original file
+  struct ceph_statx stx_orig;
+  ASSERT_EQ(ceph_statx(cmount, test_file, &stx_orig, CEPH_STATX_ALL_STATS, 0), 0);
+  // stat the symlink
+  struct ceph_statx stx_symlink_orig;
+  ASSERT_EQ(ceph_statx(cmount, test_symlink, &stx_symlink_orig, CEPH_STATX_ALL_STATS, 0), 0);
+  // ensure the statx bufs are equal
+  ASSERT_EQ(memcmp(&stx_orig, &stx_symlink_orig, sizeof(stx_orig)), 0);
+
+  sprintf(test_file, "/test_symlinks_abs_%d", getpid());
+
+  fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0666);
+  ASSERT_GT(fd, 0);
+
+  ceph_close(cmount, fd);
+
+  sprintf(test_symlink, "/test_symlinks_abs_sym_%d", getpid());
+
+  ASSERT_EQ(ceph_symlink(cmount, test_file, test_symlink), 0);
+
+  // stat the original file
+  ASSERT_EQ(ceph_statx(cmount, test_file, &stx_orig, CEPH_STATX_ALL_STATS, 0), 0);
+  // stat the symlink
+  ASSERT_EQ(ceph_statx(cmount, test_symlink, &stx_symlink_orig, CEPH_STATX_ALL_STATS, 0), 0);
+  // ensure the statx bufs are equal
+  ASSERT_TRUE(!memcmp(&stx_orig, &stx_symlink_orig, sizeof(stx_orig)));
+
+  // test lstat
+  ASSERT_EQ(ceph_statx(cmount, test_symlink, &stx_orig, CEPH_STATX_ALL_STATS, AT_SYMLINK_NOFOLLOW), 0);
+  ASSERT_TRUE(S_ISLNK(stx_orig.stx_mode));
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, DirSyms) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_dir1[256];
+  sprintf(test_dir1, "dir1_symlinks_%d", getpid());
+
+  ASSERT_EQ(ceph_mkdir(cmount, test_dir1, 0700), 0);
+
+  char test_symdir[256];
+  sprintf(test_symdir, "symdir_symlinks_%d", getpid());
+
+  ASSERT_EQ(ceph_symlink(cmount, test_dir1, test_symdir), 0);
+
+  char test_file[256];
+  sprintf(test_file, "/symdir_symlinks_%d/test_symdir_file", getpid());
+  int fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0600);
+  ASSERT_GT(fd, 0);
+  ceph_close(cmount, fd);
+
+  struct ceph_statx stx;
+  ASSERT_EQ(ceph_statx(cmount, test_file, &stx, 0, AT_SYMLINK_NOFOLLOW), 0);
+
+  // ensure that its a file not a directory we get back
+  ASSERT_TRUE(S_ISREG(stx.stx_mode));
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, LoopSyms) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_dir1[256];
+  sprintf(test_dir1, "dir1_loopsym_%d", getpid());
+
+  ASSERT_EQ(ceph_mkdir(cmount, test_dir1, 0700), 0);
+
+  char test_dir2[256];
+  sprintf(test_dir2, "/dir1_loopsym_%d/loop_dir", getpid());
+
+  ASSERT_EQ(ceph_mkdir(cmount, test_dir2, 0700), 0);
+
+  // symlink it itself:  /path/to/mysym -> /path/to/mysym
+  char test_symdir[256];
+  sprintf(test_symdir, "/dir1_loopsym_%d/loop_dir/symdir", getpid());
+
+  ASSERT_EQ(ceph_symlink(cmount, test_symdir, test_symdir), 0);
+
+  char test_file[256];
+  sprintf(test_file, "/dir1_loopsym_%d/loop_dir/symdir/test_loopsym_file", getpid());
+  int fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0600);
+  ASSERT_EQ(fd, -ELOOP);
+
+  // loop: /a -> /b, /b -> /c, /c -> /a
+  char a[264], b[264], c[264];
+  sprintf(a, "/%s/a", test_dir1);
+  sprintf(b, "/%s/b", test_dir1);
+  sprintf(c, "/%s/c", test_dir1);
+  ASSERT_EQ(ceph_symlink(cmount, a, b), 0);
+  ASSERT_EQ(ceph_symlink(cmount, b, c), 0);
+  ASSERT_EQ(ceph_symlink(cmount, c, a), 0);
+  ASSERT_EQ(ceph_open(cmount, a, O_RDWR, 0), -ELOOP);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, HardlinkNoOriginal) {
+
+  int mypid = getpid();
+
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char dir[256];
+  sprintf(dir, "/test_rmdirfail%d", mypid);
+  ASSERT_EQ(ceph_mkdir(cmount, dir, 0777), 0);
+
+  ASSERT_EQ(ceph_chdir(cmount, dir), 0);
+
+  int fd = ceph_open(cmount, "f1", O_CREAT, 0644);
+  ASSERT_GT(fd, 0);
+
+  ceph_close(cmount, fd);
+
+  // create hard link
+  ASSERT_EQ(ceph_link(cmount, "f1", "hardl1"), 0);
+
+  // remove file link points to
+  ASSERT_EQ(ceph_unlink(cmount, "f1"), 0);
+
+  ceph_shutdown(cmount);
+
+  // now cleanup
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+  ASSERT_EQ(ceph_chdir(cmount, dir), 0);
+  ASSERT_EQ(ceph_unlink(cmount, "hardl1"), 0);
+  ASSERT_EQ(ceph_rmdir(cmount, dir), 0);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, BadArgument) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  int fd = ceph_open(cmount, "test_file", O_CREAT|O_RDWR, 0666);
+  ASSERT_GT(fd, 0);
+  char buf[100];
+  ASSERT_EQ(ceph_write(cmount, fd, buf, sizeof(buf), 0), (int)sizeof(buf));
+  ASSERT_EQ(ceph_read(cmount, fd, buf, 0, 5), 0);
+  ceph_close(cmount, fd);
+  ASSERT_EQ(ceph_unlink(cmount, "test_file"), 0);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, BadFileDesc) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  ASSERT_EQ(ceph_fchmod(cmount, -1, 0655), -EBADF);
+  ASSERT_EQ(ceph_close(cmount, -1), -EBADF);
+  ASSERT_EQ(ceph_lseek(cmount, -1, 0, SEEK_SET), -EBADF);
+
+  char buf[0];
+  ASSERT_EQ(ceph_read(cmount, -1, buf, 0, 0), -EBADF);
+  ASSERT_EQ(ceph_write(cmount, -1, buf, 0, 0), -EBADF);
+
+  ASSERT_EQ(ceph_ftruncate(cmount, -1, 0), -EBADF);
+  ASSERT_EQ(ceph_fsync(cmount, -1, 0), -EBADF);
+
+  struct ceph_statx stx;
+  ASSERT_EQ(ceph_fstatx(cmount, -1, &stx, 0, 0), -EBADF);
+
+  struct sockaddr_storage addr;
+  ASSERT_EQ(ceph_get_file_stripe_address(cmount, -1, 0, &addr, 1), -EBADF);
+
+  ASSERT_EQ(ceph_get_file_stripe_unit(cmount, -1), -EBADF);
+  ASSERT_EQ(ceph_get_file_pool(cmount, -1), -EBADF);
+  char poolname[80];
+  ASSERT_EQ(ceph_get_file_pool_name(cmount, -1, poolname, sizeof(poolname)), -EBADF);
+  ASSERT_EQ(ceph_get_file_replication(cmount, -1), -EBADF);
+  ASSERT_EQ(ceph_get_file_object_size(cmount, -1), -EBADF);
+  int stripe_unit, stripe_count, object_size, pg_pool;
+  ASSERT_EQ(ceph_get_file_layout(cmount, -1, &stripe_unit, &stripe_count, &object_size, &pg_pool), -EBADF);
+  ASSERT_EQ(ceph_get_file_stripe_count(cmount, -1), -EBADF);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, ReadEmptyFile) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  // test the read_sync path in the client for zero files
+  ASSERT_EQ(ceph_conf_set(cmount, "client_debug_force_sync_read", "true"), 0);
+
+  int mypid = getpid();
+  char testf[256];
+
+  sprintf(testf, "test_reademptyfile%d", mypid);
+  int fd = ceph_open(cmount, testf, O_CREAT|O_TRUNC|O_WRONLY, 0644);
+  ASSERT_GT(fd, 0);
+
+  ceph_close(cmount, fd);
+
+  fd = ceph_open(cmount, testf, O_RDONLY, 0);
+  ASSERT_GT(fd, 0);
+
+  char buf[4096];
+  ASSERT_EQ(ceph_read(cmount, fd, buf, 4096, 0), 0);
+
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, PreadvPwritev) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  int mypid = getpid();
+  char testf[256];
+
+  sprintf(testf, "test_preadvpwritevfile%d", mypid);
+  int fd = ceph_open(cmount, testf, O_CREAT|O_RDWR, 0666);
+  ASSERT_GT(fd, 0);
+
+  char out0[] = "hello ";
+  char out1[] = "world\n";
+  struct iovec iov_out[2] = {
+	{out0, sizeof(out0)},
+	{out1, sizeof(out1)},
+  };
+  char in0[sizeof(out0)];
+  char in1[sizeof(out1)];
+  struct iovec iov_in[2] = {
+	{in0, sizeof(in0)},
+	{in1, sizeof(in1)},
+  };
+  ssize_t nwritten = iov_out[0].iov_len + iov_out[1].iov_len; 
+  ssize_t nread = iov_in[0].iov_len + iov_in[1].iov_len; 
+
+  ASSERT_EQ(ceph_pwritev(cmount, fd, iov_out, 2, 0), nwritten);
+  ASSERT_EQ(ceph_preadv(cmount, fd, iov_in, 2, 0), nread);
+  ASSERT_EQ(0, strncmp((const char*)iov_in[0].iov_base, (const char*)iov_out[0].iov_base, iov_out[0].iov_len));
+  ASSERT_EQ(0, strncmp((const char*)iov_in[1].iov_base, (const char*)iov_out[1].iov_base, iov_out[1].iov_len));
+
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, LlreadvLlwritev) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  int mypid = getpid();
+  char filename[256];
+
+  sprintf(filename, "test_llreadvllwritevfile%u", mypid);
+
+  Inode *root, *file;
+  ASSERT_EQ(ceph_ll_lookup_root(cmount, &root), 0);
+
+  Fh *fh;
+  struct ceph_statx stx;
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  ASSERT_EQ(ceph_ll_create(cmount, root, filename, 0666,
+		    O_RDWR|O_CREAT|O_TRUNC, &file, &fh, &stx, 0, 0, perms), 0);
+
+  /* Reopen read-only */
+  char out0[] = "hello ";
+  char out1[] = "world\n";
+  struct iovec iov_out[2] = {
+	{out0, sizeof(out0)},
+	{out1, sizeof(out1)},
+  };
+  char in0[sizeof(out0)];
+  char in1[sizeof(out1)];
+  struct iovec iov_in[2] = {
+	{in0, sizeof(in0)},
+	{in1, sizeof(in1)},
+  };
+  ssize_t nwritten = iov_out[0].iov_len + iov_out[1].iov_len;
+  ssize_t nread = iov_in[0].iov_len + iov_in[1].iov_len;
+
+  ASSERT_EQ(ceph_ll_writev(cmount, fh, iov_out, 2, 0), nwritten);
+  ASSERT_EQ(ceph_ll_readv(cmount, fh, iov_in, 2, 0), nread);
+  ASSERT_EQ(0, strncmp((const char*)iov_in[0].iov_base, (const char*)iov_out[0].iov_base, iov_out[0].iov_len));
+  ASSERT_EQ(0, strncmp((const char*)iov_in[1].iov_base, (const char*)iov_out[1].iov_base, iov_out[1].iov_len));
+
+  ceph_ll_close(cmount, fh);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, StripeUnitGran) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+  ASSERT_GT(ceph_get_stripe_unit_granularity(cmount), 0);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, Rename) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  int mypid = getpid();
+  char path_src[256];
+  char path_dst[256];
+
+  /* make a source file */
+  sprintf(path_src, "test_rename_src%d", mypid);
+  int fd = ceph_open(cmount, path_src, O_CREAT|O_TRUNC|O_WRONLY, 0777);
+  ASSERT_GT(fd, 0);
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+
+  /* rename to a new dest path */
+  sprintf(path_dst, "test_rename_dst%d", mypid);
+  ASSERT_EQ(0, ceph_rename(cmount, path_src, path_dst));
+
+  /* test that dest path exists */
+  struct ceph_statx stx;
+  ASSERT_EQ(0, ceph_statx(cmount, path_dst, &stx, 0, 0));
+
+  /* test that src path doesn't exist */
+  ASSERT_EQ(-ENOENT, ceph_statx(cmount, path_src, &stx, 0, AT_SYMLINK_NOFOLLOW));
+
+  /* rename with non-existent source path */
+  ASSERT_EQ(-ENOENT, ceph_rename(cmount, path_src, path_dst));
+
+  ASSERT_EQ(0, ceph_unlink(cmount, path_dst));
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, UseUnmounted) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+
+  struct statvfs stvfs;
+  EXPECT_EQ(-ENOTCONN, ceph_statfs(cmount, "/", &stvfs));
+  EXPECT_EQ(-ENOTCONN, ceph_get_local_osd(cmount));
+  EXPECT_EQ(-ENOTCONN, ceph_chdir(cmount, "/"));
+
+  struct ceph_dir_result *dirp;
+  EXPECT_EQ(-ENOTCONN, ceph_opendir(cmount, "/", &dirp));
+  EXPECT_EQ(-ENOTCONN, ceph_closedir(cmount, dirp));
+
+  ceph_readdir(cmount, dirp);
+  EXPECT_EQ(ENOTCONN, errno);
+
+  struct dirent rdent;
+  EXPECT_EQ(-ENOTCONN, ceph_readdir_r(cmount, dirp, &rdent));
+
+  struct ceph_statx stx;
+  EXPECT_EQ(-ENOTCONN, ceph_readdirplus_r(cmount, dirp, &rdent, &stx, 0, 0, NULL));
+  EXPECT_EQ(-ENOTCONN, ceph_getdents(cmount, dirp, NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_getdnames(cmount, dirp, NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_telldir(cmount, dirp));
+  EXPECT_EQ(-ENOTCONN, ceph_link(cmount, "/", "/link"));
+  EXPECT_EQ(-ENOTCONN, ceph_unlink(cmount, "/path"));
+  EXPECT_EQ(-ENOTCONN, ceph_rename(cmount, "/path", "/path"));
+  EXPECT_EQ(-ENOTCONN, ceph_mkdir(cmount, "/", 0655));
+  EXPECT_EQ(-ENOTCONN, ceph_mkdirs(cmount, "/", 0655));
+  EXPECT_EQ(-ENOTCONN, ceph_rmdir(cmount, "/path"));
+  EXPECT_EQ(-ENOTCONN, ceph_readlink(cmount, "/path", NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_symlink(cmount, "/path", "/path"));
+  EXPECT_EQ(-ENOTCONN, ceph_statx(cmount, "/path", &stx, 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_setattrx(cmount, "/path", &stx, 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_getxattr(cmount, "/path", "name", NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_lgetxattr(cmount, "/path", "name", NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_listxattr(cmount, "/path", NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_llistxattr(cmount, "/path", NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_removexattr(cmount, "/path", "name"));
+  EXPECT_EQ(-ENOTCONN, ceph_lremovexattr(cmount, "/path", "name"));
+  EXPECT_EQ(-ENOTCONN, ceph_setxattr(cmount, "/path", "name", NULL, 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_lsetxattr(cmount, "/path", "name", NULL, 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_fsetattrx(cmount, 0, &stx, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_chmod(cmount, "/path", 0));
+  EXPECT_EQ(-ENOTCONN, ceph_fchmod(cmount, 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_chown(cmount, "/path", 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_lchown(cmount, "/path", 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_fchown(cmount, 0, 0, 0));
+
+  struct utimbuf utb;
+  EXPECT_EQ(-ENOTCONN, ceph_utime(cmount, "/path", &utb));
+  EXPECT_EQ(-ENOTCONN, ceph_truncate(cmount, "/path", 0));
+  EXPECT_EQ(-ENOTCONN, ceph_mknod(cmount, "/path", 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_open(cmount, "/path", 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_open_layout(cmount, "/path", 0, 0, 0, 0, 0, "pool"));
+  EXPECT_EQ(-ENOTCONN, ceph_close(cmount, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_lseek(cmount, 0, 0, SEEK_SET));
+  EXPECT_EQ(-ENOTCONN, ceph_read(cmount, 0, NULL, 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_write(cmount, 0, NULL, 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_ftruncate(cmount, 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_fsync(cmount, 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_fstatx(cmount, 0, &stx, 0, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_sync_fs(cmount));
+  EXPECT_EQ(-ENOTCONN, ceph_get_file_stripe_unit(cmount, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_get_file_stripe_count(cmount, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_get_file_layout(cmount, 0, NULL, NULL ,NULL ,NULL));
+  EXPECT_EQ(-ENOTCONN, ceph_get_file_object_size(cmount, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_get_file_pool(cmount, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_get_file_pool_name(cmount, 0, NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_get_file_replication(cmount, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_get_path_replication(cmount, "/path"));
+  EXPECT_EQ(-ENOTCONN, ceph_get_path_layout(cmount, "/path", NULL, NULL, NULL, NULL));
+  EXPECT_EQ(-ENOTCONN, ceph_get_path_object_size(cmount, "/path"));
+  EXPECT_EQ(-ENOTCONN, ceph_get_path_stripe_count(cmount, "/path"));
+  EXPECT_EQ(-ENOTCONN, ceph_get_path_stripe_unit(cmount, "/path"));
+  EXPECT_EQ(-ENOTCONN, ceph_get_path_pool(cmount, "/path"));
+  EXPECT_EQ(-ENOTCONN, ceph_get_path_pool_name(cmount, "/path", NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_get_pool_name(cmount, 0, NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_get_file_stripe_address(cmount, 0, 0, NULL, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_localize_reads(cmount, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_debug_get_fd_caps(cmount, 0));
+  EXPECT_EQ(-ENOTCONN, ceph_debug_get_file_caps(cmount, "/path"));
+  EXPECT_EQ(-ENOTCONN, ceph_get_stripe_unit_granularity(cmount));
+  EXPECT_EQ(-ENOTCONN, ceph_get_pool_id(cmount, "data"));
+  EXPECT_EQ(-ENOTCONN, ceph_get_pool_replication(cmount, 1));
+
+  ceph_release(cmount);
+}
+
+TEST(LibCephFS, GetPoolId) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char name[80];
+  memset(name, 0, sizeof(name));
+  ASSERT_LE(0, ceph_get_path_pool_name(cmount, "/", name, sizeof(name)));
+  ASSERT_GE(ceph_get_pool_id(cmount, name), 0);
+  ASSERT_EQ(ceph_get_pool_id(cmount, "weflkjwelfjwlkejf"), -ENOENT);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, GetPoolReplication) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  /* negative pools */
+  ASSERT_EQ(ceph_get_pool_replication(cmount, -10), -ENOENT);
+
+  /* valid pool */
+  int pool_id;
+  int stripe_unit, stripe_count, object_size;
+  ASSERT_EQ(0, ceph_get_path_layout(cmount, "/", &stripe_unit, &stripe_count,
+				    &object_size, &pool_id));
+  ASSERT_GE(pool_id, 0);
+  ASSERT_GT(ceph_get_pool_replication(cmount, pool_id), 0);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, GetExtentOsds) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+
+  EXPECT_EQ(-ENOTCONN, ceph_get_file_extent_osds(cmount, 0, 0, NULL, NULL, 0));
+
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  int stripe_unit = (1<<18);
+
+  /* make a file! */
+  char test_file[256];
+  sprintf(test_file, "test_extent_osds_%d", getpid());
+  int fd = ceph_open_layout(cmount, test_file, O_CREAT|O_RDWR, 0666,
+      stripe_unit, 2, stripe_unit*2, NULL);
+  ASSERT_GT(fd, 0);
+
+  /* get back how many osds > 0 */
+  int ret = ceph_get_file_extent_osds(cmount, fd, 0, NULL, NULL, 0);
+  EXPECT_GT(ret, 0);
+
+  int64_t len;
+  int osds[ret];
+
+  /* full stripe extent */
+  EXPECT_EQ(ret, ceph_get_file_extent_osds(cmount, fd, 0, &len, osds, ret));
+  EXPECT_EQ(len, (int64_t)stripe_unit);
+
+  /* half stripe extent */
+  EXPECT_EQ(ret, ceph_get_file_extent_osds(cmount, fd, stripe_unit/2, &len, osds, ret));
+  EXPECT_EQ(len, (int64_t)stripe_unit/2);
+
+  /* 1.5 stripe unit offset -1 byte */
+  EXPECT_EQ(ret, ceph_get_file_extent_osds(cmount, fd, 3*stripe_unit/2-1, &len, osds, ret));
+  EXPECT_EQ(len, (int64_t)stripe_unit/2+1);
+
+  /* 1.5 stripe unit offset +1 byte */
+  EXPECT_EQ(ret, ceph_get_file_extent_osds(cmount, fd, 3*stripe_unit/2+1, &len, osds, ret));
+  EXPECT_EQ(len, (int64_t)stripe_unit/2-1);
+
+  /* only when more than 1 osd */
+  if (ret > 1) {
+    EXPECT_EQ(-ERANGE, ceph_get_file_extent_osds(cmount, fd, 0, NULL, osds, 1));
+  }
+
+  ceph_close(cmount, fd);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, GetOsdCrushLocation) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+
+  EXPECT_EQ(-ENOTCONN, ceph_get_osd_crush_location(cmount, 0, NULL, 0));
+
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  ASSERT_EQ(ceph_get_osd_crush_location(cmount, 0, NULL, 1), -EINVAL);
+
+  char path[256];
+  ASSERT_EQ(ceph_get_osd_crush_location(cmount, 9999999, path, 0), -ENOENT);
+  ASSERT_EQ(ceph_get_osd_crush_location(cmount, -1, path, 0), -EINVAL);
+
+  char test_file[256];
+  sprintf(test_file, "test_osds_loc_%d", getpid());
+  int fd = ceph_open(cmount, test_file, O_CREAT|O_RDWR, 0666);
+  ASSERT_GT(fd, 0);
+
+  /* get back how many osds > 0 */
+  int ret = ceph_get_file_extent_osds(cmount, fd, 0, NULL, NULL, 0);
+  EXPECT_GT(ret, 0);
+
+  /* full stripe extent */
+  int osds[ret];
+  EXPECT_EQ(ret, ceph_get_file_extent_osds(cmount, fd, 0, NULL, osds, ret));
+
+  ASSERT_GT(ceph_get_osd_crush_location(cmount, 0, path, 0), 0);
+  ASSERT_EQ(ceph_get_osd_crush_location(cmount, 0, path, 1), -ERANGE);
+
+  for (int i = 0; i < ret; i++) {
+    int len = ceph_get_osd_crush_location(cmount, osds[i], path, sizeof(path));
+    ASSERT_GT(len, 0);
+    int pos = 0;
+    while (pos < len) {
+      std::string type(path + pos);
+      ASSERT_GT((int)type.size(), 0);
+      pos += type.size() + 1;
+
+      std::string name(path + pos);
+      ASSERT_GT((int)name.size(), 0);
+      pos += name.size() + 1;
+    }
+  }
+
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, GetOsdAddr) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+
+  EXPECT_EQ(-ENOTCONN, ceph_get_osd_addr(cmount, 0, NULL));
+
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  ASSERT_EQ(-EINVAL, ceph_get_osd_addr(cmount, 0, NULL));
+
+  struct sockaddr_storage addr;
+  ASSERT_EQ(-ENOENT, ceph_get_osd_addr(cmount, -1, &addr));
+  ASSERT_EQ(-ENOENT, ceph_get_osd_addr(cmount, 9999999, &addr));
+
+  ASSERT_EQ(0, ceph_get_osd_addr(cmount, 0, &addr));
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, OpenNoClose) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  pid_t mypid = getpid();
+  char str_buf[256];
+  sprintf(str_buf, "open_no_close_dir%d", mypid);
+  ASSERT_EQ(0, ceph_mkdirs(cmount, str_buf, 0777));
+
+  struct ceph_dir_result *ls_dir = NULL;
+  ASSERT_EQ(ceph_opendir(cmount, str_buf, &ls_dir), 0);
+
+  sprintf(str_buf, "open_no_close_file%d", mypid);
+  int fd = ceph_open(cmount, str_buf, O_RDONLY|O_CREAT, 0666);
+  ASSERT_LT(0, fd);
+
+  // shutdown should force close opened file/dir
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, Nlink) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  Inode *root, *dir, *file;
+
+  ASSERT_EQ(ceph_ll_lookup_root(cmount, &root), 0);
+
+  char dirname[32], filename[32], linkname[32];
+  sprintf(dirname, "nlinkdir%x", getpid());
+  sprintf(filename, "nlinkorig%x", getpid());
+  sprintf(linkname, "nlinklink%x", getpid());
+
+  struct ceph_statx stx;
+  Fh *fh;
+  UserPerm *perms = ceph_mount_perms(cmount);
+
+  ASSERT_EQ(ceph_ll_mkdir(cmount, root, dirname, 0755, &dir, &stx, 0, 0, perms), 0);
+  ASSERT_EQ(ceph_ll_create(cmount, dir, filename, 0666, O_RDWR|O_CREAT|O_EXCL,
+			   &file, &fh, &stx, CEPH_STATX_NLINK, 0, perms), 0);
+  ASSERT_EQ(ceph_ll_close(cmount, fh), 0);
+  ASSERT_EQ(stx.stx_nlink, (nlink_t)1);
+
+  ASSERT_EQ(ceph_ll_link(cmount, file, dir, linkname, perms), 0);
+  ASSERT_EQ(ceph_ll_getattr(cmount, file, &stx, CEPH_STATX_NLINK, 0, perms), 0);
+  ASSERT_EQ(stx.stx_nlink, (nlink_t)2);
+
+  ASSERT_EQ(ceph_ll_unlink(cmount, dir, linkname, perms), 0);
+  ASSERT_EQ(ceph_ll_lookup(cmount, dir, filename, &file, &stx,
+			   CEPH_STATX_NLINK, 0, perms), 0);
+  ASSERT_EQ(stx.stx_nlink, (nlink_t)1);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, SlashDotDot) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  struct ceph_statx	stx;
+  ASSERT_EQ(ceph_statx(cmount, "/.", &stx, CEPH_STATX_INO, 0), 0);
+
+  ino_t ino = stx.stx_ino;
+  ASSERT_EQ(ceph_statx(cmount, "/..", &stx, CEPH_STATX_INO, 0), 0);
+
+  /* At root, "." and ".." should be the same inode */
+  ASSERT_EQ(ino, stx.stx_ino);
+
+  /* Test accessing the parent of an unlinked directory */
+  char dir1[32], dir2[56];
+  sprintf(dir1, "/sldotdot%x", getpid());
+  sprintf(dir2, "%s/sub%x", dir1, getpid());
+
+  ASSERT_EQ(ceph_mkdir(cmount, dir1, 0755), 0);
+  ASSERT_EQ(ceph_mkdir(cmount, dir2, 0755), 0);
+
+  ASSERT_EQ(ceph_chdir(cmount, dir2), 0);
+
+  /* Test behavior when unlinking cwd */
+  struct ceph_dir_result *rdir;
+  ASSERT_EQ(ceph_opendir(cmount, ".", &rdir), 0);
+  ASSERT_EQ(ceph_rmdir(cmount, dir2), 0);
+
+  /* get "." entry */
+  struct dirent *result = ceph_readdir(cmount, rdir);
+  ino = result->d_ino;
+
+  /* get ".." entry */
+  result = ceph_readdir(cmount, rdir);
+  ASSERT_EQ(ino, result->d_ino);
+  ceph_closedir(cmount, rdir);
+
+  /* Make sure it works same way when mounting subtree */
+  ASSERT_EQ(ceph_unmount(cmount), 0);
+  ASSERT_EQ(ceph_mount(cmount, dir1), 0);
+  ASSERT_EQ(ceph_statx(cmount, "/..", &stx, CEPH_STATX_INO, 0), 0);
+
+  /* Test readdir behavior */
+  ASSERT_EQ(ceph_opendir(cmount, "/", &rdir), 0);
+  result = ceph_readdir(cmount, rdir);
+  ASSERT_TRUE(result != NULL);
+  ASSERT_STREQ(result->d_name, ".");
+  ino = result->d_ino;
+  result = ceph_readdir(cmount, rdir);
+  ASSERT_TRUE(result != NULL);
+  ASSERT_STREQ(result->d_name, "..");
+  ASSERT_EQ(ino, result->d_ino);
+
+  ceph_shutdown(cmount);
+}
+
+static inline bool
+timespec_eq(timespec const& lhs, timespec const& rhs)
+{
+  return lhs.tv_sec == rhs.tv_sec && lhs.tv_nsec == rhs.tv_nsec;
+}
+
+TEST(LibCephFS, Btime) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  char filename[32];
+  sprintf(filename, "/getattrx%x", getpid());
+
+  ceph_unlink(cmount, filename);
+  int fd = ceph_open(cmount, filename, O_RDWR|O_CREAT|O_EXCL, 0666);
+  ASSERT_LT(0, fd);
+
+  /* make sure fstatx works */
+  struct ceph_statx	stx;
+
+  ASSERT_EQ(ceph_fstatx(cmount, fd, &stx, CEPH_STATX_CTIME|CEPH_STATX_BTIME, 0), 0);
+  ASSERT_TRUE(stx.stx_mask & (CEPH_STATX_CTIME|CEPH_STATX_BTIME));
+  ASSERT_TRUE(timespec_eq(stx.stx_ctime, stx.stx_btime));
+  ceph_close(cmount, fd);
+
+  ASSERT_EQ(ceph_statx(cmount, filename, &stx, CEPH_STATX_CTIME|CEPH_STATX_BTIME, 0), 0);
+  ASSERT_TRUE(timespec_eq(stx.stx_ctime, stx.stx_btime));
+  ASSERT_TRUE(stx.stx_mask & (CEPH_STATX_CTIME|CEPH_STATX_BTIME));
+
+  struct timespec old_btime = stx.stx_btime;
+
+  /* Now sleep, do a chmod and verify that the ctime changed, but btime didn't */
+  sleep(1);
+  ASSERT_EQ(ceph_chmod(cmount, filename, 0644), 0);
+  ASSERT_EQ(ceph_statx(cmount, filename, &stx, CEPH_STATX_CTIME|CEPH_STATX_BTIME, 0), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_BTIME);
+  ASSERT_TRUE(timespec_eq(stx.stx_btime, old_btime));
+  ASSERT_FALSE(timespec_eq(stx.stx_ctime, stx.stx_btime));
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, SetBtime) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  char filename[32];
+  sprintf(filename, "/setbtime%x", getpid());
+
+  ceph_unlink(cmount, filename);
+  int fd = ceph_open(cmount, filename, O_RDWR|O_CREAT|O_EXCL, 0666);
+  ASSERT_LT(0, fd);
+  ceph_close(cmount, fd);
+
+  struct ceph_statx stx;
+  struct timespec old_btime = { 1, 2 };
+
+  stx.stx_btime = old_btime;
+
+  ASSERT_EQ(ceph_setattrx(cmount, filename, &stx, CEPH_SETATTR_BTIME, 0), 0);
+
+  ASSERT_EQ(ceph_statx(cmount, filename, &stx, CEPH_STATX_BTIME, 0), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_BTIME);
+  ASSERT_TRUE(timespec_eq(stx.stx_btime, old_btime));
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, LazyStatx) {
+  struct ceph_mount_info *cmount1, *cmount2;
+  ASSERT_EQ(ceph_create(&cmount1, NULL), 0);
+  ASSERT_EQ(ceph_create(&cmount2, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount1, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount2, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount1, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount2, NULL));
+  ASSERT_EQ(ceph_mount(cmount1, "/"), 0);
+  ASSERT_EQ(ceph_mount(cmount2, "/"), 0);
+
+  char filename[32];
+  sprintf(filename, "lazystatx%x", getpid());
+
+  Inode *root1, *file1, *root2, *file2;
+  struct ceph_statx stx;
+  Fh *fh;
+  UserPerm *perms1 = ceph_mount_perms(cmount1);
+  UserPerm *perms2 = ceph_mount_perms(cmount2);
+
+  ASSERT_EQ(ceph_ll_lookup_root(cmount1, &root1), 0);
+  ceph_ll_unlink(cmount1, root1, filename, perms1);
+  ASSERT_EQ(ceph_ll_create(cmount1, root1, filename, 0666, O_RDWR|O_CREAT|O_EXCL,
+			   &file1, &fh, &stx, 0, 0, perms1), 0);
+  ASSERT_EQ(ceph_ll_close(cmount1, fh), 0);
+
+  ASSERT_EQ(ceph_ll_lookup_root(cmount2, &root2), 0);
+
+  ASSERT_EQ(ceph_ll_lookup(cmount2, root2, filename, &file2, &stx, CEPH_STATX_CTIME, 0, perms2), 0);
+
+  struct timespec old_ctime = stx.stx_ctime;
+
+  /*
+   * Now sleep, do a chmod on the first client and the see whether we get a
+   * different ctime with a statx that uses AT_NO_ATTR_SYNC
+   */
+  sleep(1);
+  stx.stx_mode = 0644;
+  ASSERT_EQ(ceph_ll_setattr(cmount1, file1, &stx, CEPH_SETATTR_MODE, perms1), 0);
+
+  ASSERT_EQ(ceph_ll_getattr(cmount2, file2, &stx, CEPH_STATX_CTIME, AT_NO_ATTR_SYNC, perms2), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_CTIME);
+  ASSERT_TRUE(stx.stx_ctime.tv_sec == old_ctime.tv_sec &&
+	      stx.stx_ctime.tv_nsec == old_ctime.tv_nsec);
+
+  ceph_shutdown(cmount1);
+  ceph_shutdown(cmount2);
+}
+
+TEST(LibCephFS, ChangeAttr) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  char filename[32];
+  sprintf(filename, "/changeattr%x", getpid());
+
+  ceph_unlink(cmount, filename);
+  int fd = ceph_open(cmount, filename, O_RDWR|O_CREAT|O_EXCL, 0666);
+  ASSERT_LT(0, fd);
+
+  struct ceph_statx	stx;
+  ASSERT_EQ(ceph_statx(cmount, filename, &stx, CEPH_STATX_VERSION, 0), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_VERSION);
+
+  uint64_t old_change_attr = stx.stx_version;
+
+  /* do chmod, and check whether change_attr changed */
+  ASSERT_EQ(ceph_chmod(cmount, filename, 0644), 0);
+  ASSERT_EQ(ceph_statx(cmount, filename, &stx, CEPH_STATX_VERSION, 0), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_VERSION);
+  ASSERT_NE(stx.stx_version, old_change_attr);
+  old_change_attr = stx.stx_version;
+
+  /* now do a write and see if it changed again */
+  ASSERT_EQ(3, ceph_write(cmount, fd, "foo", 3, 0));
+  ASSERT_EQ(ceph_statx(cmount, filename, &stx, CEPH_STATX_VERSION, 0), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_VERSION);
+  ASSERT_NE(stx.stx_version, old_change_attr);
+  old_change_attr = stx.stx_version;
+
+  /* Now truncate and check again */
+  ASSERT_EQ(0, ceph_ftruncate(cmount, fd, 0));
+  ASSERT_EQ(ceph_statx(cmount, filename, &stx, CEPH_STATX_VERSION, 0), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_VERSION);
+  ASSERT_NE(stx.stx_version, old_change_attr);
+
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, DirChangeAttr) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  char dirname[32], filename[56];
+  sprintf(dirname, "/dirchange%x", getpid());
+  sprintf(filename, "%s/foo", dirname);
+
+  ASSERT_EQ(ceph_mkdir(cmount, dirname, 0755), 0);
+
+  struct ceph_statx	stx;
+  ASSERT_EQ(ceph_statx(cmount, dirname, &stx, CEPH_STATX_VERSION, 0), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_VERSION);
+
+  uint64_t old_change_attr = stx.stx_version;
+
+  int fd = ceph_open(cmount, filename, O_RDWR|O_CREAT|O_EXCL, 0666);
+  ASSERT_LT(0, fd);
+  ceph_close(cmount, fd);
+
+  ASSERT_EQ(ceph_statx(cmount, dirname, &stx, CEPH_STATX_VERSION, 0), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_VERSION);
+  ASSERT_NE(stx.stx_version, old_change_attr);
+
+  old_change_attr = stx.stx_version;
+
+  ASSERT_EQ(ceph_unlink(cmount, filename), 0);
+  ASSERT_EQ(ceph_statx(cmount, dirname, &stx, CEPH_STATX_VERSION, 0), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_VERSION);
+  ASSERT_NE(stx.stx_version, old_change_attr);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, SetSize) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  char filename[32];
+  sprintf(filename, "/setsize%x", getpid());
+
+  ceph_unlink(cmount, filename);
+  int fd = ceph_open(cmount, filename, O_RDWR|O_CREAT|O_EXCL, 0666);
+  ASSERT_LT(0, fd);
+
+  struct ceph_statx stx;
+  uint64_t size = 8388608;
+  stx.stx_size = size;
+  ASSERT_EQ(ceph_fsetattrx(cmount, fd, &stx, CEPH_SETATTR_SIZE), 0);
+  ASSERT_EQ(ceph_fstatx(cmount, fd, &stx, CEPH_STATX_SIZE, 0), 0);
+  ASSERT_EQ(stx.stx_size, size);
+
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, ClearSetuid) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  Inode *root;
+  ASSERT_EQ(ceph_ll_lookup_root(cmount, &root), 0);
+
+  char filename[32];
+  sprintf(filename, "clearsetuid%x", getpid());
+
+  Fh *fh;
+  Inode *in;
+  struct ceph_statx stx;
+  const mode_t after_mode = S_IRWXU;
+  const mode_t before_mode = S_IRWXU | S_ISUID | S_ISGID;
+  const unsigned want = CEPH_STATX_UID|CEPH_STATX_GID|CEPH_STATX_MODE;
+  UserPerm *usercred = ceph_mount_perms(cmount);
+
+  ceph_ll_unlink(cmount, root, filename, usercred);
+  ASSERT_EQ(ceph_ll_create(cmount, root, filename, before_mode,
+			   O_RDWR|O_CREAT|O_EXCL, &in, &fh, &stx, want, 0,
+			   usercred), 0);
+
+  ASSERT_EQ(stx.stx_mode & (mode_t)ALLPERMS, before_mode);
+
+  // write
+  ASSERT_EQ(ceph_ll_write(cmount, fh, 0, 3, "foo"), 3);
+  ASSERT_EQ(ceph_ll_getattr(cmount, in, &stx, CEPH_STATX_MODE, 0, usercred), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_MODE);
+  ASSERT_EQ(stx.stx_mode & (mode_t)ALLPERMS, after_mode);
+
+  // reset mode
+  stx.stx_mode = before_mode;
+  ASSERT_EQ(ceph_ll_setattr(cmount, in, &stx, CEPH_STATX_MODE, usercred), 0);
+  ASSERT_EQ(ceph_ll_getattr(cmount, in, &stx, CEPH_STATX_MODE, 0, usercred), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_MODE);
+  ASSERT_EQ(stx.stx_mode & (mode_t)ALLPERMS, before_mode);
+
+  // truncate
+  stx.stx_size = 1;
+  ASSERT_EQ(ceph_ll_setattr(cmount, in, &stx, CEPH_SETATTR_SIZE, usercred), 0);
+  ASSERT_EQ(ceph_ll_getattr(cmount, in, &stx, CEPH_STATX_MODE, 0, usercred), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_MODE);
+  ASSERT_EQ(stx.stx_mode & (mode_t)ALLPERMS, after_mode);
+
+  // reset mode
+  stx.stx_mode = before_mode;
+  ASSERT_EQ(ceph_ll_setattr(cmount, in, &stx, CEPH_STATX_MODE, usercred), 0);
+  ASSERT_EQ(ceph_ll_getattr(cmount, in, &stx, CEPH_STATX_MODE, 0, usercred), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_MODE);
+  ASSERT_EQ(stx.stx_mode & (mode_t)ALLPERMS, before_mode);
+
+  // chown  -- for this we need to be "root"
+  UserPerm *rootcred = ceph_userperm_new(0, 0, 0, NULL);
+  ASSERT_TRUE(rootcred);
+  stx.stx_uid++;
+  stx.stx_gid++;
+  ASSERT_EQ(ceph_ll_setattr(cmount, in, &stx, CEPH_SETATTR_UID|CEPH_SETATTR_GID, rootcred), 0);
+  ASSERT_EQ(ceph_ll_getattr(cmount, in, &stx, CEPH_STATX_MODE, 0, usercred), 0);
+  ASSERT_TRUE(stx.stx_mask & CEPH_STATX_MODE);
+  ASSERT_EQ(stx.stx_mode & (mode_t)ALLPERMS, after_mode);
+
+  /* test chown with supplementary groups, and chown with/without exe bit */
+  uid_t u = 65534;
+  gid_t g = 65534;
+  gid_t gids[] = {65533,65532};
+  UserPerm *altcred = ceph_userperm_new(u, g, sizeof gids / sizeof gids[0], gids);
+  stx.stx_uid = u;
+  stx.stx_gid = g;
+  mode_t m = S_ISGID|S_ISUID|S_IRUSR|S_IWUSR;
+  stx.stx_mode = m;
+  ASSERT_EQ(ceph_ll_setattr(cmount, in, &stx, CEPH_STATX_MODE|CEPH_SETATTR_UID|CEPH_SETATTR_GID, rootcred), 0);
+  ASSERT_EQ(ceph_ll_getattr(cmount, in, &stx, CEPH_STATX_MODE, 0, altcred), 0);
+  ASSERT_EQ(stx.stx_mode&(mode_t)ALLPERMS, m);
+  /* not dropped without exe bit */
+  stx.stx_gid = gids[0];
+  ASSERT_EQ(ceph_ll_setattr(cmount, in, &stx, CEPH_SETATTR_GID, altcred), 0);
+  ASSERT_EQ(ceph_ll_getattr(cmount, in, &stx, CEPH_STATX_MODE, 0, altcred), 0);
+  ASSERT_EQ(stx.stx_mode&(mode_t)ALLPERMS, m);
+  /* now check dropped with exe bit */
+  m = S_ISGID|S_ISUID|S_IRWXU;
+  stx.stx_mode = m;
+  ASSERT_EQ(ceph_ll_setattr(cmount, in, &stx, CEPH_STATX_MODE, altcred), 0);
+  ASSERT_EQ(ceph_ll_getattr(cmount, in, &stx, CEPH_STATX_MODE, 0, altcred), 0);
+  ASSERT_EQ(stx.stx_mode&(mode_t)ALLPERMS, m);
+  stx.stx_gid = gids[1];
+  ASSERT_EQ(ceph_ll_setattr(cmount, in, &stx, CEPH_SETATTR_GID, altcred), 0);
+  ASSERT_EQ(ceph_ll_getattr(cmount, in, &stx, CEPH_STATX_MODE, 0, altcred), 0);
+  ASSERT_EQ(stx.stx_mode&(mode_t)ALLPERMS, m&(S_IRWXU|S_IRWXG|S_IRWXO));
+  ceph_userperm_destroy(altcred);
+
+  ASSERT_EQ(ceph_ll_close(cmount, fh), 0);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, OperationsOnRoot)
+{
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+
+  char dirname[32];
+  sprintf(dirname, "/somedir%x", getpid());
+
+  ASSERT_EQ(ceph_mkdir(cmount, dirname, 0755), 0);
+
+  ASSERT_EQ(ceph_rmdir(cmount, "/"), -EBUSY);
+
+  ASSERT_EQ(ceph_link(cmount, "/", "/"), -EEXIST);
+  ASSERT_EQ(ceph_link(cmount, dirname, "/"), -EEXIST);
+  ASSERT_EQ(ceph_link(cmount, "nonExisitingDir", "/"), -ENOENT);
+
+  ASSERT_EQ(ceph_unlink(cmount, "/"), -EISDIR);
+
+  ASSERT_EQ(ceph_rename(cmount, "/", "/"), -EBUSY);
+  ASSERT_EQ(ceph_rename(cmount, dirname, "/"), -EBUSY);
+  ASSERT_EQ(ceph_rename(cmount, "nonExistingDir", "/"), -EBUSY);
+  ASSERT_EQ(ceph_rename(cmount, "/", dirname), -EBUSY);
+  ASSERT_EQ(ceph_rename(cmount, "/", "nonExistingDir"), -EBUSY);
+
+  ASSERT_EQ(ceph_mkdir(cmount, "/", 0777), -EEXIST);
+
+  ASSERT_EQ(ceph_mknod(cmount, "/", 0, 0), -EEXIST);
+
+  ASSERT_EQ(ceph_symlink(cmount, "/", "/"), -EEXIST);
+  ASSERT_EQ(ceph_symlink(cmount, dirname, "/"), -EEXIST);
+  ASSERT_EQ(ceph_symlink(cmount, "nonExistingDir", "/"), -EEXIST);
+
+  ceph_shutdown(cmount);
+}
+
+static void shutdown_racer_func()
+{
+  const int niter = 32;
+  struct ceph_mount_info *cmount;
+  int i;
+
+  for (i = 0; i < niter; ++i) {
+    ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+    ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+    ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+    ASSERT_EQ(ceph_mount(cmount, "/"), 0);
+    ceph_shutdown(cmount);
+  }
+}
+
+// See tracker #20988
+TEST(LibCephFS, ShutdownRace)
+{
+  const int nthreads = 128;
+  std::thread threads[nthreads];
+
+  // Need a bunch of fd's for this test
+  struct rlimit rold, rnew;
+  ASSERT_EQ(getrlimit(RLIMIT_NOFILE, &rold), 0);
+  rnew = rold;
+  rnew.rlim_cur = rnew.rlim_max;
+  ASSERT_EQ(setrlimit(RLIMIT_NOFILE, &rnew), 0);
+
+  for (int i = 0; i < nthreads; ++i)
+    threads[i] = std::thread(shutdown_racer_func);
+
+  for (int i = 0; i < nthreads; ++i)
+    threads[i].join();
+  ASSERT_EQ(setrlimit(RLIMIT_NOFILE, &rold), 0);
+}
+
+static void get_current_time_utimbuf(struct utimbuf *utb)
+{
+  utime_t t = ceph_clock_now();
+  utb->actime = t.sec();
+  utb->modtime = t.sec();
+}
+
+static void get_current_time_timeval(struct timeval tv[2])
+{
+  utime_t t = ceph_clock_now();
+  t.copy_to_timeval(&tv[0]);
+  t.copy_to_timeval(&tv[1]);
+}
+
+static void get_current_time_timespec(struct timespec ts[2])
+{
+  utime_t t = ceph_clock_now();
+  t.to_timespec(&ts[0]);
+  t.to_timespec(&ts[1]);
+}
+
+TEST(LibCephFS, TestUtime) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_file[256];
+  sprintf(test_file, "test_utime_file_%d", getpid());
+  int fd = ceph_open(cmount, test_file, O_CREAT, 0666);
+  ASSERT_GT(fd, 0);
+
+  struct utimbuf utb;
+  struct ceph_statx stx;
+
+  get_current_time_utimbuf(&utb);
+
+  // ceph_utime()
+  EXPECT_EQ(0, ceph_utime(cmount, test_file, &utb));
+  ASSERT_EQ(ceph_statx(cmount, test_file, &stx,
+                       CEPH_STATX_MTIME|CEPH_STATX_ATIME, 0), 0);
+  ASSERT_EQ(utime_t(stx.stx_atime), utime_t(utb.actime, 0));
+  ASSERT_EQ(utime_t(stx.stx_mtime), utime_t(utb.modtime, 0));
+
+  get_current_time_utimbuf(&utb);
+
+  // ceph_futime()
+  EXPECT_EQ(0, ceph_futime(cmount, fd, &utb));
+  ASSERT_EQ(ceph_statx(cmount, test_file, &stx,
+                       CEPH_STATX_MTIME|CEPH_STATX_ATIME, 0), 0);
+  ASSERT_EQ(utime_t(stx.stx_atime), utime_t(utb.actime, 0));
+  ASSERT_EQ(utime_t(stx.stx_mtime), utime_t(utb.modtime, 0));
+
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, TestUtimes) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_file[256];
+  char test_symlink[256];
+
+  sprintf(test_file, "test_utimes_file_%d", getpid());
+  sprintf(test_symlink, "test_utimes_symlink_%d", getpid());
+  int fd = ceph_open(cmount, test_file, O_CREAT, 0666);
+  ASSERT_GT(fd, 0);
+
+  ASSERT_EQ(ceph_symlink(cmount, test_file, test_symlink), 0);
+
+  struct timeval times[2];
+  struct ceph_statx stx;
+
+  get_current_time_timeval(times);
+
+  // ceph_utimes() on symlink, validate target file time
+  EXPECT_EQ(0, ceph_utimes(cmount, test_symlink, times));
+  ASSERT_EQ(ceph_statx(cmount, test_symlink, &stx,
+                       CEPH_STATX_MTIME|CEPH_STATX_ATIME, 0), 0);
+  ASSERT_EQ(utime_t(stx.stx_atime), utime_t(times[0]));
+  ASSERT_EQ(utime_t(stx.stx_mtime), utime_t(times[1]));
+
+  get_current_time_timeval(times);
+
+  // ceph_lutimes() on symlink, validate symlink time
+  EXPECT_EQ(0, ceph_lutimes(cmount, test_symlink, times));
+  ASSERT_EQ(ceph_statx(cmount, test_symlink, &stx,
+                       CEPH_STATX_MTIME|CEPH_STATX_ATIME, AT_SYMLINK_NOFOLLOW), 0);
+  ASSERT_EQ(utime_t(stx.stx_atime), utime_t(times[0]));
+  ASSERT_EQ(utime_t(stx.stx_mtime), utime_t(times[1]));
+
+  get_current_time_timeval(times);
+
+  // ceph_futimes()
+  EXPECT_EQ(0, ceph_futimes(cmount, fd, times));
+  ASSERT_EQ(ceph_statx(cmount, test_file, &stx,
+                       CEPH_STATX_MTIME|CEPH_STATX_ATIME, 0), 0);
+  ASSERT_EQ(utime_t(stx.stx_atime), utime_t(times[0]));
+  ASSERT_EQ(utime_t(stx.stx_mtime), utime_t(times[1]));
+
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, TestFutimens) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_file[256];
+
+  sprintf(test_file, "test_futimens_file_%d", getpid());
+  int fd = ceph_open(cmount, test_file, O_CREAT, 0666);
+  ASSERT_GT(fd, 0);
+
+  struct timespec times[2];
+  struct ceph_statx stx;
+
+  get_current_time_timespec(times);
+
+  // ceph_futimens()
+  EXPECT_EQ(0, ceph_futimens(cmount, fd, times));
+  ASSERT_EQ(ceph_statx(cmount, test_file, &stx,
+                       CEPH_STATX_MTIME|CEPH_STATX_ATIME, 0), 0);
+  ASSERT_EQ(utime_t(stx.stx_atime), utime_t(times[0]));
+  ASSERT_EQ(utime_t(stx.stx_mtime), utime_t(times[1]));
+
+  ceph_close(cmount, fd);
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, OperationsOnDotDot) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char c_dir[512], c_dir_dot[1024], c_dir_dotdot[1024];
+  char c_non_existent_dir[1024], c_non_existent_dirs[1024];
+  char c_temp[1024];
+
+  pid_t mypid = getpid();
+  sprintf(c_dir, "/oodd_dir_%d", mypid);
+  sprintf(c_dir_dot, "/%s/.", c_dir);
+  sprintf(c_dir_dotdot, "/%s/..", c_dir);
+  sprintf(c_non_existent_dir, "/%s/../oodd_nonexistent/..", c_dir);
+  sprintf(c_non_existent_dirs,
+          "/%s/../ood_nonexistent1_%d/oodd_nonexistent2_%d", c_dir, mypid, mypid);
+  sprintf(c_temp, "/oodd_temp_%d", mypid);
+
+  ASSERT_EQ(0, ceph_mkdir(cmount, c_dir, 0777));
+  ASSERT_EQ(-EEXIST, ceph_mkdir(cmount, c_dir_dot, 0777));
+  ASSERT_EQ(-EEXIST, ceph_mkdir(cmount, c_dir_dotdot, 0777));
+  ASSERT_EQ(0, ceph_mkdirs(cmount, c_non_existent_dirs, 0777));
+
+  ASSERT_EQ(-ENOTEMPTY, ceph_rmdir(cmount, c_dir_dot));
+  ASSERT_EQ(-ENOTEMPTY, ceph_rmdir(cmount, c_dir_dotdot));
+  // non existent directory should return -ENOENT
+  ASSERT_EQ(-ENOENT, ceph_rmdir(cmount, c_non_existent_dir));
+
+  ASSERT_EQ(-EBUSY, ceph_rename(cmount, c_dir_dot, c_temp));
+  ASSERT_EQ(0, ceph_chdir(cmount, c_dir));
+  ASSERT_EQ(0, ceph_mkdir(cmount, c_temp, 0777));
+  ASSERT_EQ(-EBUSY, ceph_rename(cmount, c_temp, ".."));
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, SnapXattrs) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_snap_xattr_file[256];
+  char c_temp[PATH_MAX];
+  char gxattrv[128];
+  char gxattrv2[128];
+  int xbuflen = sizeof(gxattrv);
+  pid_t mypid = getpid();
+
+  sprintf(test_snap_xattr_file, "test_snap_xattr_%d", mypid);
+  int fd = ceph_open(cmount, test_snap_xattr_file, O_CREAT, 0666);
+  ASSERT_GT(fd, 0);
+  ceph_close(cmount, fd);
+
+  sprintf(c_temp, "/.snap/test_snap_xattr_snap_%d", mypid);
+  ASSERT_EQ(0, ceph_mkdir(cmount, c_temp, 0777));
+
+  int alen = ceph_getxattr(cmount, c_temp, "ceph.snap.btime", (void *)gxattrv, xbuflen);
+  // xattr value is secs.nsecs (don't assume zero-term)
+  ASSERT_LT(0, alen);
+  ASSERT_LT(alen, xbuflen);
+  gxattrv[alen] = '\0';
+  char *s = strchrnul(gxattrv, '.');
+  ASSERT_LT(s, gxattrv + alen);
+  ASSERT_EQ('.', *s);
+  *s = '\0';
+  utime_t btime = utime_t(strtoull(gxattrv, NULL, 10), strtoull(s + 1, NULL, 10));
+  *s = '.';  // restore for later strcmp
+
+  // file within the snapshot should carry the same btime
+  sprintf(c_temp, "/.snap/test_snap_xattr_snap_%d/%s", mypid, test_snap_xattr_file);
+
+  int alen2 = ceph_getxattr(cmount, c_temp, "ceph.snap.btime", (void *)gxattrv2, xbuflen);
+  ASSERT_EQ(alen, alen2);
+  ASSERT_EQ(0, strncmp(gxattrv, gxattrv2, alen));
+
+  // non-snap file shouldn't carry the xattr
+  alen = ceph_getxattr(cmount, test_snap_xattr_file, "ceph.snap.btime", (void *)gxattrv2, xbuflen);
+  ASSERT_EQ(-ENODATA, alen);
+
+  // create a second snapshot
+  sprintf(c_temp, "/.snap/test_snap_xattr_snap2_%d", mypid);
+  ASSERT_EQ(0, ceph_mkdir(cmount, c_temp, 0777));
+
+  // check that the btime for the newer snapshot is > older
+  alen = ceph_getxattr(cmount, c_temp, "ceph.snap.btime", (void *)gxattrv2, xbuflen);
+  ASSERT_LT(0, alen);
+  ASSERT_LT(alen, xbuflen);
+  gxattrv2[alen] = '\0';
+  s = strchrnul(gxattrv2, '.');
+  ASSERT_LT(s, gxattrv2 + alen);
+  ASSERT_EQ('.', *s);
+  *s = '\0';
+  utime_t new_btime = utime_t(strtoull(gxattrv2, NULL, 10), strtoull(s + 1, NULL, 10));
+  ASSERT_LT(btime, new_btime);
+
+  // check that the snap.btime vxattr appears in listxattr()
+  char xattrlist[512];
+  int len = ceph_listxattr(cmount, c_temp, xattrlist, sizeof(xattrlist));
+  ASSERT_GT(len, 0);
+  ASSERT_GE(sizeof(xattrlist), (size_t)len);
+  char *p = xattrlist;
+  int found = 0;
+  while (len > 0) {
+    if (strcmp(p, "ceph.snap.btime") == 0)
+      found++;
+    len -= strlen(p) + 1;
+    p += strlen(p) + 1;
+  }
+  ASSERT_EQ(found, 1);
+
+  // listxattr() shouldn't return snap.btime vxattr for non-snaps
+  len = ceph_listxattr(cmount, test_snap_xattr_file, xattrlist, sizeof(xattrlist));
+  ASSERT_GE(sizeof(xattrlist), (size_t)len);
+  p = xattrlist;
+  found = 0;
+  while (len > 0) {
+    if (strcmp(p, "ceph.snap.btime") == 0)
+      found++;
+    len -= strlen(p) + 1;
+    p += strlen(p) + 1;
+  }
+  ASSERT_EQ(found, 0);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, SnapQuota) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(ceph_create(&cmount, NULL), 0);
+  ASSERT_EQ(ceph_conf_read_file(cmount, NULL), 0);
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(ceph_mount(cmount, NULL), 0);
+
+  char test_snap_dir_quota_xattr[256];
+  char test_snap_subdir_quota_xattr[256];
+  char test_snap_subdir_noquota_xattr[256];
+  char xattrk[128];
+  char xattrv[128];
+  char c_temp[PATH_MAX];
+  char gxattrv[128];
+  int xbuflen = sizeof(gxattrv);
+  pid_t mypid = getpid();
+
+  // create dir and set quota
+  sprintf(test_snap_dir_quota_xattr, "test_snap_dir_quota_xattr_%d", mypid);
+  ASSERT_EQ(0, ceph_mkdir(cmount, test_snap_dir_quota_xattr, 0777));
+
+  sprintf(xattrk, "ceph.quota.max_bytes");
+  sprintf(xattrv, "65536");
+  ASSERT_EQ(0, ceph_setxattr(cmount, test_snap_dir_quota_xattr, xattrk, (void *)xattrv, 5, XATTR_CREATE));
+
+  // create subdir and set quota
+  sprintf(test_snap_subdir_quota_xattr, "test_snap_dir_quota_xattr_%d/subdir_quota", mypid);
+  ASSERT_EQ(0, ceph_mkdirs(cmount, test_snap_subdir_quota_xattr, 0777));
+
+  sprintf(xattrk, "ceph.quota.max_bytes");
+  sprintf(xattrv, "32768");
+  ASSERT_EQ(0, ceph_setxattr(cmount, test_snap_subdir_quota_xattr, xattrk, (void *)xattrv, 5, XATTR_CREATE));
+
+  // create subdir with no quota
+  sprintf(test_snap_subdir_noquota_xattr, "test_snap_dir_quota_xattr_%d/subdir_noquota", mypid);
+  ASSERT_EQ(0, ceph_mkdirs(cmount, test_snap_subdir_noquota_xattr, 0777));
+
+  // snapshot dir
+  sprintf(c_temp, "/.snap/test_snap_dir_quota_xattr_snap_%d", mypid);
+  ASSERT_EQ(0, ceph_mkdirs(cmount, c_temp, 0777));
+
+  // check dir quota under snap
+  sprintf(c_temp, "/.snap/test_snap_dir_quota_xattr_snap_%d/test_snap_dir_quota_xattr_%d", mypid, mypid);
+  int alen = ceph_getxattr(cmount, c_temp, "ceph.quota.max_bytes", (void *)gxattrv, xbuflen);
+  ASSERT_LT(0, alen);
+  ASSERT_LT(alen, xbuflen);
+  gxattrv[alen] = '\0';
+  ASSERT_STREQ(gxattrv, "65536");
+
+  // check subdir quota under snap
+  sprintf(c_temp, "/.snap/test_snap_dir_quota_xattr_snap_%d/test_snap_dir_quota_xattr_%d/subdir_quota", mypid, mypid);
+  alen = ceph_getxattr(cmount, c_temp, "ceph.quota.max_bytes", (void *)gxattrv, xbuflen);
+  ASSERT_LT(0, alen);
+  ASSERT_LT(alen, xbuflen);
+  gxattrv[alen] = '\0';
+  ASSERT_STREQ(gxattrv, "32768");
+
+  // ensure subdir noquota xattr under snap
+  sprintf(c_temp, "/.snap/test_snap_dir_quota_xattr_snap_%d/test_snap_dir_quota_xattr_%d/subdir_noquota", mypid, mypid);
+  EXPECT_EQ(-ENODATA, ceph_getxattr(cmount, c_temp, "ceph.quota.max_bytes", (void *)gxattrv, xbuflen));
+
+  // listxattr() shouldn't return ceph.quota.max_bytes vxattr
+  sprintf(c_temp, "/.snap/test_snap_dir_quota_xattr_snap_%d/test_snap_dir_quota_xattr_%d", mypid, mypid);
+  char xattrlist[512];
+  int len = ceph_listxattr(cmount, c_temp, xattrlist, sizeof(xattrlist));
+  ASSERT_GE(sizeof(xattrlist), (size_t)len);
+  char *p = xattrlist;
+  int found = 0;
+  while (len > 0) {
+    if (strcmp(p, "ceph.quota.max_bytes") == 0)
+      found++;
+    len -= strlen(p) + 1;
+    p += strlen(p) + 1;
+  }
+  ASSERT_EQ(found, 0);
+
+  ceph_shutdown(cmount);
+}
+
+TEST(LibCephFS, Lseek) {
+  struct ceph_mount_info *cmount;
+  ASSERT_EQ(0, ceph_create(&cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));
+  ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));
+  ASSERT_EQ(0, ceph_mount(cmount, "/"));
+
+  char c_path[1024];
+  sprintf(c_path, "test_lseek_%d", getpid());
+  int fd = ceph_open(cmount, c_path, O_RDWR|O_CREAT|O_TRUNC, 0666);
+  ASSERT_LT(0, fd);
+
+  const char *out_buf = "hello world";
+  size_t size = strlen(out_buf);
+  ASSERT_EQ(ceph_write(cmount, fd, out_buf, size, 0), (int)size);
+
+  /* basic SEEK_SET/END/CUR tests */
+  ASSERT_EQ(0, ceph_lseek(cmount, fd, 0, SEEK_SET));
+  ASSERT_EQ(size, ceph_lseek(cmount, fd, 0, SEEK_END));
+  ASSERT_EQ(0, ceph_lseek(cmount, fd, -size, SEEK_CUR));
+
+  /* Test basic functionality and out of bounds conditions for SEEK_HOLE/DATA */
+#ifdef SEEK_HOLE
+  ASSERT_EQ(size, ceph_lseek(cmount, fd, 0, SEEK_HOLE));
+  ASSERT_EQ(-ENXIO, ceph_lseek(cmount, fd, -1, SEEK_HOLE));
+  ASSERT_EQ(-ENXIO, ceph_lseek(cmount, fd, size + 1, SEEK_HOLE));
+#endif
+#ifdef SEEK_DATA
+  ASSERT_EQ(0, ceph_lseek(cmount, fd, 0, SEEK_DATA));
+  ASSERT_EQ(-ENXIO, ceph_lseek(cmount, fd, -1, SEEK_DATA));
+  ASSERT_EQ(-ENXIO, ceph_lseek(cmount, fd, size + 1, SEEK_DATA));
+#endif
+
+  ASSERT_EQ(0, ceph_close(cmount, fd));
+  ceph_shutdown(cmount);
+}