Fix UserEventSyncBridge not calling processor's DisableSync

The call is implemented in the base class's implementation,
ModelTypeSyncBridge::DisableSync(). Without such a call, the
processor never takes care of deleting the metadata.

Because UserEventSyncBridge is a commit-only type (in fact the
only one), it forces initial_sync_done when loading metadata. This
is now moved to the processor, because it's common for all
commit-only types and it also needs to be taken care of if
DisableSync() is followed by enable-sync.

In order to test this, some refactoring of tests was needed (and I
chose to introduce MockModelTypeChangeProcessor), because:
a) The former tests had bugs (didn't actually verify some values,
   because callbacks not being run was treated as success)

b) The DisableSync() flow (currently) destroys and recreates the
   change processor, which requires some forwarding proxy to allow
   tests to verify the state across destructions of the processor.

Bug: 819233
Change-Id: I0f1524850ef65b795b7f7082ce4e85171b8e2898
Reviewed-on: https://chromium-review.googlesource.com/951604
Commit-Queue: Mikel Astiz <mastiz@chromium.org>
Reviewed-by: Jan Krcal <jkrcal@chromium.org>
Cr-Commit-Position: refs/heads/master@{#541805}

chrome/browser/sync/test/integration/single_client_user_events_sync_test.cc

@@ -254,7 +254,13 @@ IN_PROC_BROWSER_TEST_F(SingleClientUserEventsSyncTest, NoHistory) {
 
   event_service->RecordUserEvent(testEvent1);
   event_service->RecordUserEvent(consent1);
+
+  // Wait until the first two events are committed before disabling sync,
+  // because disabled TYPED_URLS also disables user event sync, dropping all
+  // uncommitted consents.
+  EXPECT_TRUE(ExpectUserEvents({testEvent1, consent1}));
   ASSERT_TRUE(GetClient(0)->DisableSyncForDatatype(syncer::TYPED_URLS));
+
   event_service->RecordUserEvent(testEvent2);
   event_service->RecordUserEvent(consent2);
   ASSERT_TRUE(GetClient(0)->EnableSyncForDatatype(syncer::TYPED_URLS));

components/sync/BUILD.gn

@@ -720,6 +720,8 @@ static_library("test_support_model") {
     "model/fake_sync_change_processor.h",
     "model/fake_syncable_service.cc",
     "model/fake_syncable_service.h",
+    "model/mock_model_type_change_processor.cc",
+    "model/mock_model_type_change_processor.h",
     "model/model_type_store_test_util.cc",
     "model/model_type_store_test_util.h",
     "model/recording_model_type_change_processor.cc",

components/sync/model/mock_model_type_change_processor.cc

+// Copyright 2018 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "components/sync/model/mock_model_type_change_processor.h"
+
+#include <utility>
+
+#include "base/bind.h"
+#include "base/memory/ptr_util.h"
+#include "components/sync/model/metadata_batch.h"
+
+namespace syncer {
+namespace {
+
+class ForwardingModelTypeChangeProcessor : public ModelTypeChangeProcessor {
+ public:
+  // |other| must not be nullptr and must outlive this object.
+  explicit ForwardingModelTypeChangeProcessor(ModelTypeChangeProcessor* other)
+      : other_(other) {}
+  ~ForwardingModelTypeChangeProcessor() override{};
+
+  void Put(const std::string& client_tag,
+           std::unique_ptr<EntityData> entity_data,
+           MetadataChangeList* metadata_change_list) override {
+    other_->Put(client_tag, std::move(entity_data), metadata_change_list);
+  }
+
+  void Delete(const std::string& client_tag,
+              MetadataChangeList* metadata_change_list) override {
+    other_->Delete(client_tag, metadata_change_list);
+  }
+
+  void UpdateStorageKey(const EntityData& entity_data,
+                        const std::string& storage_key,
+                        MetadataChangeList* metadata_change_list) override {
+    other_->UpdateStorageKey(entity_data, storage_key, metadata_change_list);
+  }
+
+  void UntrackEntity(const EntityData& entity_data) override {
+    other_->UntrackEntity(entity_data);
+  }
+
+  void ModelReadyToSync(std::unique_ptr<MetadataBatch> batch) override {
+    other_->ModelReadyToSync(std::move(batch));
+  }
+
+  void OnSyncStarting(const ModelErrorHandler& error_handler,
+                      const StartCallback& callback) override {
+    other_->OnSyncStarting(error_handler, callback);
+  }
+
+  void DisableSync() override { other_->DisableSync(); }
+
+  bool IsTrackingMetadata() override { return other_->IsTrackingMetadata(); }
+
+  void ReportError(const ModelError& error) override {
+    other_->ReportError(error);
+  }
+
+ private:
+  ModelTypeChangeProcessor* other_;
+};
+
+}  // namespace
+
+MockModelTypeChangeProcessor::MockModelTypeChangeProcessor() {}
+
+MockModelTypeChangeProcessor::~MockModelTypeChangeProcessor() {}
+
+void MockModelTypeChangeProcessor::Put(
+    const std::string& storage_key,
+    std::unique_ptr<EntityData> entity_data,
+    MetadataChangeList* metadata_change_list) {
+  DoPut(storage_key, entity_data.get(), metadata_change_list);
+}
+
+void MockModelTypeChangeProcessor::ModelReadyToSync(
+    std::unique_ptr<MetadataBatch> batch) {
+  DoModelReadyToSync(batch.get());
+}
+
+ModelTypeSyncBridge::ChangeProcessorFactory
+MockModelTypeChangeProcessor::FactoryForBridgeTest() {
+  return base::BindRepeating(
+      [](ModelTypeChangeProcessor* processor, ModelType, ModelTypeSyncBridge*) {
+        return base::WrapUnique<ModelTypeChangeProcessor>(
+            new ForwardingModelTypeChangeProcessor(processor));
+      },
+      base::Unretained(this));
+}
+
+}  //  namespace syncer

components/sync/model/mock_model_type_change_processor.h

+// Copyright 2018 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef COMPONENTS_SYNC_MODEL_MOCK_MODEL_TYPE_CHANGE_PROCESSOR_H_
+#define COMPONENTS_SYNC_MODEL_MOCK_MODEL_TYPE_CHANGE_PROCESSOR_H_
+
+#include <memory>
+#include <string>
+
+#include "components/sync/model/model_type_change_processor.h"
+#include "components/sync/model/model_type_sync_bridge.h"
+#include "testing/gmock/include/gmock/gmock.h"
+
+namespace syncer {
+
+class MockModelTypeChangeProcessor : public ModelTypeChangeProcessor {
+ public:
+  MockModelTypeChangeProcessor();
+  virtual ~MockModelTypeChangeProcessor();
+
+  // TODO(crbug.com/729950): Use unique_ptr here direclty once move-only
+  // arguments are supported in gMock.
+  MOCK_METHOD3(DoPut,
+               void(const std::string& storage_key,
+                    EntityData* entity_data,
+                    MetadataChangeList* metadata_change_list));
+  void Put(const std::string& storage_key,
+           std::unique_ptr<EntityData> entity_data,
+           MetadataChangeList* metadata_change_list) override;
+  MOCK_METHOD2(Delete,
+               void(const std::string& storage_key,
+                    MetadataChangeList* metadata_change_list));
+  MOCK_METHOD3(UpdateStorageKey,
+               void(const EntityData& entity_data,
+                    const std::string& storage_key,
+                    MetadataChangeList* metadata_change_list));
+  MOCK_METHOD1(UntrackEntity, void(const EntityData& entity_data));
+  // TODO(crbug.com/729950): Use unique_ptr here direclty once move-only
+  // arguments are supported in gMock.
+  MOCK_METHOD1(DoModelReadyToSync, void(MetadataBatch* batch));
+  void ModelReadyToSync(std::unique_ptr<MetadataBatch> batch) override;
+  MOCK_METHOD2(OnSyncStarting,
+               void(const ModelErrorHandler& error_handler,
+                    const StartCallback& callback));
+  MOCK_METHOD0(DisableSync, void());
+  MOCK_METHOD0(IsTrackingMetadata, bool());
+  MOCK_METHOD1(ReportError, void(const ModelError& error));
+
+  // Returns a callback that constructs a processor that forwards all calls to
+  // |this|. |*this| must outlive the returned factory.
+  ModelTypeSyncBridge::ChangeProcessorFactory FactoryForBridgeTest();
+};
+
+}  //  namespace syncer
+
+#endif  // COMPONENTS_SYNC_MODEL_MOCK_MODEL_TYPE_CHANGE_PROCESSOR_H_

components/sync/model/model_type_sync_bridge_unittest.cc

@@ -7,8 +7,9 @@
 #include <utility>
 
 #include "base/bind.h"
-#include "components/sync/model/fake_model_type_change_processor.h"
+#include "base/bind_helpers.h"
 #include "components/sync/model/metadata_batch.h"
+#include "components/sync/model/mock_model_type_change_processor.h"
 #include "components/sync/model/stub_model_type_sync_bridge.h"
 #include "components/sync/protocol/model_type_state.pb.h"
 #include "testing/gmock/include/gmock/gmock.h"
@@ -17,149 +18,56 @@
 namespace syncer {
 namespace {
 
-// A mock MTCP that lets verify DisableSync and ModelReadyToSync were called in
-// the ways that we expect.
-class MockModelTypeChangeProcessor : public FakeModelTypeChangeProcessor {
- public:
-  explicit MockModelTypeChangeProcessor(const base::Closure& disabled_callback)
-      : disabled_callback_(disabled_callback) {}
-  ~MockModelTypeChangeProcessor() override {}
-
-  void DisableSync() override { disabled_callback_.Run(); }
-
-  bool IsTrackingMetadata() override { return metadata_batch_ != nullptr; }
-
-  void ModelReadyToSync(std::unique_ptr<MetadataBatch> batch) override {
-    EXPECT_NE(nullptr, batch);
-    metadata_batch_ = std::move(batch);
-  }
-
-  MetadataBatch* metadata_batch() { return metadata_batch_.get(); }
-
- private:
-  // This callback is invoked when DisableSync() is called, instead of
-  // remembering that this event happened in our own state. The reason for this
-  // is that after DisableSync() is called on us, the bridge is going to
-  // destroy this processor instance, and any state would be lost. The callback
-  // allows this information to reach somewhere safe instead.
-  base::Closure disabled_callback_;
-
-  std::unique_ptr<MetadataBatch> metadata_batch_;
-};
-
-class MockModelTypeSyncBridge : public StubModelTypeSyncBridge {
- public:
-  MockModelTypeSyncBridge()
-      : StubModelTypeSyncBridge(
-            base::Bind(&MockModelTypeSyncBridge::CreateProcessor,
-                       base::Unretained(this))) {}
-  ~MockModelTypeSyncBridge() override {}
-
-  MockModelTypeChangeProcessor* change_processor() const {
-    return static_cast<MockModelTypeChangeProcessor*>(
-        ModelTypeSyncBridge::change_processor());
-  }
-
-  bool processor_disable_sync_called() const {
-    return processor_disable_sync_called_;
-  }
-
- private:
-  std::unique_ptr<ModelTypeChangeProcessor> CreateProcessor(
-      ModelType type,
-      ModelTypeSyncBridge* bridge) {
-    return std::make_unique<MockModelTypeChangeProcessor>(
-        base::Bind(&MockModelTypeSyncBridge::OnProcessorDisableSync,
-                   base::Unretained(this)));
-  }
+using testing::Return;
+using testing::_;
 
-  void OnProcessorDisableSync() { processor_disable_sync_called_ = true; }
-
-  bool processor_disable_sync_called_ = false;
-};
+MATCHER(IsEmptyMetadataBatch, "") {
+  return arg != nullptr &&
+         sync_pb::ModelTypeState().SerializeAsString() ==
+             arg->GetModelTypeState().SerializeAsString() &&
+         arg->TakeAllMetadata().empty();
+}
 
 class ModelTypeSyncBridgeTest : public ::testing::Test {
  public:
-  ModelTypeSyncBridgeTest() {}
+  ModelTypeSyncBridgeTest() : bridge_(mock_processor_.FactoryForBridgeTest()) {}
   ~ModelTypeSyncBridgeTest() override {}
 
   void OnSyncStarting() {
-    bridge_.OnSyncStarting(
-        ModelErrorHandler(),
-        base::Bind(&ModelTypeSyncBridgeTest::OnProcessorStarted,
-                   base::Unretained(this)));
+    bridge_.OnSyncStarting(ModelErrorHandler(), base::DoNothing());
   }
 
-  bool start_callback_called() const { return start_callback_called_; }
-  MockModelTypeSyncBridge* bridge() { return &bridge_; }
+  StubModelTypeSyncBridge* bridge() { return &bridge_; }
+  MockModelTypeChangeProcessor* processor() { return &mock_processor_; }
 
  private:
-  void OnProcessorStarted(
-      std::unique_ptr<ActivationContext> activation_context) {
-    start_callback_called_ = true;
-  }
-
-  bool start_callback_called_ = false;
-  MockModelTypeSyncBridge bridge_;
+  testing::NiceMock<MockModelTypeChangeProcessor> mock_processor_;
+  StubModelTypeSyncBridge bridge_;
 };
 
 // OnSyncStarting should create a processor and call OnSyncStarting on it.
 TEST_F(ModelTypeSyncBridgeTest, OnSyncStarting) {
-  EXPECT_FALSE(start_callback_called());
+  EXPECT_CALL(*processor(), OnSyncStarting(_, _));
   OnSyncStarting();
-
-  // FakeModelTypeProcessor is the one that calls the callback, so if it was
-  // called then we know the call on the processor was made.
-  EXPECT_TRUE(start_callback_called());
 }
 
-// DisableSync should call DisableSync on the processor and then delete it.
+// DisableSync should call DisableSync on the processor.
 TEST_F(ModelTypeSyncBridgeTest, DisableSync) {
-  ASSERT_FALSE(bridge()->processor_disable_sync_called());
+  EXPECT_CALL(*processor(), DisableSync());
   bridge()->DisableSync();
-
-  // Disabling also wipes out metadata, and the bridge should have told the new
-  // processor about this.
-  EXPECT_TRUE(bridge()->processor_disable_sync_called());
-  EXPECT_EQ(nullptr, bridge()->change_processor()->metadata_batch());
 }
 
 // DisableSync should propagate the model readiness (IsTrackingMetadata()).
 TEST_F(ModelTypeSyncBridgeTest, PropagateModelReadyToSyncInDisableSync) {
-  ASSERT_EQ(nullptr, bridge()->change_processor()->metadata_batch());
-  ASSERT_FALSE(bridge()->change_processor()->IsTrackingMetadata());
-
   // Model is not ready to sync, so it should remain so after DisableSync().
+  ON_CALL(*processor(), IsTrackingMetadata()).WillByDefault(Return(false));
+  EXPECT_CALL(*processor(), DoModelReadyToSync(_)).Times(0);
   bridge()->DisableSync();
-  EXPECT_EQ(nullptr, bridge()->change_processor()->metadata_batch());
-  EXPECT_FALSE(bridge()->change_processor()->IsTrackingMetadata());
-
-  // Mimic the model being ready to sync.
-  sync_pb::ModelTypeState initial_state;
-  initial_state.set_initial_sync_done(true);
-  auto initial_batch = std::make_unique<MetadataBatch>();
-  initial_batch->SetModelTypeState(initial_state);
-  bridge()->change_processor()->ModelReadyToSync(std::move(initial_batch));
-
-  ASSERT_TRUE(bridge()->change_processor()->IsTrackingMetadata());
-  ASSERT_NE(nullptr, bridge()->change_processor()->metadata_batch());
-  ASSERT_TRUE(bridge()
-                  ->change_processor()
-                  ->metadata_batch()
-                  ->GetModelTypeState()
-                  .initial_sync_done());
-
-  // Model is ready to sync, so it should remain so after DisableSync().
-  // However, the metadata should have been cleared.
+
+  // If the Model is ready to sync, so it should remain so after DisableSync().
+  ON_CALL(*processor(), IsTrackingMetadata()).WillByDefault(Return(true));
+  EXPECT_CALL(*processor(), DoModelReadyToSync(IsEmptyMetadataBatch()));
   bridge()->DisableSync();
-  EXPECT_TRUE(bridge()->change_processor()->IsTrackingMetadata());
-  MetadataBatch* batch = bridge()->change_processor()->metadata_batch();
-  ASSERT_NE(nullptr, batch);
-
-  EXPECT_FALSE(batch->GetModelTypeState().initial_sync_done());
-  EXPECT_EQ(sync_pb::ModelTypeState().SerializeAsString(),
-            batch->GetModelTypeState().SerializeAsString());
-  EXPECT_EQ(0U, batch->TakeAllMetadata().size());
 }
 
 // ResolveConflicts should return USE_REMOTE unless the remote data is deleted.

components/sync/model_impl/client_tag_based_model_type_processor.cc

@@ -117,6 +117,9 @@ void ClientTagBasedModelTypeProcessor::ModelReadyToSync(
     // First time syncing; initialize metadata.
     model_type_state_.mutable_progress_marker()->set_data_type_id(
         GetSpecificsFieldNumberFromModelType(type_));
+    // For commit-only types, no updates are expected and hence we can consider
+    // initial_sync_done().
+    model_type_state_.set_initial_sync_done(commit_only_);
   }
 
   ConnectIfReady();

components/sync/model_impl/client_tag_based_model_type_processor_unittest.cc

@@ -713,6 +713,7 @@ TEST_F(ClientTagBasedModelTypeProcessorTest, LocalCreateItem) {
 TEST_F(ClientTagBasedModelTypeProcessorTest, CommitOnlySimple) {
   ResetState(false, true);
   InitializeToReadyState();
+  EXPECT_TRUE(db().model_type_state().initial_sync_done());
 
   bridge()->WriteItem(kKey1, kValue1);
   worker()->VerifyPendingCommits({kHash1});

components/sync/user_events/user_event_sync_bridge.cc

@@ -145,6 +145,7 @@ std::string UserEventSyncBridge::GetStorageKey(const EntityData& entity_data) {
 }
 
 void UserEventSyncBridge::DisableSync() {
+  ModelTypeSyncBridge::DisableSync();
   // No data should be retained through sign out.
   store_->ReadAllData(base::BindOnce(
       &UserEventSyncBridge::OnReadAllDataToDelete, base::AsWeakPtr(this)));
@@ -213,15 +214,6 @@ void UserEventSyncBridge::OnReadAllMetadata(
   if (error) {
     change_processor()->ReportError(*error);
   } else {
-    if (!metadata_batch->GetModelTypeState().initial_sync_done()) {
-      // We have never initialized before, force it to true. We are not going to
-      // ever have a GetUpdates because our type is commit only.
-      // TODO(skym): Do we need to worry about saving this back ourselves? Or
-      // does that get taken care for us?
-      ModelTypeState state = metadata_batch->GetModelTypeState();
-      state.set_initial_sync_done(true);
-      metadata_batch->SetModelTypeState(state);
-    }
     change_processor()->ModelReadyToSync(std::move(metadata_batch));
   }
 }

components/sync/user_events/user_event_sync_bridge_unittest.cc

@@ -13,50 +13,45 @@
 #include "base/run_loop.h"
 #include "components/sync/driver/fake_sync_service.h"
 #include "components/sync/model/data_batch.h"
+#include "components/sync/model/mock_model_type_change_processor.h"
 #include "components/sync/model/model_type_store_test_util.h"
-#include "components/sync/model/recording_model_type_change_processor.h"
 #include "components/sync/protocol/sync.pb.h"
+#include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
-using sync_pb::UserEventSpecifics;
-
 namespace syncer {
-
 namespace {
 
-void VerifyEqual(const UserEventSpecifics& s1, const UserEventSpecifics& s2) {
-  EXPECT_EQ(s1.event_time_usec(), s2.event_time_usec());
-  EXPECT_EQ(s1.navigation_id(), s2.navigation_id());
-  EXPECT_EQ(s1.session_id(), s2.session_id());
-}
-
-void VerifyDataBatchCount(int expected_count,
-                          std::unique_ptr<DataBatch> batch) {
-  int actual_count = 0;
-  while (batch->HasNext()) {
-    ++actual_count;
-    batch->Next();
+using sync_pb::UserEventSpecifics;
+using testing::ElementsAre;
+using testing::Invoke;
+using testing::IsEmpty;
+using testing::IsNull;
+using testing::NotNull;
+using testing::Pair;
+using testing::Pointee;
+using testing::Return;
+using testing::SaveArg;
+using testing::SizeIs;
+using testing::UnorderedElementsAre;
+using testing::_;
+
+MATCHER_P(MatchesUserEvent, expected, "") {
+  if (!arg.has_user_event()) {
+    *result_listener << "which is not a user event";
+    return false;
   }
-  EXPECT_EQ(expected_count, actual_count);
-}
-
-void VerifyDataBatch(std::map<std::string, UserEventSpecifics> expected,
-                     std::unique_ptr<DataBatch> batch) {
-  while (batch->HasNext()) {
-    const KeyAndData& pair = batch->Next();
-    auto iter = expected.find(pair.first);
-    ASSERT_NE(iter, expected.end());
-    VerifyEqual(iter->second, pair.second->specifics.user_event());
-    // Removing allows us to verify we don't see the same item multiple times,
-    // and that we saw everything we expected.
-    expected.erase(iter);
+  const UserEventSpecifics& actual = arg.user_event();
+  if (actual.event_time_usec() != expected.event_time_usec()) {
+    return false;
   }
-  EXPECT_TRUE(expected.empty());
-}
-
-base::Callback<void(std::unique_ptr<DataBatch> batch)> VerifyCallback(
-    std::map<std::string, UserEventSpecifics> expected) {
-  return base::Bind(&VerifyDataBatch, expected);
+  if (actual.navigation_id() != expected.navigation_id()) {
+    return false;
+  }
+  if (actual.session_id() != expected.session_id()) {
+    return false;
+  }
+  return true;
 }
 
 UserEventSpecifics CreateSpecifics(int64_t event_time_usec,
@@ -102,14 +97,8 @@ class UserEventSyncBridgeTest : public testing::Test {
   UserEventSyncBridgeTest() {
     bridge_ = std::make_unique<UserEventSyncBridge>(
         ModelTypeStoreTestUtil::FactoryForInMemoryStoreForTest(),
-        RecordingModelTypeChangeProcessor::FactoryForBridgeTest(&processor_),
-        &test_global_id_mapper_);
-  }
-
-  ~UserEventSyncBridgeTest() override {
-    // Get[All]Data() calls are async, so this will run the verification they
-    // call in their callbacks.
-    base::RunLoop().RunUntilIdle();
+        mock_processor_.FactoryForBridgeTest(), &test_global_id_mapper_);
+    ON_CALL(*processor(), IsTrackingMetadata()).WillByDefault(Return(true));
   }
 
   std::string GetStorageKey(const UserEventSpecifics& specifics) {
@@ -119,67 +108,135 @@ class UserEventSyncBridgeTest : public testing::Test {
   }
 
   UserEventSyncBridge* bridge() { return bridge_.get(); }
-  RecordingModelTypeChangeProcessor* processor() { return processor_; }
+  MockModelTypeChangeProcessor* processor() { return &mock_processor_; }
   TestGlobalIdMapper* mapper() { return &test_global_id_mapper_; }
 
+  std::map<std::string, sync_pb::EntitySpecifics> GetAllData() {
+    base::RunLoop loop;
+    std::unique_ptr<DataBatch> batch;
+    bridge_->GetAllData(base::BindOnce(
+        [](base::RunLoop* loop, std::unique_ptr<DataBatch>* out_batch,
+           std::unique_ptr<DataBatch> batch) {
+          *out_batch = std::move(batch);
+          loop->Quit();
+        },
+        &loop, &batch));
+    loop.Run();
+    EXPECT_NE(nullptr, batch);
+
+    std::map<std::string, sync_pb::EntitySpecifics> storage_key_to_specifics;
+    if (batch != nullptr) {
+      while (batch->HasNext()) {
+        const syncer::KeyAndData& pair = batch->Next();
+        storage_key_to_specifics[pair.first] = pair.second->specifics;
+      }
+    }
+    return storage_key_to_specifics;
+  }
+
+  std::unique_ptr<sync_pb::EntitySpecifics> GetData(
+      const std::string& storage_key) {
+    base::RunLoop loop;
+    std::unique_ptr<DataBatch> batch;
+    bridge_->GetData(
+        {storage_key},
+        base::BindOnce(
+            [](base::RunLoop* loop, std::unique_ptr<DataBatch>* out_batch,
+               std::unique_ptr<DataBatch> batch) {
+              *out_batch = std::move(batch);
+              loop->Quit();
+            },
+            &loop, &batch));
+    loop.Run();
+    EXPECT_NE(nullptr, batch);
+
+    std::unique_ptr<sync_pb::EntitySpecifics> specifics;
+    if (batch != nullptr && batch->HasNext()) {
+      const syncer::KeyAndData& pair = batch->Next();
+      specifics =
+          std::make_unique<sync_pb::EntitySpecifics>(pair.second->specifics);
+      EXPECT_FALSE(batch->HasNext());
+    }
+    return specifics;
+  }
+
  private:
   std::unique_ptr<UserEventSyncBridge> bridge_;
-  RecordingModelTypeChangeProcessor* processor_;
+  testing::NiceMock<MockModelTypeChangeProcessor> mock_processor_;
   TestGlobalIdMapper test_global_id_mapper_;
   base::MessageLoop message_loop_;
 };
 
 TEST_F(UserEventSyncBridgeTest, MetadataIsInitialized) {
+  EXPECT_CALL(*processor(), DoModelReadyToSync(NotNull()));
   base::RunLoop().RunUntilIdle();
-  EXPECT_TRUE(processor()->metadata()->GetModelTypeState().initial_sync_done());
 }
 
 TEST_F(UserEventSyncBridgeTest, SingleRecord) {
   const UserEventSpecifics specifics(CreateSpecifics(1u, 2u, 3u));
+  std::string storage_key;
+  EXPECT_CALL(*processor(), DoPut(_, _, _)).WillOnce(SaveArg<0>(&storage_key));
   bridge()->RecordUserEvent(std::make_unique<UserEventSpecifics>(specifics));
-  EXPECT_EQ(1u, processor()->put_multimap().size());
 
-  const std::string storage_key = processor()->put_multimap().begin()->first;
-  bridge()->GetData({storage_key}, VerifyCallback({{storage_key, specifics}}));
-  bridge()->GetData({"bogus"}, base::Bind(&VerifyDataBatchCount, 0));
-  bridge()->GetAllData(VerifyCallback({{storage_key, specifics}}));
+  EXPECT_THAT(GetData(storage_key), Pointee(MatchesUserEvent(specifics)));
+  EXPECT_THAT(GetData("bogus"), IsNull());
+  EXPECT_THAT(GetAllData(),
+              ElementsAre(Pair(storage_key, MatchesUserEvent(specifics))));
+}
+
+TEST_F(UserEventSyncBridgeTest, DisableSync) {
+  const UserEventSpecifics specifics(CreateSpecifics(1u, 2u, 3u));
+  bridge()->RecordUserEvent(std::make_unique<UserEventSpecifics>(specifics));
+  ASSERT_THAT(GetAllData(), SizeIs(1));
 
+  EXPECT_CALL(*processor(), DisableSync());
   bridge()->DisableSync();
 
   // Disabling deletes records through multiple round trips, if we quickly call
   // GetAllData() we're going to beat the deletions to the storage task.
   base::RunLoop().RunUntilIdle();
 
-  bridge()->GetAllData(base::Bind(&VerifyDataBatchCount, 0));
-  EXPECT_EQ(0u, processor()->delete_set().size());
+  EXPECT_THAT(GetAllData(), IsEmpty());
 }
 
 TEST_F(UserEventSyncBridgeTest, MultipleRecords) {
+  std::set<std::string> unique_storage_keys;
+  EXPECT_CALL(*processor(), DoPut(_, _, _))
+      .Times(4)
+      .WillRepeatedly(
+          Invoke([&unique_storage_keys](
+                     const std::string& storage_key, EntityData* entity_data,
+                     MetadataChangeList* metadata_change_list) {
+            unique_storage_keys.insert(storage_key);
+          }));
+
   bridge()->RecordUserEvent(SpecificsUniquePtr(1u, 1u, 1u));
   bridge()->RecordUserEvent(SpecificsUniquePtr(1u, 1u, 2u));
   bridge()->RecordUserEvent(SpecificsUniquePtr(1u, 2u, 2u));
   bridge()->RecordUserEvent(SpecificsUniquePtr(2u, 2u, 2u));
 
-  EXPECT_EQ(4u, processor()->put_multimap().size());
-  std::set<std::string> unique_storage_keys;
-  for (const auto& kv : processor()->put_multimap()) {
-    unique_storage_keys.insert(kv.first);
-  }
   EXPECT_EQ(2u, unique_storage_keys.size());
-  bridge()->GetAllData(base::Bind(&VerifyDataBatchCount, 2));
+  EXPECT_THAT(GetAllData(), SizeIs(2));
 }
 
 TEST_F(UserEventSyncBridgeTest, ApplySyncChanges) {
+  std::string storage_key1;
+  std::string storage_key2;
+  EXPECT_CALL(*processor(), DoPut(_, _, _))
+      .WillOnce(SaveArg<0>(&storage_key1))
+      .WillOnce(SaveArg<0>(&storage_key2));
+
   bridge()->RecordUserEvent(SpecificsUniquePtr(1u, 1u, 1u));
   bridge()->RecordUserEvent(SpecificsUniquePtr(2u, 2u, 2u));
-  bridge()->GetAllData(base::Bind(&VerifyDataBatchCount, 2));
+  EXPECT_THAT(GetAllData(), SizeIs(2));
 
-  const std::string storage_key = processor()->put_multimap().begin()->first;
   auto error_on_delete =
       bridge()->ApplySyncChanges(bridge()->CreateMetadataChangeList(),
-                                 {EntityChange::CreateDelete(storage_key)});
+                                 {EntityChange::CreateDelete(storage_key1)});
   EXPECT_FALSE(error_on_delete);
-  bridge()->GetAllData(base::Bind(&VerifyDataBatchCount, 1));
+  EXPECT_THAT(GetAllData(), SizeIs(1));
+  EXPECT_THAT(GetData(storage_key1), IsNull());
+  EXPECT_THAT(GetData(storage_key2), NotNull());
 }
 
 TEST_F(UserEventSyncBridgeTest, HandleGlobalIdChange) {
@@ -188,31 +245,34 @@ TEST_F(UserEventSyncBridgeTest, HandleGlobalIdChange) {
   int64_t third_id = 13;
   int64_t fourth_id = 14;
 
+  std::string storage_key;
+  EXPECT_CALL(*processor(), DoPut(_, _, _)).WillOnce(SaveArg<0>(&storage_key));
+
   // This id update should be applied to the event as it is initially recorded.
   mapper()->ChangeId(first_id, second_id);
   bridge()->RecordUserEvent(SpecificsUniquePtr(1u, first_id, 2u));
-  const std::string storage_key = processor()->put_multimap().begin()->first;
-  EXPECT_EQ(1u, processor()->put_multimap().size());
-  bridge()->GetAllData(
-      VerifyCallback({{storage_key, CreateSpecifics(1u, second_id, 2u)}}));
+  EXPECT_THAT(GetAllData(),
+              ElementsAre(Pair(storage_key, MatchesUserEvent(CreateSpecifics(
+                                                1u, second_id, 2u)))));
 
   // This id update is done while the event is "in flight", and should result in
   // it being updated and re-sent to sync.
+  EXPECT_CALL(*processor(), DoPut(storage_key, _, _));
   mapper()->ChangeId(second_id, third_id);
-  EXPECT_EQ(2u, processor()->put_multimap().size());
-  bridge()->GetAllData(
-      VerifyCallback({{storage_key, CreateSpecifics(1u, third_id, 2u)}}));
+  EXPECT_THAT(GetAllData(),
+              ElementsAre(Pair(storage_key, MatchesUserEvent(CreateSpecifics(
+                                                1u, third_id, 2u)))));
   auto error_on_delete =
       bridge()->ApplySyncChanges(bridge()->CreateMetadataChangeList(),
                                  {EntityChange::CreateDelete(storage_key)});
   EXPECT_FALSE(error_on_delete);
-  bridge()->GetAllData(base::Bind(&VerifyDataBatchCount, 0));
+  EXPECT_THAT(GetAllData(), IsEmpty());
 
   // This id update should be ignored, since we received commit confirmation
   // above.
+  EXPECT_CALL(*processor(), DoPut(_, _, _)).Times(0);
   mapper()->ChangeId(third_id, fourth_id);
-  EXPECT_EQ(2u, processor()->put_multimap().size());
-  bridge()->GetAllData(base::Bind(&VerifyDataBatchCount, 0));
+  EXPECT_THAT(GetAllData(), IsEmpty());
 }
 
 TEST_F(UserEventSyncBridgeTest, MulipleEventsChanging) {
@@ -220,37 +280,46 @@ TEST_F(UserEventSyncBridgeTest, MulipleEventsChanging) {
   int64_t second_id = 12;
   int64_t third_id = 13;
   int64_t fourth_id = 14;
-  const UserEventSpecifics specifics1 = CreateSpecifics(1u, first_id, 2u);
-  const UserEventSpecifics specifics2 = CreateSpecifics(1u, first_id, 2u);
-  const UserEventSpecifics specifics3 = CreateSpecifics(1u, first_id, 2u);
+  const UserEventSpecifics specifics1 = CreateSpecifics(101u, first_id, 2u);
+  const UserEventSpecifics specifics2 = CreateSpecifics(102u, second_id, 4u);
+  const UserEventSpecifics specifics3 = CreateSpecifics(103u, third_id, 6u);
   const std::string key1 = GetStorageKey(specifics1);
   const std::string key2 = GetStorageKey(specifics2);
   const std::string key3 = GetStorageKey(specifics3);
+  ASSERT_NE(key1, key2);
+  ASSERT_NE(key1, key3);
+  ASSERT_NE(key2, key3);
 
   bridge()->RecordUserEvent(std::make_unique<UserEventSpecifics>(specifics1));
   bridge()->RecordUserEvent(std::make_unique<UserEventSpecifics>(specifics2));
   bridge()->RecordUserEvent(std::make_unique<UserEventSpecifics>(specifics3));
-  bridge()->GetAllData(VerifyCallback(
-      {{key1, specifics1}, {key2, specifics2}, {key3, specifics3}}));
+  ASSERT_THAT(GetAllData(),
+              UnorderedElementsAre(Pair(key1, MatchesUserEvent(specifics1)),
+                                   Pair(key2, MatchesUserEvent(specifics2)),
+                                   Pair(key3, MatchesUserEvent(specifics3))));
 
   mapper()->ChangeId(second_id, fourth_id);
-  bridge()->GetAllData(
-      VerifyCallback({{key1, specifics1},
-                      {key2, CreateSpecifics(3u, fourth_id, 4u)},
-                      {key3, specifics3}}));
+  EXPECT_THAT(
+      GetAllData(),
+      UnorderedElementsAre(
+          Pair(key1, MatchesUserEvent(specifics1)),
+          Pair(key2, MatchesUserEvent(CreateSpecifics(102u, fourth_id, 4u))),
+          Pair(key3, MatchesUserEvent(specifics3))));
 
   mapper()->ChangeId(first_id, fourth_id);
   mapper()->ChangeId(third_id, fourth_id);
-  bridge()->GetAllData(
-      VerifyCallback({{key1, CreateSpecifics(1u, fourth_id, 2u)},
-                      {key2, CreateSpecifics(3u, fourth_id, 4u)},
-                      {key3, CreateSpecifics(5u, fourth_id, 6u)}}));
+  EXPECT_THAT(
+      GetAllData(),
+      UnorderedElementsAre(
+          Pair(key1, MatchesUserEvent(CreateSpecifics(101u, fourth_id, 2u))),
+          Pair(key2, MatchesUserEvent(CreateSpecifics(102u, fourth_id, 4u))),
+          Pair(key3, MatchesUserEvent(CreateSpecifics(103u, fourth_id, 6u)))));
 }
 
 TEST_F(UserEventSyncBridgeTest, RecordBeforeMetadataLoads) {
-  processor()->SetIsTrackingMetadata(false);
+  ON_CALL(*processor(), IsTrackingMetadata()).WillByDefault(Return(false));
   bridge()->RecordUserEvent(SpecificsUniquePtr(1u, 2u, 3u));
-  bridge()->GetAllData(VerifyCallback({}));
+  EXPECT_THAT(GetAllData(), IsEmpty());
 }
 
 }  // namespace