Adding upstream version 18.2.2.upstream/18.2.2

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

5 files changed, 4958 insertions, 0 deletions

diff --git a/src/test/librbd/io/test_mock_CopyupRequest.cc b/src/test/librbd/io/test_mock_CopyupRequest.cc
new file mode 100644
index 000000000..a4fe54af2
--- /dev/null
+++ b/src/test/librbd/io/test_mock_CopyupRequest.cc
@@ -0,0 +1,1340 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/librbd/test_mock_fixture.h"
+#include "test/librbd/test_support.h"
+#include "test/librbd/mock/MockExclusiveLock.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "test/librbd/mock/MockJournal.h"
+#include "test/librbd/mock/MockObjectMap.h"
+#include "test/librados_test_stub/MockTestMemIoCtxImpl.h"
+#include "test/librados_test_stub/MockTestMemRadosClient.h"
+#include "include/rbd/librbd.hpp"
+#include "librbd/api/Io.h"
+#include "librbd/deep_copy/ObjectCopyRequest.h"
+#include "librbd/io/CopyupRequest.h"
+#include "librbd/io/ImageDispatchSpec.h"
+#include "librbd/io/ObjectRequest.h"
+#include "librbd/io/ReadResult.h"
+#include "librbd/io/Utils.h"
+
+namespace librbd {
+namespace {
+
+struct MockTestImageCtx : public MockImageCtx {
+  MockTestImageCtx(ImageCtx &image_ctx,
+                   MockTestImageCtx* mock_parent_image_ctx = nullptr)
+    : MockImageCtx(image_ctx) {
+    parent = mock_parent_image_ctx;
+  }
+  ~MockTestImageCtx() override {
+    // copyups need to complete prior to attempting to delete this object
+    wait_for_async_ops();
+  }
+
+  std::map<uint64_t, librbd::io::CopyupRequest<librbd::MockTestImageCtx>*> copyup_list;
+};
+
+} // anonymous namespace
+
+namespace util {
+
+inline ImageCtx *get_image_ctx(MockImageCtx *image_ctx) {
+  return image_ctx->image_ctx;
+}
+
+} // namespace util
+
+namespace deep_copy {
+
+template <>
+struct ObjectCopyRequest<librbd::MockTestImageCtx> {
+  static ObjectCopyRequest* s_instance;
+  static ObjectCopyRequest* create(librbd::MockImageCtx* parent_image_ctx,
+                                   librbd::MockTestImageCtx* image_ctx,
+                                   librados::snap_t src_snap_id_start,
+                                   librados::snap_t dst_snap_id_start,
+                                   const SnapMap &snap_map,
+                                   uint64_t object_number, uint32_t flags,
+                                   Handler*, Context *on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->object_number = object_number;
+    s_instance->flatten = (
+      (flags & deep_copy::OBJECT_COPY_REQUEST_FLAG_FLATTEN) != 0);
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  uint64_t object_number;
+  bool flatten;
+  Context *on_finish;
+
+  ObjectCopyRequest() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+ObjectCopyRequest<librbd::MockTestImageCtx>* ObjectCopyRequest<librbd::MockTestImageCtx>::s_instance = nullptr;
+
+} // namespace deep_copy
+
+namespace io {
+
+namespace util {
+
+template <>
+void area_to_object_extents(MockTestImageCtx* image_ctx, uint64_t offset,
+                            uint64_t length, ImageArea area,
+                            uint64_t buffer_offset,
+                            striper::LightweightObjectExtents* object_extents) {
+  Striper::file_to_extents(image_ctx->cct, &image_ctx->layout, offset, length,
+                           0, buffer_offset, object_extents);
+}
+
+template <>
+std::pair<Extents, ImageArea> object_to_area_extents(
+    MockTestImageCtx* image_ctx, uint64_t object_no,
+    const Extents& object_extents) {
+  Extents extents;
+  for (auto [off, len] : object_extents) {
+    Striper::extent_to_file(image_ctx->cct, &image_ctx->layout, object_no, off,
+                            len, extents);
+  }
+  return {std::move(extents), ImageArea::DATA};
+}
+
+} // namespace util
+
+template <>
+struct ObjectRequest<librbd::MockTestImageCtx> {
+  static void add_write_hint(librbd::MockTestImageCtx&,
+                             neorados::WriteOp*) {
+  }
+};
+
+template <>
+struct AbstractObjectWriteRequest<librbd::MockTestImageCtx> {
+  C_SaferCond ctx;
+  void handle_copyup(int r) {
+    ctx.complete(r);
+  }
+
+  MOCK_CONST_METHOD0(get_pre_write_object_map_state, uint8_t());
+  MOCK_CONST_METHOD0(is_empty_write_op, bool());
+
+  MOCK_METHOD1(add_copyup_ops, void(neorados::WriteOp*));
+};
+
+} // namespace io
+} // namespace librbd
+
+static bool operator==(const SnapContext& rhs, const SnapContext& lhs) {
+  return (rhs.seq == lhs.seq && rhs.snaps == lhs.snaps);
+}
+
+#include "librbd/AsyncObjectThrottle.cc"
+#include "librbd/io/CopyupRequest.cc"
+
+MATCHER_P(IsRead, image_extents, "") {
+  auto req = boost::get<librbd::io::ImageDispatchSpec::Read>(&arg->request);
+  return (req != nullptr && image_extents == arg->image_extents);
+}
+
+namespace librbd {
+namespace io {
+
+using ::testing::_;
+using ::testing::InSequence;
+using ::testing::Invoke;
+using ::testing::Return;
+using ::testing::StrEq;
+using ::testing::WithArg;
+using ::testing::WithArgs;
+using ::testing::WithoutArgs;
+
+struct TestMockIoCopyupRequest : public TestMockFixture {
+  typedef CopyupRequest<librbd::MockTestImageCtx> MockCopyupRequest;
+  typedef ObjectRequest<librbd::MockTestImageCtx> MockObjectRequest;
+  typedef AbstractObjectWriteRequest<librbd::MockTestImageCtx> MockAbstractObjectWriteRequest;
+  typedef deep_copy::ObjectCopyRequest<librbd::MockTestImageCtx> MockObjectCopyRequest;
+
+  void SetUp() override {
+    TestMockFixture::SetUp();
+    if (!is_feature_enabled(RBD_FEATURE_LAYERING)) {
+      return;
+    }
+
+    m_parent_image_name = m_image_name;
+    m_image_name = get_temp_image_name();
+
+    librbd::Image image;
+    librbd::RBD rbd;
+    ASSERT_EQ(0, rbd.open(m_ioctx, image, m_parent_image_name.c_str(),
+                          nullptr));
+    ASSERT_EQ(0, image.snap_create("one"));
+    ASSERT_EQ(0, image.snap_protect("one"));
+
+    uint64_t features;
+    ASSERT_EQ(0, image.features(&features));
+    image.close();
+
+    int order = 0;
+    ASSERT_EQ(0, rbd.clone(m_ioctx, m_parent_image_name.c_str(), "one", m_ioctx,
+                           m_image_name.c_str(), features, &order));
+  }
+
+  void expect_get_parent_overlap(MockTestImageCtx& mock_image_ctx,
+                                 librados::snap_t snap_id, uint64_t overlap,
+                                 int r) {
+    EXPECT_CALL(mock_image_ctx, get_parent_overlap(snap_id, _))
+      .WillOnce(WithArg<1>(Invoke([overlap, r](uint64_t *o) {
+                             *o = overlap;
+                             return r;
+                           })));
+  }
+
+  void expect_prune_parent_extents(MockTestImageCtx& mock_image_ctx,
+                                   uint64_t overlap, uint64_t object_overlap) {
+    EXPECT_CALL(mock_image_ctx, prune_parent_extents(_, _, overlap, _))
+      .WillOnce(WithoutArgs(Invoke([object_overlap]() {
+                              return object_overlap;
+                            })));
+  }
+
+  void expect_read_parent(librbd::MockTestImageCtx& mock_image_ctx,
+                          const Extents& image_extents,
+                          const std::string& data, int r) {
+    EXPECT_CALL(*mock_image_ctx.io_image_dispatcher,
+                send(IsRead(image_extents)))
+      .WillOnce(Invoke(
+        [&mock_image_ctx, image_extents, data, r](io::ImageDispatchSpec* spec) {
+          auto req = boost::get<librbd::io::ImageDispatchSpec::Read>(
+            &spec->request);
+          ASSERT_TRUE(req != nullptr);
+
+          if (r < 0) {
+            spec->fail(r);
+            return;
+          }
+
+          spec->dispatch_result = DISPATCH_RESULT_COMPLETE;
+
+          auto aio_comp = spec->aio_comp;
+          aio_comp->read_result = std::move(req->read_result);
+          aio_comp->read_result.set_image_extents(image_extents);
+          aio_comp->set_request_count(1);
+          auto ctx = new ReadResult::C_ImageReadRequest(aio_comp, 0,
+                                                        image_extents);
+          ctx->bl.append(data);
+          mock_image_ctx.image_ctx->op_work_queue->queue(ctx, r);
+        }));
+  }
+
+  void expect_copyup(MockTestImageCtx& mock_image_ctx, uint64_t snap_id,
+                     const std::string& oid, const std::string& data, int r) {
+    bufferlist in_bl;
+    in_bl.append(data);
+
+    SnapContext snapc;
+    if (snap_id == CEPH_NOSNAP) {
+      snapc = mock_image_ctx.snapc;
+    }
+
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    EXPECT_CALL(mock_io_ctx,
+                exec(oid, _, StrEq("rbd"), StrEq("copyup"),
+                     ContentsEqual(in_bl), _, _, snapc))
+      .WillOnce(Return(r));
+  }
+
+  void expect_sparse_copyup(MockTestImageCtx &mock_image_ctx, uint64_t snap_id,
+                            const std::string &oid,
+                            const std::map<uint64_t, uint64_t> &extent_map,
+                            const std::string &data, int r) {
+    bufferlist data_bl;
+    data_bl.append(data);
+
+    bufferlist in_bl;
+    encode(extent_map, in_bl);
+    encode(data_bl, in_bl);
+
+    SnapContext snapc;
+    if (snap_id == CEPH_NOSNAP) {
+      snapc = mock_image_ctx.snapc;
+    }
+
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    EXPECT_CALL(mock_io_ctx,
+                exec(oid, _, StrEq("rbd"), StrEq("sparse_copyup"),
+                     ContentsEqual(in_bl), _, _, snapc))
+      .WillOnce(Return(r));
+  }
+
+  void expect_write(MockTestImageCtx& mock_image_ctx, uint64_t snap_id,
+                    const std::string& oid, int r) {
+    SnapContext snapc;
+    if (snap_id == CEPH_NOSNAP) {
+      snapc = mock_image_ctx.snapc;
+    }
+
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    EXPECT_CALL(mock_io_ctx, write(oid, _, 0, 0, snapc))
+      .WillOnce(Return(r));
+  }
+
+  void expect_test_features(MockTestImageCtx& mock_image_ctx) {
+    EXPECT_CALL(mock_image_ctx, test_features(_, _))
+      .WillRepeatedly(WithArg<0>(Invoke([&mock_image_ctx](uint64_t features) {
+              return (mock_image_ctx.features & features) != 0;
+            })));
+  }
+
+  void expect_is_lock_owner(MockTestImageCtx& mock_image_ctx) {
+    if (mock_image_ctx.exclusive_lock != nullptr) {
+      EXPECT_CALL(*mock_image_ctx.exclusive_lock,
+                  is_lock_owner()).WillRepeatedly(Return(true));
+    }
+  }
+
+  void expect_is_empty_write_op(MockAbstractObjectWriteRequest& mock_write_request,
+                                bool is_empty) {
+    EXPECT_CALL(mock_write_request, is_empty_write_op())
+      .WillOnce(Return(is_empty));
+  }
+
+  void expect_add_copyup_ops(MockAbstractObjectWriteRequest& mock_write_request) {
+    EXPECT_CALL(mock_write_request, add_copyup_ops(_))
+      .WillOnce(Invoke([](neorados::WriteOp* op) {
+                  op->write(0, bufferlist{});
+                }));
+  }
+
+  void expect_get_pre_write_object_map_state(MockTestImageCtx& mock_image_ctx,
+                                             MockAbstractObjectWriteRequest& mock_write_request,
+                                             uint8_t state) {
+    if (mock_image_ctx.object_map != nullptr) {
+      EXPECT_CALL(mock_write_request, get_pre_write_object_map_state())
+        .WillOnce(Return(state));
+    }
+  }
+
+  void expect_object_map_at(MockTestImageCtx& mock_image_ctx,
+                            uint64_t object_no, uint8_t state) {
+    if (mock_image_ctx.object_map != nullptr) {
+      EXPECT_CALL(*mock_image_ctx.object_map, at(object_no))
+        .WillOnce(Return(state));
+    }
+  }
+
+  void expect_object_map_update(MockTestImageCtx& mock_image_ctx,
+                                uint64_t snap_id, uint64_t object_no,
+                                uint8_t state, bool updated, int ret_val) {
+    if (mock_image_ctx.object_map != nullptr) {
+      if (!mock_image_ctx.image_ctx->test_features(RBD_FEATURE_FAST_DIFF) &&
+          state == OBJECT_EXISTS_CLEAN) {
+        state = OBJECT_EXISTS;
+      }
+
+      EXPECT_CALL(*mock_image_ctx.object_map,
+                  aio_update(snap_id, object_no, object_no + 1, state,
+                             boost::optional<uint8_t>(), _,
+                             (snap_id != CEPH_NOSNAP), _))
+        .WillOnce(WithArg<7>(Invoke([&mock_image_ctx, updated, ret_val](Context *ctx) {
+                               if (updated) {
+                                 mock_image_ctx.op_work_queue->queue(ctx, ret_val);
+                               }
+                               return updated;
+                             })));
+    }
+  }
+
+  void expect_object_copy(MockTestImageCtx& mock_image_ctx,
+                          MockObjectCopyRequest& mock_object_copy_request,
+                          bool flatten, int r) {
+    EXPECT_CALL(mock_object_copy_request, send())
+      .WillOnce(Invoke(
+        [&mock_image_ctx, &mock_object_copy_request, flatten, r]() {
+          ASSERT_EQ(flatten, mock_object_copy_request.flatten);
+          mock_image_ctx.op_work_queue->queue(
+            mock_object_copy_request.on_finish, r);
+        }));
+  }
+
+  void expect_prepare_copyup(MockTestImageCtx& mock_image_ctx, int r = 0) {
+    EXPECT_CALL(*mock_image_ctx.io_object_dispatcher,
+            prepare_copyup(_, _)).WillOnce(Return(r));
+  }
+
+  void expect_prepare_copyup(MockTestImageCtx& mock_image_ctx,
+                             const SparseBufferlist& in_sparse_bl,
+                             const SparseBufferlist& out_sparse_bl) {
+    EXPECT_CALL(*mock_image_ctx.io_object_dispatcher,
+                prepare_copyup(_, _))
+      .WillOnce(WithArg<1>(Invoke(
+        [in_sparse_bl, out_sparse_bl]
+        (SnapshotSparseBufferlist* snap_sparse_bl) {
+          auto& sparse_bl = (*snap_sparse_bl)[0];
+          EXPECT_EQ(in_sparse_bl, sparse_bl);
+
+          sparse_bl = out_sparse_bl;
+          return 0;
+        })));
+  }
+
+  void flush_async_operations(librbd::ImageCtx* ictx) {
+    api::Io<>::flush(*ictx);
+  }
+
+  std::string m_parent_image_name;
+};
+
+TEST_F(TestMockIoCopyupRequest, Standard) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+  expect_prepare_copyup(mock_image_ctx);
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_add_copyup_ops(mock_write_request);
+  expect_sparse_copyup(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0),
+                       {{0, 4096}}, data, 0);
+  expect_write(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, StandardWithSnaps) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->image_lock.lock();
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "2", 2, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "1", 1, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->snapc = {2, {2, 1}};
+  ictx->image_lock.unlock();
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_test_features(mock_image_ctx);
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+  expect_prepare_copyup(mock_image_ctx);
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, 1, 0, OBJECT_EXISTS, true, 0);
+  expect_object_map_update(mock_image_ctx, 2, 0, OBJECT_EXISTS_CLEAN, true, 0);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_add_copyup_ops(mock_write_request);
+  expect_sparse_copyup(mock_image_ctx, 0, ictx->get_object_name(0), {{0, 4096}},
+                       data, 0);
+  expect_write(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, CopyOnRead) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+  expect_prepare_copyup(mock_image_ctx);
+
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_sparse_copyup(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0),
+                       {{0, 4096}}, data, 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->send();
+  flush_async_operations(ictx);
+}
+
+TEST_F(TestMockIoCopyupRequest, CopyOnReadWithSnaps) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->image_lock.lock();
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "1", 1, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->snapc = {1, {1}};
+  ictx->image_lock.unlock();
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_test_features(mock_image_ctx);
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+  expect_prepare_copyup(mock_image_ctx);
+
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, 1, 0, OBJECT_EXISTS, true, 0);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS_CLEAN,
+                           true, 0);
+
+  expect_sparse_copyup(mock_image_ctx, 0, ictx->get_object_name(0), {{0, 4096}},
+                       data, 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->send();
+  flush_async_operations(ictx);
+}
+
+TEST_F(TestMockIoCopyupRequest, DeepCopy) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  MockObjectCopyRequest mock_object_copy_request;
+  mock_image_ctx.migration_info = {1, "", "", "image id", "", {}, ictx->size,
+                                   true};
+  expect_is_empty_write_op(mock_write_request, false);
+  expect_object_copy(mock_image_ctx, mock_object_copy_request, true, 0);
+
+  expect_is_empty_write_op(mock_write_request, false);
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_add_copyup_ops(mock_write_request);
+  expect_sparse_copyup(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0),
+                       {}, "", 0);
+  expect_write(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, DeepCopyOnRead) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  MockObjectCopyRequest mock_object_copy_request;
+  mock_image_ctx.migration_info = {1, "", "", "image id", "", {}, ictx->size,
+                                   false};
+  expect_object_copy(mock_image_ctx, mock_object_copy_request, true, 0);
+
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->send();
+  flush_async_operations(ictx);
+}
+
+TEST_F(TestMockIoCopyupRequest, DeepCopyWithPostSnaps) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->image_lock.lock();
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "3", 3, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "2", 2, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "1", 1, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->snapc = {3, {3, 2, 1}};
+  ictx->image_lock.unlock();
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_test_features(mock_image_ctx);
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  MockObjectCopyRequest mock_object_copy_request;
+  mock_image_ctx.migration_info = {1, "", "", "image id", "",
+                                   {{CEPH_NOSNAP, {2, 1}}},
+                                   ictx->size, true};
+  expect_is_empty_write_op(mock_write_request, false);
+  expect_object_copy(mock_image_ctx, mock_object_copy_request, true, 0);
+
+  expect_is_empty_write_op(mock_write_request, false);
+  expect_get_parent_overlap(mock_image_ctx, 1, 0, 0);
+  expect_get_parent_overlap(mock_image_ctx, 2, 1, 0);
+  expect_prune_parent_extents(mock_image_ctx, 1, 1);
+  expect_get_parent_overlap(mock_image_ctx, 3, 1, 0);
+  expect_prune_parent_extents(mock_image_ctx, 1, 1);
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, 2, 0, OBJECT_EXISTS, true, 0);
+  expect_object_map_update(mock_image_ctx, 3, 0, OBJECT_EXISTS_CLEAN, true, 0);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_add_copyup_ops(mock_write_request);
+  expect_sparse_copyup(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0),
+                       {}, "", 0);
+  expect_write(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, DeepCopyWithPreAndPostSnaps) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->image_lock.lock();
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "4", 4, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "3", 3, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "2", 2, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "1", 1, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->snapc = {4, {4, 3, 2, 1}};
+  ictx->image_lock.unlock();
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_test_features(mock_image_ctx);
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  MockObjectCopyRequest mock_object_copy_request;
+  mock_image_ctx.migration_info = {1, "", "", "image id", "",
+                                   {{CEPH_NOSNAP, {2, 1}}, {10, {1}}},
+                                   ictx->size, true};
+  expect_is_empty_write_op(mock_write_request, false);
+  expect_object_copy(mock_image_ctx, mock_object_copy_request, true, 0);
+
+  expect_is_empty_write_op(mock_write_request, false);
+  expect_get_parent_overlap(mock_image_ctx, 2, 0, 0);
+  expect_get_parent_overlap(mock_image_ctx, 3, 1, 0);
+  expect_prune_parent_extents(mock_image_ctx, 1, 1);
+  expect_get_parent_overlap(mock_image_ctx, 4, 1, 0);
+  expect_prune_parent_extents(mock_image_ctx, 1, 1);
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, 3, 0, OBJECT_EXISTS_CLEAN, true, 0);
+  expect_object_map_update(mock_image_ctx, 4, 0, OBJECT_EXISTS_CLEAN, true, 0);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_add_copyup_ops(mock_write_request);
+  expect_sparse_copyup(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0),
+                       {}, "", 0);
+  expect_write(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, ZeroedCopyup) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  expect_prepare_copyup(mock_image_ctx);
+  expect_is_empty_write_op(mock_write_request, false);
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_add_copyup_ops(mock_write_request);
+  expect_sparse_copyup(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0),
+                       {}, "", 0);
+  expect_write(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, ZeroedCopyOnRead) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '\0');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+  expect_prepare_copyup(mock_image_ctx);
+
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_sparse_copyup(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0),
+                       {}, "", 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->send();
+  flush_async_operations(ictx);
+}
+
+TEST_F(TestMockIoCopyupRequest, NoOpCopyup) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, "", -ENOENT);
+
+  expect_prepare_copyup(mock_image_ctx);
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  expect_is_empty_write_op(mock_write_request, true);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, RestartWrite) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+  expect_prepare_copyup(mock_image_ctx);
+
+  MockAbstractObjectWriteRequest mock_write_request1;
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request1,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  expect_add_copyup_ops(mock_write_request1);
+  expect_sparse_copyup(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0),
+                       {{0, 4096}}, data, 0);
+
+  MockAbstractObjectWriteRequest mock_write_request2;
+  auto& mock_io_ctx = librados::get_mock_io_ctx(
+    mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+  EXPECT_CALL(mock_io_ctx, write(ictx->get_object_name(0), _, 0, 0, _))
+    .WillOnce(WithoutArgs(Invoke([req, &mock_write_request2]() {
+                            req->append_request(&mock_write_request2, {});
+                            return 0;
+                          })));
+
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request1, {});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request1.ctx.wait());
+  ASSERT_EQ(-ERESTART, mock_write_request2.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, ReadFromParentError) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, "", -EPERM);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  MockAbstractObjectWriteRequest mock_write_request;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(-EPERM, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, PrepareCopyupError) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+          mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+  expect_prepare_copyup(mock_image_ctx, -EIO);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  MockAbstractObjectWriteRequest mock_write_request;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(-EIO, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, DeepCopyError) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  MockObjectCopyRequest mock_object_copy_request;
+  mock_image_ctx.migration_info = {1, "", "", "image id", "", {}, ictx->size,
+                                   true};
+  expect_is_empty_write_op(mock_write_request, false);
+  expect_object_copy(mock_image_ctx, mock_object_copy_request, true, -EPERM);
+
+  expect_is_empty_write_op(mock_write_request, false);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(-EPERM, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, UpdateObjectMapError) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING | RBD_FEATURE_OBJECT_MAP);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+  expect_prepare_copyup(mock_image_ctx);
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           -EINVAL);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(-EINVAL, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, CopyupError) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->image_lock.lock();
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "1", 1, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->snapc = {1, {1}};
+  ictx->image_lock.unlock();
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_test_features(mock_image_ctx);
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+  expect_prepare_copyup(mock_image_ctx);
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, 1, 0, OBJECT_EXISTS, true, 0);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_add_copyup_ops(mock_write_request);
+  expect_sparse_copyup(mock_image_ctx, 0, ictx->get_object_name(0), {{0, 4096}},
+                       data, -EPERM);
+  expect_write(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(-EPERM, mock_write_request.ctx.wait());
+  flush_async_operations(ictx);
+}
+
+TEST_F(TestMockIoCopyupRequest, SparseCopyupNotSupported) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+  mock_image_ctx.enable_sparse_copyup = false;
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+  expect_prepare_copyup(mock_image_ctx);
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_add_copyup_ops(mock_write_request);
+  expect_copyup(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), data, 0);
+  expect_write(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, ProcessCopyup) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+
+  bufferlist in_prepare_bl;
+  in_prepare_bl.append(std::string(3072, '1'));
+  bufferlist out_prepare_bl;
+  out_prepare_bl.substr_of(in_prepare_bl, 0, 1024);
+  expect_prepare_copyup(
+    mock_image_ctx,
+    {{1024U, {3072U, {SPARSE_EXTENT_STATE_DATA, 3072,
+                      std::move(in_prepare_bl)}}}},
+    {{2048U, {1024U, {SPARSE_EXTENT_STATE_DATA, 1024,
+                      std::move(out_prepare_bl)}}}});
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_add_copyup_ops(mock_write_request);
+  expect_sparse_copyup(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0),
+                       {{2048, 1024}}, data.substr(0, 1024), 0);
+  expect_write(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {{0, 1024}});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request.ctx.wait());
+}
+
+TEST_F(TestMockIoCopyupRequest, ProcessCopyupOverwrite) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->image_lock.lock();
+  ictx->add_snap(cls::rbd::UserSnapshotNamespace(), "1", 1, ictx->size,
+                 ictx->parent_md, RBD_PROTECTION_STATUS_UNPROTECTED,
+                 0, {});
+  ictx->snapc = {1, {1}};
+  ictx->image_lock.unlock();
+
+  MockTestImageCtx mock_parent_image_ctx(*ictx->parent);
+  MockTestImageCtx mock_image_ctx(*ictx, &mock_parent_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  MockJournal mock_journal;
+  MockObjectMap mock_object_map;
+  initialize_features(ictx, mock_image_ctx, mock_exclusive_lock, mock_journal,
+                      mock_object_map);
+
+  expect_test_features(mock_image_ctx);
+  expect_op_work_queue(mock_image_ctx);
+  expect_is_lock_owner(mock_image_ctx);
+
+  InSequence seq;
+
+  std::string data(4096, '1');
+  expect_read_parent(mock_parent_image_ctx, {{0, 4096}}, data, 0);
+
+  bufferlist in_prepare_bl;
+  in_prepare_bl.append(data);
+  bufferlist out_prepare_bl;
+  out_prepare_bl.substr_of(in_prepare_bl, 0, 1024);
+  expect_prepare_copyup(
+    mock_image_ctx,
+    {{0, {4096, {SPARSE_EXTENT_STATE_DATA, 4096,
+                 std::move(in_prepare_bl)}}}},
+    {{0, {1024, {SPARSE_EXTENT_STATE_DATA, 1024, bufferlist{out_prepare_bl}}}},
+     {2048, {1024, {SPARSE_EXTENT_STATE_DATA, 1024,
+                    bufferlist{out_prepare_bl}}}}});
+
+  MockAbstractObjectWriteRequest mock_write_request;
+  expect_get_pre_write_object_map_state(mock_image_ctx, mock_write_request,
+                                        OBJECT_EXISTS);
+  expect_object_map_at(mock_image_ctx, 0, OBJECT_NONEXISTENT);
+  expect_object_map_update(mock_image_ctx, 1, 0, OBJECT_EXISTS, true, 0);
+  expect_object_map_update(mock_image_ctx, CEPH_NOSNAP, 0, OBJECT_EXISTS, true,
+                           0);
+
+  expect_add_copyup_ops(mock_write_request);
+  expect_sparse_copyup(mock_image_ctx, 0, ictx->get_object_name(0),
+                       {{0, 1024}, {2048, 1024}}, data.substr(0, 2048), 0);
+  expect_write(mock_image_ctx, CEPH_NOSNAP, ictx->get_object_name(0), 0);
+
+  auto req = new MockCopyupRequest(&mock_image_ctx, 0, {{0, 4096}},
+                                   ImageArea::DATA, {});
+  mock_image_ctx.copyup_list[0] = req;
+  req->append_request(&mock_write_request, {{0, 1024}});
+  req->send();
+
+  ASSERT_EQ(0, mock_write_request.ctx.wait());
+}
+
+} // namespace io
+} // namespace librbd
diff --git a/src/test/librbd/io/test_mock_ImageRequest.cc b/src/test/librbd/io/test_mock_ImageRequest.cc
new file mode 100644
index 000000000..9d6423d66
--- /dev/null
+++ b/src/test/librbd/io/test_mock_ImageRequest.cc
@@ -0,0 +1,738 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/librbd/test_mock_fixture.h"
+#include "test/librbd/test_support.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "test/librbd/mock/MockJournal.h"
+#include "test/librbd/mock/cache/MockImageCache.h"
+#include "librbd/io/ImageRequest.h"
+#include "librbd/io/ObjectDispatchSpec.h"
+#include "librbd/io/Utils.h"
+
+namespace librbd {
+namespace {
+
+struct MockTestImageCtx;
+
+struct MockTestJournal : public MockJournal {
+  MOCK_METHOD4(append_write_event, uint64_t(uint64_t, size_t,
+                                            const bufferlist &, bool));
+  MOCK_METHOD5(append_compare_and_write_event, uint64_t(uint64_t, size_t,
+                                                        const bufferlist &,
+                                                        const bufferlist &,
+                                                        bool));
+  MOCK_METHOD5(append_io_event_mock, uint64_t(const journal::EventEntry&,
+                                              uint64_t, size_t, bool, int));
+  uint64_t append_io_event(journal::EventEntry &&event_entry,
+                           uint64_t offset, size_t length,
+                           bool flush_entry, int filter_ret_val) {
+    // googlemock doesn't support move semantics
+    return append_io_event_mock(event_entry, offset, length, flush_entry,
+                                filter_ret_val);
+  }
+
+  MOCK_METHOD2(commit_io_event, void(uint64_t, int));
+};
+
+struct MockTestImageCtx : public MockImageCtx {
+  MockTestImageCtx(ImageCtx &image_ctx) : MockImageCtx(image_ctx) {
+  }
+
+  MockTestJournal* journal;
+};
+
+} // anonymous namespace
+
+namespace util {
+
+inline ImageCtx *get_image_ctx(MockTestImageCtx *image_ctx) {
+  return image_ctx->image_ctx;
+}
+
+} // namespace util
+} // namespace librbd
+
+#include "librbd/io/ImageRequest.cc"
+
+namespace librbd {
+namespace io {
+
+namespace util {
+
+template <>
+void area_to_object_extents(MockTestImageCtx* image_ctx, uint64_t offset,
+                            uint64_t length, ImageArea area,
+                            uint64_t buffer_offset,
+                            striper::LightweightObjectExtents* object_extents) {
+  Striper::file_to_extents(image_ctx->cct, &image_ctx->layout, offset, length,
+                           0, buffer_offset, object_extents);
+}
+
+template <>
+std::pair<Extents, ImageArea> object_to_area_extents(
+    MockTestImageCtx* image_ctx, uint64_t object_no,
+    const Extents& object_extents) {
+  Extents extents;
+  for (auto [off, len] : object_extents) {
+    Striper::extent_to_file(image_ctx->cct, &image_ctx->layout, object_no, off,
+                            len, extents);
+  }
+  return {std::move(extents), ImageArea::DATA};
+}
+
+} // namespace util
+
+using ::testing::_;
+using ::testing::InSequence;
+using ::testing::Invoke;
+using ::testing::Return;
+using ::testing::WithArg;
+using ::testing::WithoutArgs;
+using ::testing::Exactly;
+
+struct TestMockIoImageRequest : public TestMockFixture {
+  typedef ImageRequest<librbd::MockTestImageCtx> MockImageRequest;
+  typedef ImageReadRequest<librbd::MockTestImageCtx> MockImageReadRequest;
+  typedef ImageWriteRequest<librbd::MockTestImageCtx> MockImageWriteRequest;
+  typedef ImageDiscardRequest<librbd::MockTestImageCtx> MockImageDiscardRequest;
+  typedef ImageFlushRequest<librbd::MockTestImageCtx> MockImageFlushRequest;
+  typedef ImageWriteSameRequest<librbd::MockTestImageCtx> MockImageWriteSameRequest;
+  typedef ImageCompareAndWriteRequest<librbd::MockTestImageCtx> MockImageCompareAndWriteRequest;
+  typedef ImageListSnapsRequest<librbd::MockTestImageCtx> MockImageListSnapsRequest;
+
+  void expect_is_journal_appending(MockTestJournal &mock_journal, bool appending) {
+    EXPECT_CALL(mock_journal, is_journal_appending())
+      .WillOnce(Return(appending));
+  }
+
+  void expect_get_modify_timestamp(MockTestImageCtx &mock_image_ctx,
+                                   bool needs_update) {
+    if (needs_update) {
+      mock_image_ctx.mtime_update_interval = 5;
+      EXPECT_CALL(mock_image_ctx, get_modify_timestamp())
+        .WillOnce(Return(ceph_clock_now() - utime_t(10,0)));
+    } else {
+      mock_image_ctx.mtime_update_interval = 600;
+      EXPECT_CALL(mock_image_ctx, get_modify_timestamp())
+        .WillOnce(Return(ceph_clock_now()));
+    }
+  }
+
+  void expect_journal_append_io_event(MockTestJournal &mock_journal, uint64_t journal_tid,
+                                      uint64_t offset, size_t length) {
+    EXPECT_CALL(mock_journal, append_io_event_mock(_, offset, length, _, _))
+      .WillOnce(Return(journal_tid));
+  }
+
+  void expect_object_discard_request(MockTestImageCtx &mock_image_ctx,
+                                     uint64_t object_no, uint64_t offset,
+                                     uint32_t length, int r) {
+    EXPECT_CALL(*mock_image_ctx.io_object_dispatcher, send(_))
+      .WillOnce(Invoke([&mock_image_ctx, object_no, offset, length, r]
+                (ObjectDispatchSpec* spec) {
+                  auto* discard_spec = boost::get<ObjectDispatchSpec::DiscardRequest>(&spec->request);
+                  ASSERT_TRUE(discard_spec != nullptr);
+                  ASSERT_EQ(object_no, discard_spec->object_no);
+                  ASSERT_EQ(offset, discard_spec->object_off);
+                  ASSERT_EQ(length, discard_spec->object_len);
+
+                  spec->dispatch_result = io::DISPATCH_RESULT_COMPLETE;
+                  mock_image_ctx.image_ctx->op_work_queue->queue(&spec->dispatcher_ctx, r);
+                }));
+  }
+
+  void expect_object_request_send(MockTestImageCtx &mock_image_ctx,
+                                  int r) {
+    EXPECT_CALL(*mock_image_ctx.io_object_dispatcher, send(_))
+      .WillOnce(Invoke([&mock_image_ctx, r](ObjectDispatchSpec* spec) {
+                  spec->dispatch_result = io::DISPATCH_RESULT_COMPLETE;
+                  mock_image_ctx.image_ctx->op_work_queue->queue(&spec->dispatcher_ctx, r);
+                }));
+  }
+
+  void expect_object_list_snaps_request(MockTestImageCtx &mock_image_ctx,
+                                        uint64_t object_no,
+                                        const SnapshotDelta& snap_delta,
+                                        int r) {
+    EXPECT_CALL(*mock_image_ctx.io_object_dispatcher, send(_))
+      .WillOnce(
+        Invoke([&mock_image_ctx, object_no, snap_delta, r]
+               (ObjectDispatchSpec* spec) {
+                  auto request = boost::get<
+                    librbd::io::ObjectDispatchSpec::ListSnapsRequest>(
+                      &spec->request);
+                  ASSERT_TRUE(request != nullptr);
+                  ASSERT_EQ(object_no, request->object_no);
+
+                  *request->snapshot_delta = snap_delta;
+                  spec->dispatch_result = io::DISPATCH_RESULT_COMPLETE;
+                  mock_image_ctx.image_ctx->op_work_queue->queue(&spec->dispatcher_ctx, r);
+                }));
+  }
+};
+
+TEST_F(TestMockIoImageRequest, AioWriteModifyTimestamp) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  mock_image_ctx.mtime_update_interval = 5;
+
+  utime_t dummy = ceph_clock_now();
+  dummy -= utime_t(10,0);
+
+  EXPECT_CALL(mock_image_ctx, get_modify_timestamp())
+      .Times(Exactly(3))
+      .WillOnce(Return(dummy))
+      .WillOnce(Return(dummy))
+      .WillOnce(Return(dummy + utime_t(10,0)));
+
+  EXPECT_CALL(mock_image_ctx, set_modify_timestamp(_))
+      .Times(Exactly(1));
+
+  InSequence seq;
+  expect_is_journal_appending(mock_journal, false);
+  expect_object_request_send(mock_image_ctx, 0);
+
+  C_SaferCond aio_comp_ctx_1, aio_comp_ctx_2;
+  AioCompletion *aio_comp_1 = AioCompletion::create_and_start(
+    &aio_comp_ctx_1, ictx, AIO_TYPE_WRITE);
+
+  AioCompletion *aio_comp_2 = AioCompletion::create_and_start(
+    &aio_comp_ctx_2, ictx, AIO_TYPE_WRITE);
+
+  bufferlist bl;
+  bl.append("1");
+  MockImageWriteRequest mock_aio_image_write_1(
+    mock_image_ctx, aio_comp_1, {{0, 1}}, ImageArea::DATA, std::move(bl),
+    0, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_write_1.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx_1.wait());
+
+  expect_is_journal_appending(mock_journal, false);
+  expect_object_request_send(mock_image_ctx, 0);
+
+  bl.append("1");
+  MockImageWriteRequest mock_aio_image_write_2(
+    mock_image_ctx, aio_comp_2, {{0, 1}}, ImageArea::DATA, std::move(bl),
+    0, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_write_2.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx_2.wait());
+}
+
+TEST_F(TestMockIoImageRequest, AioReadAccessTimestamp) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  mock_image_ctx.atime_update_interval = 5;
+
+  utime_t dummy = ceph_clock_now();
+  dummy -= utime_t(10,0);
+
+  EXPECT_CALL(mock_image_ctx, get_access_timestamp())
+      .Times(Exactly(3))
+      .WillOnce(Return(dummy))
+      .WillOnce(Return(dummy))
+      .WillOnce(Return(dummy + utime_t(10,0)));
+
+  EXPECT_CALL(mock_image_ctx, set_access_timestamp(_))
+      .Times(Exactly(1));
+
+  InSequence seq;
+  expect_object_request_send(mock_image_ctx, 0);
+
+  C_SaferCond aio_comp_ctx_1, aio_comp_ctx_2;
+  AioCompletion *aio_comp_1 = AioCompletion::create_and_start(
+    &aio_comp_ctx_1, ictx, AIO_TYPE_READ);
+
+
+  ReadResult rr;
+  MockImageReadRequest mock_aio_image_read_1(
+    mock_image_ctx, aio_comp_1, {{0, 1}}, ImageArea::DATA, std::move(rr),
+    mock_image_ctx.get_data_io_context(), 0, 0, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_read_1.send();
+  }
+  ASSERT_EQ(1, aio_comp_ctx_1.wait());
+
+  AioCompletion *aio_comp_2 = AioCompletion::create_and_start(
+    &aio_comp_ctx_2, ictx, AIO_TYPE_READ);
+  expect_object_request_send(mock_image_ctx, 0);
+
+  MockImageReadRequest mock_aio_image_read_2(
+    mock_image_ctx, aio_comp_2, {{0, 1}}, ImageArea::DATA, std::move(rr),
+    mock_image_ctx.get_data_io_context(), 0, 0, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_read_2.send();
+  }
+  ASSERT_EQ(1, aio_comp_ctx_2.wait());
+}
+
+TEST_F(TestMockIoImageRequest, PartialDiscard) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->discard_granularity_bytes = 0;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.journal = nullptr;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_object_discard_request(mock_image_ctx, 0, 16, 63, 0);
+  expect_object_discard_request(mock_image_ctx, 0, 84, 100, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_DISCARD);
+  MockImageDiscardRequest mock_aio_image_discard(
+    mock_image_ctx, aio_comp, {{16, 63}, {84, 100}}, ImageArea::DATA,
+    ictx->discard_granularity_bytes, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_discard.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, TailDiscard) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ASSERT_EQ(0, resize(ictx, ictx->layout.object_size));
+  ictx->discard_granularity_bytes = 2 * ictx->layout.object_size;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.journal = nullptr;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_object_discard_request(
+    mock_image_ctx, 0, ictx->layout.object_size - 1024, 1024, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_DISCARD);
+  MockImageDiscardRequest mock_aio_image_discard(
+    mock_image_ctx, aio_comp,
+    {{ictx->layout.object_size - 1024, 1024}}, ImageArea::DATA,
+    ictx->discard_granularity_bytes, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_discard.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, DiscardGranularity) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ASSERT_EQ(0, resize(ictx, ictx->layout.object_size));
+  ictx->discard_granularity_bytes = 32;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.journal = nullptr;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_object_discard_request(mock_image_ctx, 0, 32, 32, 0);
+  expect_object_discard_request(mock_image_ctx, 0, 96, 64, 0);
+  expect_object_discard_request(
+    mock_image_ctx, 0, ictx->layout.object_size - 32, 32, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_DISCARD);
+  MockImageDiscardRequest mock_aio_image_discard(
+    mock_image_ctx, aio_comp,
+    {{16, 63}, {96, 31}, {84, 100}, {ictx->layout.object_size - 33, 33}},
+    ImageArea::DATA, ictx->discard_granularity_bytes, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_discard.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, PartialDiscardJournalAppendEnabled) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->discard_granularity_bytes = 0;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_is_journal_appending(mock_journal, true);
+  expect_journal_append_io_event(mock_journal, 0, 16, 63);
+  expect_journal_append_io_event(mock_journal, 1, 84, 100);
+  expect_object_discard_request(mock_image_ctx, 0, 16, 63, 0);
+  expect_object_discard_request(mock_image_ctx, 0, 84, 100, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_DISCARD);
+  MockImageDiscardRequest mock_aio_image_discard(
+    mock_image_ctx, aio_comp, {{16, 63}, {84, 100}}, ImageArea::DATA,
+    ictx->discard_granularity_bytes, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_discard.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, TailDiscardJournalAppendEnabled) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ASSERT_EQ(0, resize(ictx, ictx->layout.object_size));
+  ictx->discard_granularity_bytes = 2 * ictx->layout.object_size;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_is_journal_appending(mock_journal, true);
+  expect_journal_append_io_event(
+    mock_journal, 0, ictx->layout.object_size - 1024, 1024);
+  expect_object_discard_request(
+    mock_image_ctx, 0, ictx->layout.object_size - 1024, 1024, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_DISCARD);
+  MockImageDiscardRequest mock_aio_image_discard(
+    mock_image_ctx, aio_comp,
+    {{ictx->layout.object_size - 1024, 1024}}, ImageArea::DATA,
+    ictx->discard_granularity_bytes, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_discard.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, PruneRequiredDiscardJournalAppendEnabled) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->discard_granularity_bytes = 32;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_is_journal_appending(mock_journal, true);
+  EXPECT_CALL(mock_journal, append_io_event_mock(_, _, _, _, _)).Times(0);
+  EXPECT_CALL(*mock_image_ctx.io_object_dispatcher, send(_)).Times(0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_DISCARD);
+  MockImageDiscardRequest mock_aio_image_discard(
+    mock_image_ctx, aio_comp, {{96, 31}}, ImageArea::DATA,
+    ictx->discard_granularity_bytes, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_discard.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, LengthModifiedDiscardJournalAppendEnabled) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->discard_granularity_bytes = 32;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_is_journal_appending(mock_journal, true);
+  expect_journal_append_io_event(mock_journal, 0, 32, 32);
+  expect_object_discard_request(mock_image_ctx, 0, 32, 32, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_DISCARD);
+  MockImageDiscardRequest mock_aio_image_discard(
+    mock_image_ctx, aio_comp, {{16, 63}}, ImageArea::DATA,
+    ictx->discard_granularity_bytes, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_discard.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, DiscardGranularityJournalAppendEnabled) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ASSERT_EQ(0, resize(ictx, ictx->layout.object_size));
+  ictx->discard_granularity_bytes = 32;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_is_journal_appending(mock_journal, true);
+  expect_journal_append_io_event(mock_journal, 0, 32, 32);
+  expect_journal_append_io_event(mock_journal, 1, 96, 64);
+  expect_journal_append_io_event(
+    mock_journal, 2, ictx->layout.object_size - 32, 32);
+  expect_object_discard_request(mock_image_ctx, 0, 32, 32, 0);
+  expect_object_discard_request(mock_image_ctx, 0, 96, 64, 0);
+  expect_object_discard_request(
+    mock_image_ctx, 0, ictx->layout.object_size - 32, 32, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_DISCARD);
+  MockImageDiscardRequest mock_aio_image_discard(
+    mock_image_ctx, aio_comp,
+    {{16, 63}, {96, 31}, {84, 100}, {ictx->layout.object_size - 33, 33}},
+    ImageArea::DATA, ictx->discard_granularity_bytes, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_discard.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, AioWriteJournalAppendDisabled) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_is_journal_appending(mock_journal, false);
+  expect_object_request_send(mock_image_ctx, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_WRITE);
+
+  bufferlist bl;
+  bl.append("1");
+  MockImageWriteRequest mock_aio_image_write(
+    mock_image_ctx, aio_comp, {{0, 1}}, ImageArea::DATA, std::move(bl), 0, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_write.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, AioDiscardJournalAppendDisabled) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->discard_granularity_bytes = 0;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_is_journal_appending(mock_journal, false);
+  expect_object_request_send(mock_image_ctx, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_DISCARD);
+  MockImageDiscardRequest mock_aio_image_discard(
+    mock_image_ctx, aio_comp, {{0, 1}}, ImageArea::DATA,
+    ictx->discard_granularity_bytes, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_discard.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, AioFlushJournalAppendDisabled) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  expect_op_work_queue(mock_image_ctx);
+
+  InSequence seq;
+  expect_is_journal_appending(mock_journal, false);
+  expect_object_request_send(mock_image_ctx, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_FLUSH);
+  MockImageFlushRequest mock_aio_image_flush(mock_image_ctx, aio_comp,
+                                             FLUSH_SOURCE_USER, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_flush.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, AioWriteSameJournalAppendDisabled) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_is_journal_appending(mock_journal, false);
+  expect_object_request_send(mock_image_ctx, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_WRITESAME);
+
+  bufferlist bl;
+  bl.append("1");
+  MockImageWriteSameRequest mock_aio_image_writesame(
+    mock_image_ctx, aio_comp, {{0, 1}}, ImageArea::DATA, std::move(bl), 0, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_writesame.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, AioCompareAndWriteJournalAppendDisabled) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockTestJournal mock_journal;
+  mock_image_ctx.journal = &mock_journal;
+
+  InSequence seq;
+  expect_get_modify_timestamp(mock_image_ctx, false);
+  expect_is_journal_appending(mock_journal, false);
+  expect_object_request_send(mock_image_ctx, 0);
+
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_COMPARE_AND_WRITE);
+
+  bufferlist cmp_bl;
+  cmp_bl.append("1");
+  bufferlist write_bl;
+  write_bl.append("1");
+  uint64_t mismatch_offset;
+  MockImageCompareAndWriteRequest mock_aio_image_write(
+    mock_image_ctx, aio_comp, {{0, 1}}, ImageArea::DATA,
+    std::move(cmp_bl), std::move(write_bl), &mismatch_offset, 0, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_aio_image_write.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+}
+
+TEST_F(TestMockIoImageRequest, ListSnaps) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.layout.object_size = 16384;
+  mock_image_ctx.layout.stripe_unit = 4096;
+  mock_image_ctx.layout.stripe_count = 2;
+
+  InSequence seq;
+
+  SnapshotDelta object_snapshot_delta;
+  object_snapshot_delta[{5,6}].insert(
+    0, 1024, {SPARSE_EXTENT_STATE_DATA, 1024});
+  object_snapshot_delta[{5,5}].insert(
+    4096, 4096, {SPARSE_EXTENT_STATE_ZEROED, 4096});
+  expect_object_list_snaps_request(mock_image_ctx, 0, object_snapshot_delta, 0);
+  object_snapshot_delta = {};
+  object_snapshot_delta[{5,6}].insert(
+    1024, 3072, {SPARSE_EXTENT_STATE_DATA, 3072});
+  object_snapshot_delta[{5,5}].insert(
+    2048, 2048, {SPARSE_EXTENT_STATE_ZEROED, 2048});
+  expect_object_list_snaps_request(mock_image_ctx, 1, object_snapshot_delta, 0);
+
+  SnapshotDelta snapshot_delta;
+  C_SaferCond aio_comp_ctx;
+  AioCompletion *aio_comp = AioCompletion::create_and_start(
+    &aio_comp_ctx, ictx, AIO_TYPE_GENERIC);
+  MockImageListSnapsRequest mock_image_list_snaps_request(
+    mock_image_ctx, aio_comp, {{0, 16384}, {16384, 16384}}, ImageArea::DATA,
+    {0, CEPH_NOSNAP}, 0, &snapshot_delta, {});
+  {
+    std::shared_lock owner_locker{mock_image_ctx.owner_lock};
+    mock_image_list_snaps_request.send();
+  }
+  ASSERT_EQ(0, aio_comp_ctx.wait());
+
+  SnapshotDelta expected_snapshot_delta;
+  expected_snapshot_delta[{5,6}].insert(
+    0, 1024, {SPARSE_EXTENT_STATE_DATA, 1024});
+  expected_snapshot_delta[{5,6}].insert(
+    5120, 3072, {SPARSE_EXTENT_STATE_DATA, 3072});
+  expected_snapshot_delta[{5,5}].insert(
+    6144, 6144, {SPARSE_EXTENT_STATE_ZEROED, 6144});
+  ASSERT_EQ(expected_snapshot_delta, snapshot_delta);
+}
+
+} // namespace io
+} // namespace librbd
diff --git a/src/test/librbd/io/test_mock_ObjectRequest.cc b/src/test/librbd/io/test_mock_ObjectRequest.cc
new file mode 100644
index 000000000..0690b7722
--- /dev/null
+++ b/src/test/librbd/io/test_mock_ObjectRequest.cc
@@ -0,0 +1,1968 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/librbd/test_mock_fixture.h"
+#include "test/librbd/test_support.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "test/librbd/mock/MockObjectMap.h"
+#include "test/librbd/mock/cache/MockImageCache.h"
+#include "test/librados_test_stub/MockTestMemIoCtxImpl.h"
+#include "test/librados_test_stub/MockTestMemRadosClient.h"
+#include "include/rbd/librbd.hpp"
+#include "librbd/io/CopyupRequest.h"
+#include "librbd/io/ImageRequest.h"
+#include "librbd/io/ObjectRequest.h"
+#include "librbd/io/Utils.h"
+
+namespace librbd {
+namespace {
+
+struct MockTestImageCtx : public MockImageCtx {
+  MockTestImageCtx(ImageCtx &image_ctx) : MockImageCtx(image_ctx) {
+  }
+};
+
+} // anonymous namespace
+
+namespace util {
+
+inline ImageCtx *get_image_ctx(MockImageCtx *image_ctx) {
+  return image_ctx->image_ctx;
+}
+
+} // namespace util
+
+namespace io {
+
+template <>
+struct CopyupRequest<librbd::MockImageCtx> {
+  MOCK_METHOD0(send, void());
+  MOCK_METHOD2(append_request, void(AbstractObjectWriteRequest<librbd::MockTestImageCtx>*,
+                                    const Extents&));
+};
+
+template <>
+struct CopyupRequest<librbd::MockTestImageCtx> : public CopyupRequest<librbd::MockImageCtx> {
+  static CopyupRequest* s_instance;
+  static CopyupRequest* create(librbd::MockTestImageCtx *ictx,
+                               uint64_t objectno, Extents &&image_extents,
+                               ImageArea area,
+                               const ZTracer::Trace& parent_trace) {
+    return s_instance;
+  }
+
+  CopyupRequest() {
+    s_instance = this;
+  }
+};
+
+CopyupRequest<librbd::MockTestImageCtx>* CopyupRequest<librbd::MockTestImageCtx>::s_instance = nullptr;
+
+template <>
+struct ImageListSnapsRequest<librbd::MockTestImageCtx> {
+  static ImageListSnapsRequest* s_instance;
+
+  AioCompletion* aio_comp;
+  Extents image_extents;
+  SnapshotDelta* snapshot_delta;
+
+  ImageListSnapsRequest() {
+    s_instance = this;
+  }
+  ImageListSnapsRequest(
+      librbd::MockImageCtx& image_ctx, AioCompletion* aio_comp,
+      Extents&& image_extents, ImageArea area, SnapIds&& snap_ids,
+      int list_snaps_flags, SnapshotDelta* snapshot_delta,
+      const ZTracer::Trace& parent_trace) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->aio_comp = aio_comp;
+    s_instance->image_extents = image_extents;
+    s_instance->snapshot_delta = snapshot_delta;
+  }
+
+
+  MOCK_METHOD0(execute_send, void());
+  void send() {
+    ceph_assert(s_instance != nullptr);
+    s_instance->execute_send();
+  }
+};
+
+ImageListSnapsRequest<librbd::MockTestImageCtx>* ImageListSnapsRequest<librbd::MockTestImageCtx>::s_instance = nullptr;
+
+namespace util {
+
+template <>
+void area_to_object_extents(MockTestImageCtx* image_ctx, uint64_t offset,
+                            uint64_t length, ImageArea area,
+                            uint64_t buffer_offset,
+                            striper::LightweightObjectExtents* object_extents) {
+  Striper::file_to_extents(image_ctx->cct, &image_ctx->layout, offset, length,
+                           0, buffer_offset, object_extents);
+}
+
+template <>
+std::pair<Extents, ImageArea> object_to_area_extents(
+    MockTestImageCtx* image_ctx, uint64_t object_no,
+    const Extents& object_extents) {
+  Extents extents;
+  for (auto [off, len] : object_extents) {
+    Striper::extent_to_file(image_ctx->cct, &image_ctx->layout, object_no, off,
+                            len, extents);
+  }
+  return {std::move(extents), ImageArea::DATA};
+}
+
+namespace {
+
+struct Mock {
+  static Mock* s_instance;
+
+  Mock() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD6(read_parent,
+               void(librbd::MockTestImageCtx *, uint64_t, ReadExtents*,
+                    librados::snap_t, const ZTracer::Trace &, Context*));
+};
+
+Mock *Mock::s_instance = nullptr;
+
+} // anonymous namespace
+
+template<> void read_parent(
+    librbd::MockTestImageCtx *image_ctx, uint64_t object_no,
+    ReadExtents* extents, librados::snap_t snap_id,
+    const ZTracer::Trace &trace, Context* on_finish) {
+  Mock::s_instance->read_parent(image_ctx, object_no, extents, snap_id, trace,
+                                on_finish);
+}
+
+} // namespace util
+
+} // namespace io
+} // namespace librbd
+
+#include "librbd/io/ObjectRequest.cc"
+
+namespace librbd {
+namespace io {
+
+using ::testing::_;
+using ::testing::DoDefault;
+using ::testing::InSequence;
+using ::testing::Invoke;
+using ::testing::Return;
+using ::testing::WithArg;
+using ::testing::WithArgs;
+
+struct TestMockIoObjectRequest : public TestMockFixture {
+  typedef ObjectRequest<librbd::MockTestImageCtx> MockObjectRequest;
+  typedef ObjectReadRequest<librbd::MockTestImageCtx> MockObjectReadRequest;
+  typedef ObjectWriteRequest<librbd::MockTestImageCtx> MockObjectWriteRequest;
+  typedef ObjectDiscardRequest<librbd::MockTestImageCtx> MockObjectDiscardRequest;
+  typedef ObjectWriteSameRequest<librbd::MockTestImageCtx> MockObjectWriteSameRequest;
+  typedef ObjectCompareAndWriteRequest<librbd::MockTestImageCtx> MockObjectCompareAndWriteRequest;
+  typedef ObjectListSnapsRequest<librbd::MockTestImageCtx> MockObjectListSnapsRequest;
+  typedef AbstractObjectWriteRequest<librbd::MockTestImageCtx> MockAbstractObjectWriteRequest;
+  typedef CopyupRequest<librbd::MockTestImageCtx> MockCopyupRequest;
+  typedef ImageListSnapsRequest<librbd::MockTestImageCtx> MockImageListSnapsRequest;
+  typedef util::Mock MockUtils;
+
+  void expect_object_may_exist(MockTestImageCtx &mock_image_ctx,
+                               uint64_t object_no, bool exists) {
+    if (mock_image_ctx.object_map != nullptr) {
+      EXPECT_CALL(*mock_image_ctx.object_map, object_may_exist(object_no))
+        .WillOnce(Return(exists));
+    }
+  }
+
+  void expect_get_object_size(MockTestImageCtx &mock_image_ctx) {
+    EXPECT_CALL(mock_image_ctx, get_object_size()).WillRepeatedly(Return(
+      mock_image_ctx.layout.object_size));
+  }
+
+  void expect_get_parent_overlap(MockTestImageCtx &mock_image_ctx,
+                                 librados::snap_t snap_id, uint64_t overlap,
+                                 int r) {
+    EXPECT_CALL(mock_image_ctx, get_parent_overlap(snap_id, _))
+      .WillOnce(WithArg<1>(Invoke([overlap, r](uint64_t *o) {
+                             *o = overlap;
+                             return r;
+                           })));
+  }
+
+  void expect_prune_parent_extents(MockTestImageCtx &mock_image_ctx,
+                                   const Extents& extents,
+                                   uint64_t overlap, uint64_t object_overlap) {
+    EXPECT_CALL(mock_image_ctx, prune_parent_extents(_, _, overlap, _))
+      .WillOnce(WithArg<0>(Invoke([extents, object_overlap](Extents& e) {
+                             e = extents;
+                             return object_overlap;
+                           })));
+  }
+
+  void expect_get_read_flags(MockTestImageCtx &mock_image_ctx,
+                             librados::snap_t snap_id, int flags) {
+    EXPECT_CALL(mock_image_ctx, get_read_flags(snap_id))
+      .WillOnce(Return(flags));
+  }
+
+  void expect_is_lock_owner(MockExclusiveLock& mock_exclusive_lock) {
+    EXPECT_CALL(mock_exclusive_lock, is_lock_owner()).WillRepeatedly(
+      Return(true));
+  }
+
+  void expect_read(MockTestImageCtx &mock_image_ctx,
+                   const std::string& oid, uint64_t off, uint64_t len,
+                   const std::string& data, int r) {
+    bufferlist bl;
+    bl.append(data);
+
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    auto& expect = EXPECT_CALL(mock_io_ctx, read(oid, len, off, _, _, _));
+    if (r < 0) {
+      expect.WillOnce(Return(r));
+    } else {
+      expect.WillOnce(WithArg<3>(Invoke([bl](bufferlist *out_bl) {
+                                   out_bl->append(bl);
+                                   return bl.length();
+                                 })));
+    }
+  }
+
+  void expect_sparse_read(MockTestImageCtx &mock_image_ctx,
+                          const std::string& oid, uint64_t off, uint64_t len,
+                          const std::string& data, int r) {
+    bufferlist bl;
+    bl.append(data);
+
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    auto& expect = EXPECT_CALL(mock_io_ctx,
+                               sparse_read(oid, off, len, _, _, _));
+    if (r < 0) {
+      expect.WillOnce(Return(r));
+    } else {
+      expect.WillOnce(WithArg<4>(Invoke([bl](bufferlist *out_bl) {
+                                   out_bl->append(bl);
+                                   return bl.length();
+                                 })));
+    }
+  }
+
+  void expect_read_parent(MockUtils &mock_utils, uint64_t object_no,
+                          ReadExtents* extents, librados::snap_t snap_id,
+                          int r) {
+    EXPECT_CALL(mock_utils,
+                read_parent(_, object_no, extents, snap_id, _, _))
+      .WillOnce(WithArg<5>(CompleteContext(r, static_cast<asio::ContextWQ*>(nullptr))));
+  }
+
+  void expect_copyup(MockCopyupRequest& mock_copyup_request, int r) {
+    EXPECT_CALL(mock_copyup_request, send())
+      .WillOnce(Invoke([]() {}));
+  }
+
+  void expect_copyup(MockCopyupRequest& mock_copyup_request,
+                     MockAbstractObjectWriteRequest** write_request, int r) {
+    EXPECT_CALL(mock_copyup_request, append_request(_, _))
+      .WillOnce(WithArg<0>(
+        Invoke([write_request](MockAbstractObjectWriteRequest *req) {
+            *write_request = req;
+          })));
+    EXPECT_CALL(mock_copyup_request, send())
+      .WillOnce(Invoke([write_request, r]() {
+                  (*write_request)->handle_copyup(r);
+                }));
+  }
+
+  void expect_object_map_update(MockTestImageCtx &mock_image_ctx,
+                                uint64_t start_object, uint64_t end_object,
+                                uint8_t state,
+                                const boost::optional<uint8_t> &current_state,
+                                bool updated, int ret_val) {
+    if (mock_image_ctx.object_map != nullptr) {
+      EXPECT_CALL(*mock_image_ctx.object_map,
+                  aio_update(CEPH_NOSNAP, start_object, end_object, state,
+                             current_state, _, false, _))
+        .WillOnce(WithArg<7>(Invoke([&mock_image_ctx, updated, ret_val](Context *ctx) {
+                               if (updated) {
+                                 mock_image_ctx.op_work_queue->queue(ctx, ret_val);
+                               }
+                               return updated;
+                             })));
+    }
+  }
+
+  void expect_assert_exists(MockTestImageCtx &mock_image_ctx, int r) {
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    EXPECT_CALL(mock_io_ctx, assert_exists(_, _))
+      .WillOnce(Return(r));
+  }
+
+  void expect_write(MockTestImageCtx &mock_image_ctx,
+                    uint64_t offset, uint64_t length, int r) {
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    auto &expect = EXPECT_CALL(mock_io_ctx,
+                               write(_, _, length, offset, _));
+    if (r < 0) {
+      expect.WillOnce(Return(r));
+    } else {
+      expect.WillOnce(DoDefault());
+    }
+  }
+
+  void expect_write_full(MockTestImageCtx &mock_image_ctx, int r) {
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    auto &expect = EXPECT_CALL(mock_io_ctx, write_full(_, _, _));
+    if (r < 0) {
+      expect.WillOnce(Return(r));
+    } else {
+      expect.WillOnce(DoDefault());
+    }
+  }
+
+  void expect_writesame(MockTestImageCtx &mock_image_ctx,
+                        uint64_t offset, uint64_t length, int r) {
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    auto &expect = EXPECT_CALL(mock_io_ctx, writesame(_, _, length, offset, _));
+    if (r < 0) {
+      expect.WillOnce(Return(r));
+    } else {
+      expect.WillOnce(DoDefault());
+    }
+  }
+
+  void expect_remove(MockTestImageCtx &mock_image_ctx, int r) {
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    auto &expect = EXPECT_CALL(mock_io_ctx, remove(_, _));
+    if (r < 0) {
+      expect.WillOnce(Return(r));
+    } else {
+      expect.WillOnce(DoDefault());
+    }
+  }
+
+  void expect_create(MockTestImageCtx &mock_image_ctx, bool exclusive) {
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    EXPECT_CALL(mock_io_ctx, create(_, exclusive, _))
+      .Times(1);
+  }
+
+  void expect_truncate(MockTestImageCtx &mock_image_ctx, int offset, int r) {
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    auto &expect = EXPECT_CALL(mock_io_ctx, truncate(_, offset, _));
+    if (r < 0) {
+      expect.WillOnce(Return(r));
+    } else {
+      expect.WillOnce(DoDefault());
+    }
+  }
+
+  void expect_zero(MockTestImageCtx &mock_image_ctx, int offset, int length,
+                   int r) {
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    auto &expect = EXPECT_CALL(mock_io_ctx, zero(_, offset, length, _));
+    if (r < 0) {
+      expect.WillOnce(Return(r));
+    } else {
+      expect.WillOnce(DoDefault());
+    }
+  }
+
+  void expect_cmpext(MockTestImageCtx &mock_image_ctx, int offset, int r) {
+    auto& mock_io_ctx = librados::get_mock_io_ctx(
+      mock_image_ctx.rados_api, *mock_image_ctx.get_data_io_context());
+    auto &expect = EXPECT_CALL(mock_io_ctx, cmpext(_, offset, _, _));
+    if (r < 0) {
+      expect.WillOnce(Return(r));
+    } else {
+      expect.WillOnce(DoDefault());
+    }
+  }
+
+  void expect_list_snaps(MockTestImageCtx &mock_image_ctx,
+                        const librados::snap_set_t& snap_set, int r) {
+    auto io_context = *mock_image_ctx.get_data_io_context();
+    io_context.read_snap(CEPH_SNAPDIR);
+    auto& mock_io_ctx = librados::get_mock_io_ctx(mock_image_ctx.rados_api,
+                                                  io_context);
+    EXPECT_CALL(mock_io_ctx, list_snaps(_, _))
+      .WillOnce(WithArg<1>(Invoke(
+        [snap_set, r](librados::snap_set_t* out_snap_set) {
+          *out_snap_set = snap_set;
+          return r;
+        })));
+  }
+
+  void expect_image_list_snaps(MockImageListSnapsRequest& req,
+                               const Extents& image_extents,
+                               const SnapshotDelta& image_snapshot_delta,
+                               int r) {
+    EXPECT_CALL(req, execute_send())
+      .WillOnce(Invoke(
+        [&req, image_extents, image_snapshot_delta, r]() {
+          ASSERT_EQ(image_extents, req.image_extents);
+          *req.snapshot_delta = image_snapshot_delta;
+
+          auto aio_comp = req.aio_comp;
+          aio_comp->set_request_count(1);
+          aio_comp->add_request();
+          aio_comp->complete_request(r);
+        }));
+  }
+};
+
+TEST_F(TestMockIoObjectRequest, Read) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->sparse_read_threshold_bytes = 8096;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_get_read_flags(mock_image_ctx, CEPH_NOSNAP, 0);
+  expect_read(mock_image_ctx, ictx->get_object_name(0), 0, 4096,
+              std::string(4096, '1'), 0);
+  expect_read(mock_image_ctx, ictx->get_object_name(0), 8192, 4096,
+              std::string(4096, '2'), 0);
+
+  C_SaferCond ctx;
+  uint64_t version;
+  ReadExtents extents = {{0, 4096}, {8192, 4096}};
+  auto req = MockObjectReadRequest::create(
+          &mock_image_ctx, 0, &extents,
+          mock_image_ctx.get_data_io_context(), 0, 0, {},
+          &version, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+
+  bufferlist expected_bl1;
+  expected_bl1.append(std::string(4096, '1'));
+  bufferlist expected_bl2;
+  expected_bl2.append(std::string(4096, '2'));
+
+  ASSERT_EQ(extents[0].extent_map.size(), 0);
+  ASSERT_EQ(extents[1].extent_map.size(), 0);
+  ASSERT_TRUE(extents[0].bl.contents_equal(expected_bl1));
+  ASSERT_TRUE(extents[1].bl.contents_equal(expected_bl2));
+}
+
+TEST_F(TestMockIoObjectRequest, SparseReadThreshold) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->sparse_read_threshold_bytes = ictx->get_object_size();
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_get_read_flags(mock_image_ctx, CEPH_NOSNAP, 0);
+  expect_sparse_read(mock_image_ctx, ictx->get_object_name(0), 0,
+                     ictx->sparse_read_threshold_bytes,
+                     std::string(ictx->sparse_read_threshold_bytes, '1'), 0);
+
+  C_SaferCond ctx;
+
+  ReadExtents extents = {{0, ictx->sparse_read_threshold_bytes}};
+  auto req = MockObjectReadRequest::create(
+    &mock_image_ctx, 0, &extents,
+    mock_image_ctx.get_data_io_context(), 0, 0, {}, nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, ReadError) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+  ictx->sparse_read_threshold_bytes = 8096;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_get_read_flags(mock_image_ctx, CEPH_NOSNAP, 0);
+  expect_read(mock_image_ctx, ictx->get_object_name(0), 0, 4096, "", -EPERM);
+
+  C_SaferCond ctx;
+  ReadExtents extents = {{0, 4096}};
+  auto req = MockObjectReadRequest::create(
+    &mock_image_ctx, 0, &extents,
+    mock_image_ctx.get_data_io_context(), 0, 0, {},
+    nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EPERM, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, ParentRead) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), RBD_FEATURE_LAYERING, &order));
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+  ictx->sparse_read_threshold_bytes = 8096;
+  ictx->clone_copy_on_read = false;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.parent = &mock_image_ctx;
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_get_read_flags(mock_image_ctx, CEPH_NOSNAP, 0);
+  expect_read(mock_image_ctx, ictx->get_object_name(0), 0, 4096, "", -ENOENT);
+
+  MockUtils mock_utils;
+  ReadExtents extents = {{0, 4096}};
+  expect_read_parent(mock_utils, 0, &extents, CEPH_NOSNAP, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectReadRequest::create(
+    &mock_image_ctx, 0, &extents,
+    mock_image_ctx.get_data_io_context(), 0, 0, {},
+    nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, ParentReadError) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), RBD_FEATURE_LAYERING, &order));
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+  ictx->sparse_read_threshold_bytes = 8096;
+  ictx->clone_copy_on_read = false;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.parent = &mock_image_ctx;
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_get_read_flags(mock_image_ctx, CEPH_NOSNAP, 0);
+  expect_read(mock_image_ctx, ictx->get_object_name(0), 0, 4096, "", -ENOENT);
+
+  MockUtils mock_utils;
+  ReadExtents extents = {{0, 4096}};
+  expect_read_parent(mock_utils, 0, &extents, CEPH_NOSNAP, -EPERM);
+
+  C_SaferCond ctx;
+  auto req = MockObjectReadRequest::create(
+    &mock_image_ctx, 0, &extents,
+    mock_image_ctx.get_data_io_context(), 0, 0, {},
+    nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-EPERM, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, SkipParentRead) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), RBD_FEATURE_LAYERING, &order));
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+  ictx->sparse_read_threshold_bytes = 8096;
+  ictx->clone_copy_on_read = false;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.parent = &mock_image_ctx;
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_get_read_flags(mock_image_ctx, CEPH_NOSNAP, 0);
+  expect_read(mock_image_ctx, ictx->get_object_name(0), 0, 4096, "", -ENOENT);
+
+  ReadExtents extents = {{0, 4096}};
+  C_SaferCond ctx;
+  auto req = MockObjectReadRequest::create(
+    &mock_image_ctx, 0, &extents, mock_image_ctx.get_data_io_context(), 0,
+    READ_FLAG_DISABLE_READ_FROM_PARENT, {}, nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(-ENOENT, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, CopyOnRead) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), RBD_FEATURE_LAYERING, &order));
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+  ictx->sparse_read_threshold_bytes = 8096;
+  ictx->clone_copy_on_read = true;
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.parent = &mock_image_ctx;
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_get_read_flags(mock_image_ctx, CEPH_NOSNAP, 0);
+  expect_read(mock_image_ctx, ictx->get_object_name(0), 0, 4096, "", -ENOENT);
+
+  MockUtils mock_utils;
+  ReadExtents extents = {{0, 4096}};
+  expect_read_parent(mock_utils, 0, &extents, CEPH_NOSNAP, 0);
+
+  MockCopyupRequest mock_copyup_request;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 4096, 0);
+  expect_prune_parent_extents(mock_image_ctx, {{0, 4096}}, 4096, 4096);
+  expect_copyup(mock_copyup_request, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectReadRequest::create(
+    &mock_image_ctx, 0, &extents,
+    mock_image_ctx.get_data_io_context(), 0, 0, {},
+    nullptr, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, Write) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_write(mock_image_ctx, 0, 4096, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectWriteRequest::create_write(
+    &mock_image_ctx, 0, 0, std::move(bl), mock_image_ctx.get_data_io_context(),
+    0, 0, std::nullopt, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, WriteWithCreateExclusiveFlag) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  // exclusive create should succeed
+  {
+    bufferlist bl;
+    bl.append(std::string(4096, '1'));
+
+    InSequence seq;
+    expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+    expect_object_may_exist(mock_image_ctx, 0, true);
+    expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false,
+                             0);
+    expect_write(mock_image_ctx, 0, 4096, 0);
+
+    C_SaferCond ctx;
+    auto req = MockObjectWriteRequest::create_write(
+            &mock_image_ctx, 0, 0, std::move(bl),
+            mock_image_ctx.get_data_io_context(), 0,
+            OBJECT_WRITE_FLAG_CREATE_EXCLUSIVE, std::nullopt, {}, &ctx);
+    req->send();
+    ASSERT_EQ(0, ctx.wait());
+  }
+
+  // exclusive create should fail since object already exists
+  {
+    bufferlist bl;
+    bl.append(std::string(4096, '1'));
+
+    InSequence seq;
+    expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+    expect_object_may_exist(mock_image_ctx, 0, true);
+    expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false,
+                             0);
+
+    C_SaferCond ctx;
+    auto req = MockObjectWriteRequest::create_write(
+            &mock_image_ctx, 0, 0, std::move(bl),
+            mock_image_ctx.get_data_io_context(), 0,
+            OBJECT_WRITE_FLAG_CREATE_EXCLUSIVE, std::nullopt, {}, &ctx);
+    req->send();
+    ASSERT_EQ(-EEXIST, ctx.wait());
+  }
+}
+
+TEST_F(TestMockIoObjectRequest, WriteWithAssertVersion) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  // write an object
+  {
+    bufferlist bl;
+    bl.append(std::string(4096, '1'));
+
+    InSequence seq;
+    expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+    expect_object_may_exist(mock_image_ctx, 0, true);
+    expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false,
+                             0);
+    expect_write(mock_image_ctx, 0, 4096, 0);
+
+    C_SaferCond ctx;
+    auto req = MockObjectWriteRequest::create_write(
+            &mock_image_ctx, 0, 0, std::move(bl),
+            mock_image_ctx.get_data_io_context(), 0, 0,
+            std::nullopt, {}, &ctx);
+    req->send();
+    ASSERT_EQ(0, ctx.wait());
+  }
+
+  // assert version should succeed (version = 1)
+  {
+    bufferlist bl;
+    bl.append(std::string(4096, '1'));
+
+    InSequence seq;
+    expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+    expect_object_may_exist(mock_image_ctx, 0, true);
+    expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false,
+                             0);
+    expect_write(mock_image_ctx, 0, 4096, 0);
+
+    C_SaferCond ctx;
+    auto req = MockObjectWriteRequest::create_write(
+            &mock_image_ctx, 0, 0, std::move(bl),
+            mock_image_ctx.get_data_io_context(), 0, 0,
+            std::make_optional(1), {}, &ctx);
+    req->send();
+    ASSERT_EQ(0, ctx.wait());
+  }
+
+  // assert with wrong (lower) version (version = 2)
+  {
+    bufferlist bl;
+    bl.append(std::string(4096, '1'));
+
+    InSequence seq;
+    expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+    expect_object_may_exist(mock_image_ctx, 0, true);
+    expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false,
+                             0);
+
+    C_SaferCond ctx;
+    auto req = MockObjectWriteRequest::create_write(
+            &mock_image_ctx, 0, 0, std::move(bl),
+            mock_image_ctx.get_data_io_context(), 0, 0,
+            std::make_optional(1), {}, &ctx);
+    req->send();
+    ASSERT_EQ(-ERANGE, ctx.wait());
+  }
+
+  // assert with wrong (higher) version (version = 2)
+  {
+    bufferlist bl;
+    bl.append(std::string(4096, '1'));
+
+    InSequence seq;
+    expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+    expect_object_may_exist(mock_image_ctx, 0, true);
+    expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false,
+                             0);
+
+    C_SaferCond ctx;
+    auto req = MockObjectWriteRequest::create_write(
+            &mock_image_ctx, 0, 0, std::move(bl),
+            mock_image_ctx.get_data_io_context(), 0, 0, std::make_optional(3),
+            {}, &ctx);
+    req->send();
+    ASSERT_EQ(-EOVERFLOW, ctx.wait());
+  }
+}
+
+TEST_F(TestMockIoObjectRequest, WriteFull) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  bufferlist bl;
+  bl.append(std::string(ictx->get_object_size(), '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_write_full(mock_image_ctx, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectWriteRequest::create_write(
+    &mock_image_ctx, 0, 0, std::move(bl), mock_image_ctx.get_data_io_context(),
+    0, 0, std::nullopt, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, WriteObjectMap) {
+  REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_op_work_queue(mock_image_ctx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+  expect_is_lock_owner(mock_exclusive_lock);
+
+  MockObjectMap mock_object_map;
+  mock_image_ctx.object_map = &mock_object_map;
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, true, 0);
+  expect_write(mock_image_ctx, 0, 4096, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectWriteRequest::create_write(
+    &mock_image_ctx, 0, 0, std::move(bl), mock_image_ctx.get_data_io_context(),
+    0, 0, std::nullopt, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, WriteError) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_write(mock_image_ctx, 0, 4096, -EPERM);
+
+  C_SaferCond ctx;
+  auto req = MockObjectWriteRequest::create_write(
+    &mock_image_ctx, 0, 0, std::move(bl), mock_image_ctx.get_data_io_context(),
+    0, 0, std::nullopt, {}, &ctx);
+  req->send();
+  ASSERT_EQ(-EPERM, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, Copyup) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), RBD_FEATURE_LAYERING, &order));
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 4096, 0);
+  expect_prune_parent_extents(mock_image_ctx, {{0, 4096}}, 4096, 4096);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_assert_exists(mock_image_ctx, -ENOENT);
+
+  MockAbstractObjectWriteRequest *write_request = nullptr;
+  MockCopyupRequest mock_copyup_request;
+  expect_copyup(mock_copyup_request, &write_request, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectWriteRequest::create_write(
+    &mock_image_ctx, 0, 0, std::move(bl), mock_image_ctx.get_data_io_context(),
+    0, 0, std::nullopt, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, CopyupRestart) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), RBD_FEATURE_LAYERING, &order));
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 4096, 0);
+  expect_prune_parent_extents(mock_image_ctx, {{0, 4096}}, 4096, 4096);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_assert_exists(mock_image_ctx, -ENOENT);
+
+  MockAbstractObjectWriteRequest *write_request = nullptr;
+  MockCopyupRequest mock_copyup_request;
+  expect_copyup(mock_copyup_request, &write_request, -ERESTART);
+  expect_assert_exists(mock_image_ctx, 0);
+  expect_write(mock_image_ctx, 0, 4096, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectWriteRequest::create_write(
+    &mock_image_ctx, 0, 0, std::move(bl), mock_image_ctx.get_data_io_context(),
+    0, 0, std::nullopt, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, CopyupOptimization) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING | RBD_FEATURE_OBJECT_MAP);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), RBD_FEATURE_LAYERING, &order));
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+  expect_is_lock_owner(mock_exclusive_lock);
+
+  MockObjectMap mock_object_map;
+  mock_image_ctx.object_map = &mock_object_map;
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 4096, 0);
+  expect_prune_parent_extents(mock_image_ctx, {{0, 4096}}, 4096, 4096);
+  expect_object_may_exist(mock_image_ctx, 0, false);
+
+  MockAbstractObjectWriteRequest *write_request = nullptr;
+  MockCopyupRequest mock_copyup_request;
+  expect_copyup(mock_copyup_request, &write_request, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectWriteRequest::create_write(
+    &mock_image_ctx, 0, 0, std::move(bl), mock_image_ctx.get_data_io_context(),
+    0, 0, std::nullopt, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, CopyupError) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), RBD_FEATURE_LAYERING, &order));
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 4096, 0);
+  expect_prune_parent_extents(mock_image_ctx, {{0, 4096}}, 4096, 4096);
+  expect_assert_exists(mock_image_ctx, -ENOENT);
+
+  MockAbstractObjectWriteRequest *write_request = nullptr;
+  MockCopyupRequest mock_copyup_request;
+  expect_copyup(mock_copyup_request, &write_request, -EPERM);
+
+  C_SaferCond ctx;
+  auto req = MockObjectWriteRequest::create_write(
+    &mock_image_ctx, 0, 0, std::move(bl), mock_image_ctx.get_data_io_context(),
+    0, 0, std::nullopt, {}, &ctx);
+  req->send();
+  ASSERT_EQ(-EPERM, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, DiscardRemove) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_PENDING, {}, false, 0);
+  expect_remove(mock_image_ctx, 0);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_NONEXISTENT,
+                           OBJECT_PENDING, false, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectDiscardRequest::create_discard(
+    &mock_image_ctx, 0, 0, mock_image_ctx.get_object_size(),
+    mock_image_ctx.get_data_io_context(), 0, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, DiscardRemoveTruncate) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  uint64_t features;
+  ASSERT_EQ(0, image.features(&features));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), features, &order));
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 4096, 0);
+  expect_prune_parent_extents(mock_image_ctx, {{0, 4096}}, 4096, 4096);
+  expect_object_may_exist(mock_image_ctx, 0, false);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_create(mock_image_ctx, false);
+  expect_truncate(mock_image_ctx, 0, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectDiscardRequest::create_discard(
+    &mock_image_ctx, 0, 0, mock_image_ctx.get_object_size(),
+    mock_image_ctx.get_data_io_context(),
+    OBJECT_DISCARD_FLAG_DISABLE_CLONE_REMOVE, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, DiscardTruncateAssertExists) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  uint64_t features;
+  ASSERT_EQ(0, image.features(&features));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), features, &order));
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, clone_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  image.close();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 4096, 0);
+  expect_prune_parent_extents(mock_image_ctx, {{0, 4096}}, 4096, 4096);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_assert_exists(mock_image_ctx, 0);
+  expect_truncate(mock_image_ctx, 0, 0);
+
+  EXPECT_CALL(get_mock_io_ctx(mock_image_ctx.data_ctx), create(_, _, _))
+    .Times(0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectDiscardRequest::create_discard(
+    &mock_image_ctx, 0, 0, mock_image_ctx.get_object_size(),
+    mock_image_ctx.get_data_io_context(),
+    OBJECT_DISCARD_FLAG_DISABLE_CLONE_REMOVE, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, DiscardTruncate) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_truncate(mock_image_ctx, 1, 0);
+
+  EXPECT_CALL(get_mock_io_ctx(mock_image_ctx.data_ctx), create(_, _, _))
+    .Times(0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectDiscardRequest::create_discard(
+    &mock_image_ctx, 0, 1, mock_image_ctx.get_object_size() - 1,
+    mock_image_ctx.get_data_io_context(), 0, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, DiscardZero) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_zero(mock_image_ctx, 1, 1, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectDiscardRequest::create_discard(
+    &mock_image_ctx, 0, 1, 1, mock_image_ctx.get_data_io_context(), 0, {},
+    &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, DiscardDisableObjectMapUpdate) {
+  REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+  expect_is_lock_owner(mock_exclusive_lock);
+
+  MockObjectMap mock_object_map;
+  mock_image_ctx.object_map = &mock_object_map;
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_remove(mock_image_ctx, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectDiscardRequest::create_discard(
+    &mock_image_ctx, 0, 0, mock_image_ctx.get_object_size(),
+    mock_image_ctx.get_data_io_context(),
+    OBJECT_DISCARD_FLAG_DISABLE_CLONE_REMOVE |
+      OBJECT_DISCARD_FLAG_DISABLE_OBJECT_MAP_UPDATE, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, DiscardNoOp) {
+  REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_op_work_queue(mock_image_ctx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+  expect_is_lock_owner(mock_exclusive_lock);
+
+  MockObjectMap mock_object_map;
+  mock_image_ctx.object_map = &mock_object_map;
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, false);
+
+  C_SaferCond ctx;
+  auto req = MockObjectDiscardRequest::create_discard(
+    &mock_image_ctx, 0, 0, mock_image_ctx.get_object_size(),
+    mock_image_ctx.get_data_io_context(),
+    OBJECT_DISCARD_FLAG_DISABLE_CLONE_REMOVE |
+      OBJECT_DISCARD_FLAG_DISABLE_OBJECT_MAP_UPDATE, {},
+    &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, WriteSame) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_writesame(mock_image_ctx, 0, 4096, 0);
+
+  C_SaferCond ctx;
+  auto req = MockObjectWriteSameRequest::create_write_same(
+    &mock_image_ctx, 0, 0, 4096, std::move(bl),
+    mock_image_ctx.get_data_io_context(), 0, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, CompareAndWrite) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  bufferlist cmp_bl;
+  cmp_bl.append_zero(4096);
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_cmpext(mock_image_ctx, 0, 0);
+  expect_write(mock_image_ctx, 0, 4096, 0);
+
+  C_SaferCond ctx;
+  uint64_t mismatch_offset;
+  auto req = MockObjectWriteSameRequest::create_compare_and_write(
+    &mock_image_ctx, 0, 0, std::move(cmp_bl), std::move(bl),
+    mock_image_ctx.get_data_io_context(), &mismatch_offset, 0, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, CompareAndWriteFull) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  bufferlist cmp_bl;
+  cmp_bl.append_zero(ictx->get_object_size());
+
+  bufferlist bl;
+  bl.append(std::string(ictx->get_object_size(), '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_cmpext(mock_image_ctx, 0, 0);
+  expect_write_full(mock_image_ctx, 0);
+
+  C_SaferCond ctx;
+  uint64_t mismatch_offset;
+  auto req = MockObjectWriteSameRequest::create_compare_and_write(
+    &mock_image_ctx, 0, 0, std::move(cmp_bl), std::move(bl),
+    mock_image_ctx.get_data_io_context(), &mismatch_offset, 0, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, CompareAndWriteCopyup) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librbd::Image image;
+  librbd::RBD rbd;
+  ASSERT_EQ(0, rbd.open(m_ioctx, image, m_image_name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+  image.close();
+
+  std::string clone_name = get_temp_image_name();
+  int order = 0;
+  ASSERT_EQ(0, rbd.clone(m_ioctx, m_image_name.c_str(), "one", m_ioctx,
+                         clone_name.c_str(), RBD_FEATURE_LAYERING, &order));
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(clone_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  bufferlist cmp_bl;
+  cmp_bl.append_zero(4096);
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 4096, 0);
+  expect_prune_parent_extents(mock_image_ctx, {{0, 4096}}, 4096, 4096);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_assert_exists(mock_image_ctx, -ENOENT);
+
+  MockAbstractObjectWriteRequest *write_request = nullptr;
+  MockCopyupRequest mock_copyup_request;
+  expect_copyup(mock_copyup_request, &write_request, 0);
+
+  expect_assert_exists(mock_image_ctx, 0);
+  expect_cmpext(mock_image_ctx, 0, 0);
+  expect_write(mock_image_ctx, 0, 4096, 0);
+
+  C_SaferCond ctx;
+  uint64_t mismatch_offset;
+  auto req = MockObjectWriteSameRequest::create_compare_and_write(
+    &mock_image_ctx, 0, 0, std::move(cmp_bl), std::move(bl),
+    mock_image_ctx.get_data_io_context(), &mismatch_offset, 0, {}, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, CompareAndWriteMismatch) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  if (ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
+    mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+    expect_is_lock_owner(mock_exclusive_lock);
+  }
+
+  MockObjectMap mock_object_map;
+  if (ictx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    mock_image_ctx.object_map = &mock_object_map;
+  }
+
+  bufferlist cmp_bl;
+  cmp_bl.append_zero(4096);
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, false, 0);
+  expect_cmpext(mock_image_ctx, 0, -MAX_ERRNO - 1);
+
+  C_SaferCond ctx;
+  uint64_t mismatch_offset;
+  auto req = MockObjectWriteSameRequest::create_compare_and_write(
+    &mock_image_ctx, 0, 0, std::move(cmp_bl), std::move(bl),
+    mock_image_ctx.get_data_io_context(), &mismatch_offset, 0, {}, &ctx);
+  req->send();
+  ASSERT_EQ(-EILSEQ, ctx.wait());
+  ASSERT_EQ(1ULL, mismatch_offset);
+}
+
+TEST_F(TestMockIoObjectRequest, ObjectMapError) {
+  REQUIRE_FEATURE(RBD_FEATURE_OBJECT_MAP);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  expect_op_work_queue(mock_image_ctx);
+  expect_get_object_size(mock_image_ctx);
+
+  MockExclusiveLock mock_exclusive_lock;
+  mock_image_ctx.exclusive_lock = &mock_exclusive_lock;
+  expect_is_lock_owner(mock_exclusive_lock);
+
+  MockObjectMap mock_object_map;
+  mock_image_ctx.object_map = &mock_object_map;
+
+  bufferlist bl;
+  bl.append(std::string(4096, '1'));
+
+  InSequence seq;
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 0, 0);
+  expect_object_may_exist(mock_image_ctx, 0, true);
+  expect_object_map_update(mock_image_ctx, 0, 1, OBJECT_EXISTS, {}, true,
+                           -EBLOCKLISTED);
+
+  C_SaferCond ctx;
+  auto req = MockObjectWriteRequest::create_write(
+    &mock_image_ctx, 0, 0, std::move(bl), mock_image_ctx.get_data_io_context(),
+    0, 0, std::nullopt, {}, &ctx);
+  req->send();
+  ASSERT_EQ(-EBLOCKLISTED, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, ListSnaps) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.snaps = {3, 4, 5, 6, 7};
+
+  librados::snap_set_t snap_set;
+  snap_set.seq = 6;
+  librados::clone_info_t clone_info;
+
+  clone_info.cloneid = 3;
+  clone_info.snaps = {3};
+  clone_info.overlap = std::vector<std::pair<uint64_t,uint64_t>>{
+    {0, 4194304}};
+  clone_info.size = 4194304;
+  snap_set.clones.push_back(clone_info);
+
+  clone_info.cloneid = 4;
+  clone_info.snaps = {4};
+  clone_info.overlap = std::vector<std::pair<uint64_t,uint64_t>>{
+    {278528, 4096}, {442368, 4096}, {1859584, 4096}, {2224128, 4096},
+    {2756608, 4096}, {3227648, 4096}, {3739648, 4096}, {3903488, 4096}};
+  clone_info.size = 4194304;
+  snap_set.clones.push_back(clone_info);
+
+  clone_info.cloneid = 6;
+  clone_info.snaps = {5, 6};
+  clone_info.overlap = std::vector<std::pair<uint64_t,uint64_t>>{
+    {425984, 4096}, {440320, 1024}, {1925120, 4096}, {2125824, 4096},
+    {2215936, 5120}, {3067904, 4096}};
+  clone_info.size = 3072000;
+  snap_set.clones.push_back(clone_info);
+
+  clone_info.cloneid = CEPH_NOSNAP;
+  clone_info.snaps = {};
+  clone_info.overlap = {};
+  clone_info.size = 4194304;
+  snap_set.clones.push_back(clone_info);
+
+  expect_list_snaps(mock_image_ctx, snap_set, 0);
+
+  SnapshotDelta snapshot_delta;
+  C_SaferCond ctx;
+  auto req = MockObjectListSnapsRequest::create(
+    &mock_image_ctx, 0,
+    {{440320, 1024}, {2122728, 1024}, {2220032, 2048}, {3072000, 4096}},
+    {3, 4, 5, 6, 7, CEPH_NOSNAP}, 0, {}, &snapshot_delta, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+
+  SnapshotDelta expected_snapshot_delta;
+  expected_snapshot_delta[{5,6}].insert(
+    440320, 1024, {SPARSE_EXTENT_STATE_DATA, 1024});
+  expected_snapshot_delta[{5,6}].insert(
+    2122728, 1024, {SPARSE_EXTENT_STATE_DATA, 1024});
+  expected_snapshot_delta[{5,6}].insert(
+    2220032, 2048, {SPARSE_EXTENT_STATE_DATA, 2048});
+  expected_snapshot_delta[{7,CEPH_NOSNAP}].insert(
+    2122728, 1024, {SPARSE_EXTENT_STATE_DATA, 1024});
+  expected_snapshot_delta[{7,CEPH_NOSNAP}].insert(
+    2221056, 1024, {SPARSE_EXTENT_STATE_DATA, 1024});
+  expected_snapshot_delta[{7,CEPH_NOSNAP}].insert(
+    3072000, 4096, {SPARSE_EXTENT_STATE_DATA, 4096});
+  expected_snapshot_delta[{5,5}].insert(
+    3072000, 4096, {SPARSE_EXTENT_STATE_ZEROED, 4096});
+  ASSERT_EQ(expected_snapshot_delta, snapshot_delta);
+}
+
+TEST_F(TestMockIoObjectRequest, ListSnapsENOENT) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  expect_list_snaps(mock_image_ctx, {}, -ENOENT);
+
+  SnapshotDelta snapshot_delta;
+  C_SaferCond ctx;
+  auto req = MockObjectListSnapsRequest::create(
+    &mock_image_ctx, 0,
+    {{440320, 1024}},
+    {0, CEPH_NOSNAP}, 0, {}, &snapshot_delta, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+
+  SnapshotDelta expected_snapshot_delta;
+  expected_snapshot_delta[{0,0}].insert(
+    440320, 1024, {SPARSE_EXTENT_STATE_DNE, 1024});
+  ASSERT_EQ(expected_snapshot_delta, snapshot_delta);
+}
+
+TEST_F(TestMockIoObjectRequest, ListSnapsDNE) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.snaps = {2, 3, 4};
+
+  librados::snap_set_t snap_set;
+  snap_set.seq = 6;
+  librados::clone_info_t clone_info;
+
+  clone_info.cloneid = 4;
+  clone_info.snaps = {3, 4};
+  clone_info.overlap = std::vector<std::pair<uint64_t,uint64_t>>{
+    {0, 4194304}};
+  clone_info.size = 4194304;
+  snap_set.clones.push_back(clone_info);
+
+  expect_list_snaps(mock_image_ctx, snap_set, 0);
+
+  SnapshotDelta snapshot_delta;
+  C_SaferCond ctx;
+  auto req = MockObjectListSnapsRequest::create(
+    &mock_image_ctx, 0,
+    {{440320, 1024}},
+    {2, 3, 4}, 0, {}, &snapshot_delta, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+
+  SnapshotDelta expected_snapshot_delta;
+  expected_snapshot_delta[{2,2}].insert(
+    440320, 1024, {SPARSE_EXTENT_STATE_DNE, 1024});
+  expected_snapshot_delta[{3,4}].insert(
+    440320, 1024, {SPARSE_EXTENT_STATE_DATA, 1024});
+  ASSERT_EQ(expected_snapshot_delta, snapshot_delta);
+}
+
+TEST_F(TestMockIoObjectRequest, ListSnapsEmpty) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  expect_list_snaps(mock_image_ctx, {}, 0);
+
+  SnapshotDelta snapshot_delta;
+  C_SaferCond ctx;
+  auto req = MockObjectListSnapsRequest::create(
+    &mock_image_ctx, 0,
+    {{440320, 1024}},
+    {0, CEPH_NOSNAP}, 0, {}, &snapshot_delta, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+
+  SnapshotDelta expected_snapshot_delta;
+  expected_snapshot_delta[{0,0}].insert(
+    440320, 1024, {SPARSE_EXTENT_STATE_ZEROED, 1024});
+  ASSERT_EQ(expected_snapshot_delta, snapshot_delta);
+}
+
+TEST_F(TestMockIoObjectRequest, ListSnapsError) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  expect_list_snaps(mock_image_ctx, {}, -EPERM);
+
+  SnapshotDelta snapshot_delta;
+  C_SaferCond ctx;
+  auto req = MockObjectListSnapsRequest::create(
+    &mock_image_ctx, 0,
+    {{440320, 1024}, {2122728, 1024}, {2220032, 2048}, {3072000, 4096}},
+    {3, CEPH_NOSNAP}, 0, {}, &snapshot_delta, &ctx);
+  req->send();
+  ASSERT_EQ(-EPERM, ctx.wait());
+}
+
+TEST_F(TestMockIoObjectRequest, ListSnapsParent) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.parent = &mock_image_ctx;
+
+  InSequence seq;
+
+  expect_list_snaps(mock_image_ctx, {}, -ENOENT);
+
+  expect_get_parent_overlap(mock_image_ctx, CEPH_NOSNAP, 4096, 0);
+  expect_prune_parent_extents(mock_image_ctx, {{0, 4096}}, 4096, 4096);
+
+  MockImageListSnapsRequest mock_image_list_snaps_request;
+  SnapshotDelta image_snapshot_delta;
+  image_snapshot_delta[{1,6}].insert(
+    0, 1024, {SPARSE_EXTENT_STATE_DATA, 1024});
+  expect_image_list_snaps(mock_image_list_snaps_request,
+                          {{0, 4096}}, image_snapshot_delta, 0);
+
+  SnapshotDelta snapshot_delta;
+  C_SaferCond ctx;
+  auto req = MockObjectListSnapsRequest::create(
+    &mock_image_ctx, 0,
+    {{440320, 1024}, {2122728, 1024}, {2220032, 2048}, {3072000, 4096}},
+    {0, CEPH_NOSNAP}, 0, {}, &snapshot_delta, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+
+  SnapshotDelta expected_snapshot_delta;
+  expected_snapshot_delta[{0,0}].insert(
+    0, 1024, {SPARSE_EXTENT_STATE_DATA, 1024});
+  ASSERT_EQ(expected_snapshot_delta, snapshot_delta);
+}
+
+TEST_F(TestMockIoObjectRequest, ListSnapsWholeObject) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  mock_image_ctx.parent = &mock_image_ctx;
+
+  InSequence seq;
+
+  librados::snap_set_t snap_set;
+  snap_set.seq = 3;
+  librados::clone_info_t clone_info;
+
+  clone_info.cloneid = 3;
+  clone_info.snaps = {3};
+  clone_info.overlap = std::vector<std::pair<uint64_t,uint64_t>>{{0, 1}};
+  clone_info.size = 4194304;
+  snap_set.clones.push_back(clone_info);
+
+  clone_info.cloneid = CEPH_NOSNAP;
+  clone_info.snaps = {};
+  clone_info.overlap = {};
+  clone_info.size = 4194304;
+  snap_set.clones.push_back(clone_info);
+
+  expect_list_snaps(mock_image_ctx, snap_set, 0);
+
+  SnapshotDelta snapshot_delta;
+  C_SaferCond ctx;
+  auto req = MockObjectListSnapsRequest::create(
+    &mock_image_ctx, 0, {{0, mock_image_ctx.layout.object_size - 1}},
+    {0, CEPH_NOSNAP}, 0, {}, &snapshot_delta, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+
+  SnapshotDelta expected_snapshot_delta;
+  expected_snapshot_delta[{CEPH_NOSNAP,CEPH_NOSNAP}].insert(
+    0, mock_image_ctx.layout.object_size - 1,
+    {SPARSE_EXTENT_STATE_DATA, mock_image_ctx.layout.object_size - 1});
+  ASSERT_EQ(expected_snapshot_delta, snapshot_delta);
+}
+
+} // namespace io
+} // namespace librbd
+
diff --git a/src/test/librbd/io/test_mock_QosImageDispatch.cc b/src/test/librbd/io/test_mock_QosImageDispatch.cc
new file mode 100644
index 000000000..acd3b97c2
--- /dev/null
+++ b/src/test/librbd/io/test_mock_QosImageDispatch.cc
@@ -0,0 +1,89 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/librbd/test_mock_fixture.h"
+#include "test/librbd/test_support.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "test/librbd/mock/exclusive_lock/MockPolicy.h"
+#include "librbd/io/ImageDispatchSpec.h"
+#include "librbd/io/ImageRequestWQ.h"
+#include "librbd/io/ImageRequest.h"
+
+namespace librbd {
+namespace io {
+
+TEST_F(TestMockIoImageRequestWQ, QosNoLimit) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  MockImageDispatchSpec mock_queued_image_request;
+  expect_was_throttled(mock_queued_image_request, false);
+  expect_set_throttled(mock_queued_image_request);
+
+  InSequence seq;
+  MockImageRequestWQ mock_image_request_wq(&mock_image_ctx, "io", 60, nullptr);
+
+  mock_image_request_wq.apply_qos_limit(IMAGE_DISPATCH_FLAG_QOS_BPS_THROTTLE, 0,
+                                        0, 1);
+
+  expect_front(mock_image_request_wq, &mock_queued_image_request);
+  expect_is_refresh_request(mock_image_ctx, false);
+  expect_is_write_op(mock_queued_image_request, true);
+  expect_dequeue(mock_image_request_wq, &mock_queued_image_request);
+  ASSERT_TRUE(mock_image_request_wq.invoke_dequeue() == &mock_queued_image_request);
+}
+
+TEST_F(TestMockIoImageRequestWQ, BPSQosNoBurst) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  MockImageDispatchSpec mock_queued_image_request;
+  expect_was_throttled(mock_queued_image_request, false);
+  expect_set_throttled(mock_queued_image_request);
+
+  InSequence seq;
+  MockImageRequestWQ mock_image_request_wq(&mock_image_ctx, "io", 60, nullptr);
+
+  mock_image_request_wq.apply_qos_limit(IMAGE_DISPATCH_FLAG_QOS_BPS_THROTTLE, 1,
+                                        0, 1);
+
+  expect_front(mock_image_request_wq, &mock_queued_image_request);
+  expect_tokens_requested(mock_queued_image_request, 2, true);
+  expect_dequeue(mock_image_request_wq, &mock_queued_image_request);
+  expect_all_throttled(mock_queued_image_request, true);
+  expect_requeue_back(mock_image_request_wq);
+  expect_signal(mock_image_request_wq);
+  ASSERT_TRUE(mock_image_request_wq.invoke_dequeue() == nullptr);
+}
+
+TEST_F(TestMockIoImageRequestWQ, BPSQosWithBurst) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+
+  MockImageDispatchSpec mock_queued_image_request;
+  expect_was_throttled(mock_queued_image_request, false);
+  expect_set_throttled(mock_queued_image_request);
+
+  InSequence seq;
+  MockImageRequestWQ mock_image_request_wq(&mock_image_ctx, "io", 60, nullptr);
+
+  mock_image_request_wq.apply_qos_limit(IMAGE_DISPATCH_FLAG_QOS_BPS_THROTTLE, 1,
+                                        1, 1);
+
+  expect_front(mock_image_request_wq, &mock_queued_image_request);
+  expect_tokens_requested(mock_queued_image_request, 2, true);
+  expect_dequeue(mock_image_request_wq, &mock_queued_image_request);
+  expect_all_throttled(mock_queued_image_request, true);
+  expect_requeue_back(mock_image_request_wq);
+  expect_signal(mock_image_request_wq);
+  ASSERT_TRUE(mock_image_request_wq.invoke_dequeue() == nullptr);
+}
+
+} // namespace io
+} // namespace librbd
diff --git a/src/test/librbd/io/test_mock_SimpleSchedulerObjectDispatch.cc b/src/test/librbd/io/test_mock_SimpleSchedulerObjectDispatch.cc
new file mode 100644
index 000000000..a3d823d23
--- /dev/null
+++ b/src/test/librbd/io/test_mock_SimpleSchedulerObjectDispatch.cc
@@ -0,0 +1,823 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/librbd/test_mock_fixture.h"
+#include "test/librbd/test_support.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "test/librbd/mock/MockSafeTimer.h"
+#include "test/librados_test_stub/MockTestMemIoCtxImpl.h"
+#include "test/librados_test_stub/MockTestMemRadosClient.h"
+#include "include/rbd/librbd.hpp"
+#include "librbd/io/ObjectDispatchSpec.h"
+#include "librbd/io/SimpleSchedulerObjectDispatch.h"
+
+namespace librbd {
+namespace {
+
+struct MockTestImageCtx : public MockImageCtx {
+  MockTestImageCtx(ImageCtx &image_ctx) : MockImageCtx(image_ctx) {
+  }
+};
+
+} // anonymous namespace
+
+namespace io {
+
+template <>
+struct TypeTraits<MockTestImageCtx> {
+  typedef ::MockSafeTimer SafeTimer;
+};
+
+template <>
+struct FlushTracker<MockTestImageCtx> {
+  FlushTracker(MockTestImageCtx*) {
+  }
+
+  void shut_down() {
+  }
+
+  void flush(Context*) {
+  }
+
+  void start_io(uint64_t) {
+  }
+
+  void finish_io(uint64_t) {
+  }
+
+};
+
+} // namespace io
+} // namespace librbd
+
+#include "librbd/io/SimpleSchedulerObjectDispatch.cc"
+
+namespace librbd {
+namespace io {
+
+using ::testing::_;
+using ::testing::InSequence;
+using ::testing::Invoke;
+using ::testing::Return;
+
+struct TestMockIoSimpleSchedulerObjectDispatch : public TestMockFixture {
+  typedef SimpleSchedulerObjectDispatch<librbd::MockTestImageCtx> MockSimpleSchedulerObjectDispatch;
+
+  MockSafeTimer m_mock_timer;
+  ceph::mutex m_mock_timer_lock =
+    ceph::make_mutex("TestMockIoSimpleSchedulerObjectDispatch::Mutex");
+
+  TestMockIoSimpleSchedulerObjectDispatch() {
+    MockTestImageCtx::set_timer_instance(&m_mock_timer, &m_mock_timer_lock);
+    EXPECT_EQ(0, _rados.conf_set("rbd_io_scheduler_simple_max_delay", "1"));
+  }
+
+  void expect_get_object_name(MockTestImageCtx &mock_image_ctx,
+                              uint64_t object_no) {
+    EXPECT_CALL(mock_image_ctx, get_object_name(object_no))
+      .WillRepeatedly(Return(
+          mock_image_ctx.image_ctx->get_object_name(object_no)));
+  }
+
+  void expect_dispatch_delayed_requests(MockTestImageCtx &mock_image_ctx,
+                                        int r) {
+    EXPECT_CALL(*mock_image_ctx.io_object_dispatcher, send(_))
+      .WillOnce(Invoke([&mock_image_ctx, r](ObjectDispatchSpec* spec) {
+                  spec->dispatch_result = io::DISPATCH_RESULT_COMPLETE;
+                  mock_image_ctx.image_ctx->op_work_queue->queue(
+                      &spec->dispatcher_ctx, r);
+                }));
+  }
+
+  void expect_cancel_timer_task(Context *timer_task) {
+      EXPECT_CALL(m_mock_timer, cancel_event(timer_task))
+        .WillOnce(Invoke([](Context *timer_task) {
+                    delete timer_task;
+                    return true;
+                  }));
+  }
+
+  void expect_add_timer_task(Context **timer_task) {
+    EXPECT_CALL(m_mock_timer, add_event_at(_, _))
+      .WillOnce(Invoke([timer_task](ceph::real_clock::time_point, Context *task) {
+                  *timer_task = task;
+                  return task;
+                }));
+  }
+
+  void expect_schedule_dispatch_delayed_requests(Context *current_task,
+                                                 Context **new_task) {
+    if (current_task != nullptr) {
+      expect_cancel_timer_task(current_task);
+    }
+    if (new_task != nullptr) {
+      expect_add_timer_task(new_task);
+    }
+  }
+
+  void run_timer_task(Context *timer_task) {
+    std::lock_guard timer_locker{m_mock_timer_lock};
+    timer_task->complete(0);
+  }
+};
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, Read) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  C_SaferCond cond;
+  Context *on_finish = &cond;
+  io::ReadExtents extents = {{0, 4096}, {8192, 4096}};
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.read(
+      0, &extents, mock_image_ctx.get_data_io_context(), 0, 0, {}, nullptr,
+      nullptr, nullptr, &on_finish, nullptr));
+  ASSERT_EQ(on_finish, &cond); // not modified
+  on_finish->complete(0);
+  ASSERT_EQ(0, cond.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, Discard) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  C_SaferCond cond;
+  Context *on_finish = &cond;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.discard(
+      0, 0, 4096, mock_image_ctx.get_data_io_context(), 0, {}, nullptr, nullptr,
+      nullptr, &on_finish, nullptr));
+  ASSERT_NE(on_finish, &cond);
+  on_finish->complete(0);
+  ASSERT_EQ(0, cond.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, Write) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  ceph::bufferlist data;
+  data.append("X");
+  int object_dispatch_flags = 0;
+  C_SaferCond cond;
+  Context *on_finish = &cond;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write(
+      0, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, nullptr,
+      &on_finish, nullptr));
+  ASSERT_NE(on_finish, &cond);
+  on_finish->complete(0);
+  ASSERT_EQ(0, cond.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, WriteSame) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  io::LightweightBufferExtents buffer_extents;
+  ceph::bufferlist data;
+  C_SaferCond cond;
+  Context *on_finish = &cond;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write_same(
+      0, 0, 4096, std::move(buffer_extents), std::move(data),
+      mock_image_ctx.get_data_io_context(), 0, {}, nullptr, nullptr, nullptr,
+      &on_finish, nullptr));
+  ASSERT_NE(on_finish, &cond);
+  on_finish->complete(0);
+  ASSERT_EQ(0, cond.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, CompareAndWrite) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  ceph::bufferlist cmp_data;
+  ceph::bufferlist write_data;
+  C_SaferCond cond;
+  Context *on_finish = &cond;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.compare_and_write(
+      0, 0, std::move(cmp_data), std::move(write_data),
+      mock_image_ctx.get_data_io_context(), 0, {}, nullptr, nullptr, nullptr,
+      nullptr, &on_finish, nullptr));
+  ASSERT_NE(on_finish, &cond);
+  on_finish->complete(0);
+  ASSERT_EQ(0, cond.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, Flush) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  io::DispatchResult dispatch_result;
+  C_SaferCond cond;
+  Context *on_finish = &cond;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.flush(
+      FLUSH_SOURCE_USER, {}, nullptr, &dispatch_result, &on_finish, nullptr));
+  ASSERT_EQ(io::DISPATCH_RESULT_CONTINUE, dispatch_result);
+  ASSERT_EQ(on_finish, &cond); // not modified
+  on_finish->complete(0);
+  ASSERT_EQ(0, cond.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, WriteDelayed) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  expect_get_object_name(mock_image_ctx, 0);
+
+  InSequence seq;
+
+  ceph::bufferlist data;
+  data.append("X");
+  int object_dispatch_flags = 0;
+  C_SaferCond cond1;
+  Context *on_finish1 = &cond1;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write(
+      0, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, nullptr, &on_finish1,
+      nullptr));
+  ASSERT_NE(on_finish1, &cond1);
+
+  Context *timer_task = nullptr;
+  expect_schedule_dispatch_delayed_requests(nullptr, &timer_task);
+
+  io::DispatchResult dispatch_result;
+  C_SaferCond cond2;
+  Context *on_finish2 = &cond2;
+  C_SaferCond on_dispatched;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish2, &on_dispatched));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish2, &cond2);
+  ASSERT_NE(timer_task, nullptr);
+
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+  expect_schedule_dispatch_delayed_requests(timer_task, nullptr);
+
+  on_finish1->complete(0);
+  ASSERT_EQ(0, cond1.wait());
+  ASSERT_EQ(0, on_dispatched.wait());
+  on_finish2->complete(0);
+  ASSERT_EQ(0, cond2.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, WriteDelayedFlush) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  expect_get_object_name(mock_image_ctx, 0);
+
+  InSequence seq;
+
+  ceph::bufferlist data;
+  data.append("X");
+  int object_dispatch_flags = 0;
+  C_SaferCond cond1;
+  Context *on_finish1 = &cond1;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write(
+      0, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, nullptr, &on_finish1,
+      nullptr));
+  ASSERT_NE(on_finish1, &cond1);
+
+  Context *timer_task = nullptr;
+  expect_schedule_dispatch_delayed_requests(nullptr, &timer_task);
+
+  io::DispatchResult dispatch_result;
+  C_SaferCond cond2;
+  Context *on_finish2 = &cond2;
+  C_SaferCond on_dispatched;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish2, &on_dispatched));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish2, &cond2);
+  ASSERT_NE(timer_task, nullptr);
+
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+  expect_schedule_dispatch_delayed_requests(timer_task, nullptr);
+
+  C_SaferCond cond3;
+  Context *on_finish3 = &cond3;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.flush(
+      FLUSH_SOURCE_USER, {}, nullptr, &dispatch_result, &on_finish3, nullptr));
+  ASSERT_EQ(io::DISPATCH_RESULT_CONTINUE, dispatch_result);
+  ASSERT_EQ(on_finish3, &cond3);
+
+  on_finish1->complete(0);
+  ASSERT_EQ(0, cond1.wait());
+  ASSERT_EQ(0, on_dispatched.wait());
+  on_finish2->complete(0);
+  ASSERT_EQ(0, cond2.wait());
+  on_finish3->complete(0);
+  ASSERT_EQ(0, cond3.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, WriteMerged) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  expect_get_object_name(mock_image_ctx, 0);
+
+  InSequence seq;
+
+  ceph::bufferlist data;
+  data.append("X");
+  int object_dispatch_flags = 0;
+  C_SaferCond cond1;
+  Context *on_finish1 = &cond1;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write(
+      0, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, nullptr, &on_finish1,
+      nullptr));
+  ASSERT_NE(on_finish1, &cond1);
+
+  Context *timer_task = nullptr;
+  expect_schedule_dispatch_delayed_requests(nullptr, &timer_task);
+
+  uint64_t object_off = 20;
+  data.clear();
+  data.append(std::string(10, 'A'));
+  io::DispatchResult dispatch_result;
+  C_SaferCond cond2;
+  Context *on_finish2 = &cond2;
+  C_SaferCond on_dispatched2;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish2, &on_dispatched2));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish2, &cond2);
+  ASSERT_NE(timer_task, nullptr);
+
+  object_off = 0;
+  data.clear();
+  data.append(std::string(10, 'B'));
+  C_SaferCond cond3;
+  Context *on_finish3 = &cond3;
+  C_SaferCond on_dispatched3;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish3, &on_dispatched3));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish3, &cond3);
+
+  object_off = 10;
+  data.clear();
+  data.append(std::string(10, 'C'));
+  C_SaferCond cond4;
+  Context *on_finish4 = &cond4;
+  C_SaferCond on_dispatched4;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {},&object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish4, &on_dispatched4));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish4, &cond4);
+
+  object_off = 30;
+  data.clear();
+  data.append(std::string(10, 'D'));
+  C_SaferCond cond5;
+  Context *on_finish5 = &cond5;
+  C_SaferCond on_dispatched5;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish5, &on_dispatched5));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish5, &cond5);
+
+  object_off = 50;
+  data.clear();
+  data.append(std::string(10, 'E'));
+  C_SaferCond cond6;
+  Context *on_finish6 = &cond6;
+  C_SaferCond on_dispatched6;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish6, &on_dispatched6));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish6, &cond6);
+
+  // expect two requests dispatched:
+  // 0~40 (merged 0~10, 10~10, 20~10, 30~10) and 50~10
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+  expect_schedule_dispatch_delayed_requests(timer_task, nullptr);
+
+  on_finish1->complete(0);
+  ASSERT_EQ(0, cond1.wait());
+  ASSERT_EQ(0, on_dispatched2.wait());
+  ASSERT_EQ(0, on_dispatched3.wait());
+  ASSERT_EQ(0, on_dispatched4.wait());
+  ASSERT_EQ(0, on_dispatched5.wait());
+  ASSERT_EQ(0, on_dispatched6.wait());
+  on_finish2->complete(0);
+  on_finish3->complete(0);
+  on_finish4->complete(0);
+  on_finish5->complete(0);
+  on_finish6->complete(0);
+  ASSERT_EQ(0, cond2.wait());
+  ASSERT_EQ(0, cond3.wait());
+  ASSERT_EQ(0, cond4.wait());
+  ASSERT_EQ(0, cond5.wait());
+  ASSERT_EQ(0, cond6.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, WriteNonSequential) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  expect_get_object_name(mock_image_ctx, 0);
+
+  InSequence seq;
+
+  ceph::bufferlist data;
+  int object_dispatch_flags = 0;
+  C_SaferCond cond1;
+  Context *on_finish1 = &cond1;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write(
+      0, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, nullptr, &on_finish1,
+      nullptr));
+  ASSERT_NE(on_finish1, &cond1);
+
+  Context *timer_task = nullptr;
+  expect_schedule_dispatch_delayed_requests(nullptr, &timer_task);
+
+  uint64_t object_off = 0;
+  data.clear();
+  data.append(std::string(10, 'X'));
+  io::DispatchResult dispatch_result;
+  C_SaferCond cond2;
+  Context *on_finish2 = &cond2;
+  C_SaferCond on_dispatched2;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish2, &on_dispatched2));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish2, &cond2);
+  ASSERT_NE(timer_task, nullptr);
+
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+  expect_schedule_dispatch_delayed_requests(timer_task, nullptr);
+
+  object_off = 5;
+  data.clear();
+  data.append(std::string(10, 'Y'));
+  C_SaferCond cond3;
+  Context *on_finish3 = &cond3;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish3, nullptr));
+  ASSERT_NE(on_finish3, &cond3);
+
+  on_finish1->complete(0);
+  ASSERT_EQ(0, cond1.wait());
+  ASSERT_EQ(0, on_dispatched2.wait());
+  on_finish2->complete(0);
+  ASSERT_EQ(0, cond2.wait());
+  on_finish3->complete(0);
+  ASSERT_EQ(0, cond3.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, Mixed) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  expect_get_object_name(mock_image_ctx, 0);
+
+  InSequence seq;
+
+  // write (1) 0~0 (in-flight)
+  // will wrap on_finish with dispatch_seq=1 to dispatch future delayed writes
+  ceph::bufferlist data;
+  int object_dispatch_flags = 0;
+  C_SaferCond cond1;
+  Context *on_finish1 = &cond1;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write(
+      0, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, nullptr, &on_finish1,
+      nullptr));
+  ASSERT_NE(on_finish1, &cond1);
+
+  // write (2) 0~10 (delayed)
+  // will wait for write (1) to finish or a non-seq io comes
+  Context *timer_task = nullptr;
+  expect_schedule_dispatch_delayed_requests(nullptr, &timer_task);
+  uint64_t object_off = 0;
+  data.clear();
+  data.append(std::string(10, 'A'));
+  io::DispatchResult dispatch_result;
+  C_SaferCond cond2;
+  Context *on_finish2 = &cond2;
+  C_SaferCond on_dispatched2;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish2, &on_dispatched2));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish2, &cond2);
+  ASSERT_NE(timer_task, nullptr);
+
+  // write (3) 10~10 (delayed)
+  // will be merged with write (2)
+  object_off = 10;
+  data.clear();
+  data.append(std::string(10, 'B'));
+  C_SaferCond cond3;
+  Context *on_finish3 = &cond3;
+  C_SaferCond on_dispatched3;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish3, &on_dispatched3));
+  ASSERT_NE(on_finish3, &cond3);
+
+  // discard (1) (non-seq io)
+  // will dispatch the delayed writes (2) and (3) and wrap on_finish
+  // with dispatch_seq=2 to dispatch future delayed writes
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+  expect_schedule_dispatch_delayed_requests(timer_task, nullptr);
+  C_SaferCond cond4;
+  Context *on_finish4 = &cond4;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.discard(
+      0, 4096, 4096, mock_image_ctx.get_data_io_context(), 0, {}, nullptr,
+      nullptr, nullptr, &on_finish4, nullptr));
+  ASSERT_NE(on_finish4, &cond4);
+  ASSERT_EQ(0, on_dispatched2.wait());
+  ASSERT_EQ(0, on_dispatched3.wait());
+
+  // write (4) 20~10 (delayed)
+  // will wait for discard (1) to finish or a non-seq io comes
+  timer_task = nullptr;
+  expect_schedule_dispatch_delayed_requests(nullptr, &timer_task);
+  object_off = 20;
+  data.clear();
+  data.append(std::string(10, 'C'));
+  C_SaferCond cond5;
+  Context *on_finish5 = &cond5;
+  C_SaferCond on_dispatched5;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish5, &on_dispatched5));
+  ASSERT_NE(on_finish5, &cond5);
+  ASSERT_NE(timer_task, nullptr);
+
+  // discard (2) (non-seq io)
+  // will dispatch the delayed write (4) and wrap on_finish with dispatch_seq=3
+  // to dispatch future delayed writes
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+  expect_schedule_dispatch_delayed_requests(timer_task, nullptr);
+  C_SaferCond cond6;
+  Context *on_finish6 = &cond6;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.discard(
+      0, 4096, 4096, mock_image_ctx.get_data_io_context(), 0, {}, nullptr,
+      nullptr, nullptr, &on_finish6, nullptr));
+  ASSERT_NE(on_finish6, &cond6);
+  ASSERT_EQ(0, on_dispatched5.wait());
+
+  // write (5) 30~10 (delayed)
+  // will wait for discard (2) to finish or a non-seq io comes
+  timer_task = nullptr;
+  expect_schedule_dispatch_delayed_requests(nullptr, &timer_task);
+  object_off = 30;
+  data.clear();
+  data.append(std::string(10, 'D'));
+  C_SaferCond cond7;
+  Context *on_finish7 = &cond7;
+  C_SaferCond on_dispatched7;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, object_off, std::move(data), mock_image_ctx.get_data_io_context(), 0,
+      0, std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish7, &on_dispatched7));
+  ASSERT_NE(on_finish7, &cond7);
+  ASSERT_NE(timer_task, nullptr);
+
+  // write (1) finishes
+  // on_finish wrapper will skip dispatch delayed write (5)
+  // due to dispatch_seq(1) < m_dispatch_seq(3)
+  on_finish1->complete(0);
+  ASSERT_EQ(0, cond1.wait());
+
+  // writes (2) and (3) finish ("dispatch delayed" is not called)
+  on_finish2->complete(0);
+  on_finish3->complete(0);
+  ASSERT_EQ(0, cond2.wait());
+  ASSERT_EQ(0, cond3.wait());
+
+  // discard (1) finishes
+  // on_finish wrapper will skip dispatch delayed write (5)
+  // due to dispatch_seq(2) < m_dispatch_seq(3)
+  on_finish4->complete(0);
+  ASSERT_EQ(0, cond4.wait());
+
+  // writes (4) finishes ("dispatch delayed" is not called)
+  on_finish5->complete(0);
+  ASSERT_EQ(0, cond5.wait());
+
+  // discard (2) finishes
+  // on_finish wrapper will dispatch the delayed write (5)
+  // due to dispatch_seq(3) == m_dispatch_seq(3)
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+  expect_schedule_dispatch_delayed_requests(timer_task, nullptr);
+  on_finish6->complete(0);
+  ASSERT_EQ(0, cond6.wait());
+  ASSERT_EQ(0, on_dispatched7.wait());
+
+  // write (5) finishes ("dispatch delayed" is not called)
+  on_finish7->complete(0);
+  ASSERT_EQ(0, cond7.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, DispatchQueue) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  expect_get_object_name(mock_image_ctx, 0);
+  expect_get_object_name(mock_image_ctx, 1);
+
+  InSequence seq;
+
+  // send 2 writes to object 0
+
+  uint64_t object_no = 0;
+  ceph::bufferlist data;
+  int object_dispatch_flags = 0;
+  C_SaferCond cond1;
+  Context *on_finish1 = &cond1;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write(
+      object_no, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, nullptr, &on_finish1,
+      nullptr));
+  ASSERT_NE(on_finish1, &cond1);
+
+  Context *timer_task = nullptr;
+  expect_schedule_dispatch_delayed_requests(nullptr, &timer_task);
+
+  data.clear();
+  io::DispatchResult dispatch_result;
+  C_SaferCond cond2;
+  Context *on_finish2 = &cond2;
+  C_SaferCond on_dispatched2;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      object_no, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish2, &on_dispatched2));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish2, &cond2);
+  ASSERT_NE(timer_task, nullptr);
+
+  // send 2 writes to object 1
+
+  object_no = 1;
+  data.clear();
+  C_SaferCond cond3;
+  Context *on_finish3 = &cond3;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write(
+      object_no, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, nullptr, &on_finish3,
+      nullptr));
+  ASSERT_NE(on_finish3, &cond3);
+
+  data.clear();
+  C_SaferCond cond4;
+  Context *on_finish4 = &cond4;
+  C_SaferCond on_dispatched4;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      object_no, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish4, &on_dispatched4));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish4, &cond4);
+
+  // finish write (1) to object 0
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+  expect_schedule_dispatch_delayed_requests(timer_task, &timer_task);
+  on_finish1->complete(0);
+  ASSERT_EQ(0, cond1.wait());
+  ASSERT_EQ(0, on_dispatched2.wait());
+
+  // finish write (2) to object 0
+  on_finish2->complete(0);
+  ASSERT_EQ(0, cond2.wait());
+
+  // finish write (1) to object 1
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+  expect_schedule_dispatch_delayed_requests(timer_task, nullptr);
+  on_finish3->complete(0);
+  ASSERT_EQ(0, cond3.wait());
+  ASSERT_EQ(0, on_dispatched4.wait());
+
+  // finish write (2) to object 1
+  on_finish4->complete(0);
+  ASSERT_EQ(0, cond4.wait());
+}
+
+TEST_F(TestMockIoSimpleSchedulerObjectDispatch, Timer) {
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  MockSimpleSchedulerObjectDispatch
+      mock_simple_scheduler_object_dispatch(&mock_image_ctx);
+
+  expect_get_object_name(mock_image_ctx, 0);
+  expect_op_work_queue(mock_image_ctx);
+
+  InSequence seq;
+
+  ceph::bufferlist data;
+  int object_dispatch_flags = 0;
+  C_SaferCond cond1;
+  Context *on_finish1 = &cond1;
+  ASSERT_FALSE(mock_simple_scheduler_object_dispatch.write(
+      0, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, nullptr, &on_finish1,
+      nullptr));
+  ASSERT_NE(on_finish1, &cond1);
+
+  Context *timer_task = nullptr;
+  expect_schedule_dispatch_delayed_requests(nullptr, &timer_task);
+
+  data.clear();
+  io::DispatchResult dispatch_result;
+  C_SaferCond cond2;
+  Context *on_finish2 = &cond2;
+  C_SaferCond on_dispatched;
+  ASSERT_TRUE(mock_simple_scheduler_object_dispatch.write(
+      0, 0, std::move(data), mock_image_ctx.get_data_io_context(), 0, 0,
+      std::nullopt, {}, &object_dispatch_flags, nullptr, &dispatch_result,
+      &on_finish2, &on_dispatched));
+  ASSERT_EQ(dispatch_result, io::DISPATCH_RESULT_COMPLETE);
+  ASSERT_NE(on_finish2, &cond2);
+  ASSERT_NE(timer_task, nullptr);
+
+  expect_dispatch_delayed_requests(mock_image_ctx, 0);
+
+  run_timer_task(timer_task);
+  ASSERT_EQ(0, on_dispatched.wait());
+
+  on_finish1->complete(0);
+  ASSERT_EQ(0, cond1.wait());
+  on_finish2->complete(0);
+  ASSERT_EQ(0, cond2.wait());
+}
+
+} // namespace io
+} // namespace librbd