Implement sync in the NonBlockingTypeProcessor

sync/engine/non_blocking_type_processor_unittest.cc

Index: sync/engine/non_blocking_type_processor_unittest.cc
diff --git a/sync/engine/non_blocking_type_processor_unittest.cc b/sync/engine/non_blocking_type_processor_unittest.cc
new file mode 100644
index 0000000000000000000000000000000000000000..d0675beaec1018c085881ec8862badeaac015a7d
--- /dev/null
+++ b/sync/engine/non_blocking_type_processor_unittest.cc
@@ -0,0 +1,517 @@
+// Copyright 2014 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 "sync/engine/non_blocking_type_processor.h"
+
+#include "sync/engine/non_blocking_sync_common.h"
+#include "sync/engine/non_blocking_type_processor_core_interface.h"
+#include "sync/internal_api/public/base/model_type.h"
+#include "sync/internal_api/public/sync_core_proxy.h"
+#include "sync/protocol/sync.pb.h"
+#include "sync/syncable/syncable_util.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace syncer {
+
+namespace {
+
+class MockNonBlockingTypeProcessorCore
+    : public NonBlockingTypeProcessorCoreInterface {
+ public:
+  MockNonBlockingTypeProcessorCore();
+  virtual ~MockNonBlockingTypeProcessorCore();
+
+  virtual void RequestCommits(const CommitRequestDataList& list) OVERRIDE;
+
+  bool is_connected_;
+
+  std::vector<CommitRequestDataList> commit_request_lists_;
+};
+
+MockNonBlockingTypeProcessorCore::MockNonBlockingTypeProcessorCore()
+    : is_connected_(false) {
+}
+
+MockNonBlockingTypeProcessorCore::~MockNonBlockingTypeProcessorCore() {
+}
+
+void MockNonBlockingTypeProcessorCore::RequestCommits(
+    const CommitRequestDataList& list) {
+  commit_request_lists_.push_back(list);
+}
+
+class MockSyncCoreProxy : public syncer::SyncCoreProxy {
+ public:
+  MockSyncCoreProxy();
+  virtual ~MockSyncCoreProxy();
+
+  virtual void ConnectTypeToCore(
+      syncer::ModelType type,
+      const DataTypeState& data_type_state,
+      base::WeakPtr<syncer::NonBlockingTypeProcessor> type_processor) OVERRIDE;
+  virtual void Disconnect(syncer::ModelType type) OVERRIDE;
+  virtual scoped_ptr<SyncCoreProxy> Clone() const OVERRIDE;
+
+  MockNonBlockingTypeProcessorCore* GetMockProcessorCore();
+
+ private:
+  explicit MockSyncCoreProxy(MockNonBlockingTypeProcessorCore*);
+
+  // The NonBlockingTypeProcessor's contract expects that it gets to own this
+  // object, so we can retain only a non-owned pointer to it.
+  //
+  // This is very unsafe, but we can get away with it since these tests are not
+  // exercising the processor <-> processor_core connection code.
+  MockNonBlockingTypeProcessorCore* mock_core_;
+};
+
+MockSyncCoreProxy::MockSyncCoreProxy()
+    : mock_core_(new MockNonBlockingTypeProcessorCore) {
+}
+
+MockSyncCoreProxy::MockSyncCoreProxy(MockNonBlockingTypeProcessorCore* core)
+    : mock_core_(core) {
+}
+
+MockSyncCoreProxy::~MockSyncCoreProxy() {
+}
+
+void MockSyncCoreProxy::ConnectTypeToCore(
+    syncer::ModelType type,
+    const DataTypeState& data_type_state,
+    base::WeakPtr<syncer::NonBlockingTypeProcessor> type_processor) {
+  // This class is allowed to participate in only one connection.
+  DCHECK(!mock_core_->is_connected_);
+  mock_core_->is_connected_ = true;
+
+  // Hands off ownership of our member to the type_processor, while keeping
+  // an unsafe pointer to it.  This is why we can only connect once.
+  scoped_ptr<NonBlockingTypeProcessorCoreInterface> core(mock_core_);
+
+  type_processor->OnConnect(core.Pass());
+}
+
+void MockSyncCoreProxy::Disconnect(syncer::ModelType type) {
+  // This mock object is not meant for connect and disconnect tests.
+  NOTREACHED() << "Not implemented";
+}
+
+scoped_ptr<SyncCoreProxy> MockSyncCoreProxy::Clone() const {
+  // There's no sensible way to clone this MockSyncCoreProxy.
+  return scoped_ptr<SyncCoreProxy>(new MockSyncCoreProxy(mock_core_));
+}
+
+MockNonBlockingTypeProcessorCore* MockSyncCoreProxy::GetMockProcessorCore() {
+  return mock_core_;
+}
+
+}  // namespace
+
+// Tests the sync engine parts of NonBlockingTypeProcessor.
+//
+// The NonBlockingTypeProcessor contains a non-trivial amount of code dedicated
+// to turning the sync engine on and off again.  That code is fairly well
+// tested in the NonBlockingDataTypeController unit tests and it doesn't need
+// to be re-tested here.
+//
+// These tests skip past initialization and focus on steady state sync engine
+// behvior.  This is where we test how the processor responds to the model's
+// requests to make changes to its data, the messages incoming fro the sync
+// server, and what happens when the two conflict.
+//
+// Inputs:
+// - Initial state from permanent storage. (TODO)
+// - Create, update or delete requests from the model.
+// - Update responses and commit responses from the server.
+//
+// Outputs:
+// - Writes to permanent storage. (TODO)
+// - Callbacks into the model. (TODO)
+// - Requests to the sync thread.  Tested with MockNonBlockingTypeProcessorCore.
+class NonBlockingTypeProcessorTest : public ::testing::Test {
+ public:
+  NonBlockingTypeProcessorTest();
+  virtual ~NonBlockingTypeProcessorTest();
+
+  // Explicit initialization step.  Kept separate to allow tests to inject
+  // on-disk state before the test begins.
+  void Initialize();
+
+  // Local data modification.  Emulates signals from the model thread.
+  void WriteItem(const std::string& tag, const std::string& value);
+  void DeleteItem(const std::string& tag);
+
+  // Emulate updates from the server.
+  // This harness has some functionality to help emulate server behavior.
+  // See the definitions of these methods for more information.
+  void UpdateFromServer(int64 version_offset,
+                        const std::string& tag,
+                        const std::string& value);
+  void TombstoneFromServer(int64 version_offset, const std::string& tag);
+
+  // Read emitted commit requests as batches.
+  size_t GetNumCommitRequestLists();
+  CommitRequestDataList GetNthCommitRequestList(size_t n);
+
+  // Read emitted commit requests by tag, most recent only.
+  bool HasCommitRequestForTag(const std::string& tag);
+  CommitRequestData GetLatestCommitRequestForTag(const std::string& tag);
+
+  // Sends the processor a successful commit response.
+  void SuccessfulCommitResponse(const CommitRequestData& request_data);
+
+ private:
+  std::string GenerateId(const std::string& tag) const;
+  std::string GenerateTagHash(const std::string& tag) const;
+  sync_pb::EntitySpecifics GenerateSpecifics(const std::string& tag,
+                                             const std::string& value) const;
+
+  int64 GetServerVersion(const std::string& tag);
+  void SetServerVersion(const std::string& tag, int64 version);
+
+  const ModelType type_;
+
+  scoped_ptr<MockSyncCoreProxy> mock_sync_core_proxy_;
+  scoped_ptr<NonBlockingTypeProcessor> processor_;
+  MockNonBlockingTypeProcessorCore* mock_processor_core_;
+
+  DataTypeState data_type_state_;
+
+  std::map<const std::string, int64> server_versions_;
+};
+
+NonBlockingTypeProcessorTest::NonBlockingTypeProcessorTest()
+    : type_(PREFERENCES),
+      mock_sync_core_proxy_(new MockSyncCoreProxy()),
+      processor_(new NonBlockingTypeProcessor(type_)) {
+}
+
+NonBlockingTypeProcessorTest::~NonBlockingTypeProcessorTest() {
+}
+
+void NonBlockingTypeProcessorTest::Initialize() {
+  processor_->Enable(mock_sync_core_proxy_->Clone());
+  mock_processor_core_ = mock_sync_core_proxy_->GetMockProcessorCore();
+}
+
+void NonBlockingTypeProcessorTest::WriteItem(const std::string& tag,
+                                             const std::string& value) {
+  processor_->Put(tag, GenerateSpecifics(tag, value));
+}
+
+void NonBlockingTypeProcessorTest::DeleteItem(const std::string& tag) {
+  processor_->Delete(tag);
+}
+
+void NonBlockingTypeProcessorTest::UpdateFromServer(int64 version_offset,
+                                                    const std::string& tag,
+                                                    const std::string& value) {
+  // Overwrite the existing server version if this is the new highest version.
+  int64 old_version = GetServerVersion(tag);
+  int64 version = old_version + version_offset;
+  if (version > old_version) {
+    SetServerVersion(tag, version);
+  }
+
+  UpdateResponseData data;
+  data.id = GenerateId(tag);
+  data.client_tag_hash = GenerateTagHash(tag);
+  data.response_version = version;
+  data.ctime = base::Time::UnixEpoch() + base::TimeDelta::FromDays(1);
+  data.mtime = data.ctime + base::TimeDelta::FromSeconds(version);
+  data.non_unique_name = tag;
+  data.deleted = false;
+  data.specifics = GenerateSpecifics(tag, value);
+
+  UpdateResponseDataList list;
+  list.push_back(data);
+
+  processor_->ReceiveUpdateResponse(data_type_state_, list);
+}
+
+void NonBlockingTypeProcessorTest::TombstoneFromServer(int64 version_offset,
+                                                       const std::string& tag) {
+  // Overwrite the existing server version if this is the new highest version.
+  int64 old_version = GetServerVersion(tag);
+  int64 version = old_version + version_offset;
+  if (version > old_version) {
+    SetServerVersion(tag, version);
+  }
+
+  UpdateResponseData data;
+  data.id = GenerateId(tag);
+  data.client_tag_hash = GenerateTagHash(tag);
+  data.response_version = version;
+  data.ctime = base::Time::UnixEpoch() + base::TimeDelta::FromDays(1);
+  data.mtime = data.ctime + base::TimeDelta::FromSeconds(version);
+  data.non_unique_name = tag;
+  data.deleted = true;
+
+  UpdateResponseDataList list;
+  list.push_back(data);
+
+  processor_->ReceiveUpdateResponse(data_type_state_, list);
+}
+
+void NonBlockingTypeProcessorTest::SuccessfulCommitResponse(
+    const CommitRequestData& request_data) {
+  const std::string& client_tag = request_data.client_tag;
+  CommitResponseData response_data;
+
+  if (request_data.base_version == 0) {
+    // Server assigns new ID to newly committed items.
+    DCHECK(request_data.id.empty());
+    response_data.id = request_data.id;
+  } else {
+    // Otherwise we reuse the ID from the request.
+    response_data.id = GenerateId(client_tag);
+  }
+
+  response_data.client_tag_hash = GenerateTagHash(client_tag);
+  response_data.sequence_number = request_data.sequence_number;
+
+  // Increment the server version on successful commit.
+  int64 version = GetServerVersion(client_tag);
+  version++;
+  SetServerVersion(client_tag, version);
+
+  response_data.response_version = version;
+
+  CommitResponseDataList list;
+  list.push_back(response_data);
+
+  processor_->ReceiveCommitResponse(data_type_state_, list);
+}
+
+int64 NonBlockingTypeProcessorTest::GetServerVersion(const std::string& tag) {
+  std::map<const std::string, int64>::const_iterator it;
+  it = server_versions_.find(tag);
+  if (it == server_versions_.end()) {
+    return 0;
+  } else {
+    return it->second;
+  }
+}
+
+void NonBlockingTypeProcessorTest::SetServerVersion(const std::string& tag,
+                                                    int64 version) {
+  server_versions_[tag] = version;
+}
+
+std::string NonBlockingTypeProcessorTest::GenerateId(
+    const std::string& tag) const {
+  return "FakeId:" + tag;
+}
+
+std::string NonBlockingTypeProcessorTest::GenerateTagHash(
+    const std::string& tag) const {
+  return syncable::GenerateSyncableHash(type_, tag);
+}
+
+sync_pb::EntitySpecifics NonBlockingTypeProcessorTest::GenerateSpecifics(
+    const std::string& tag,
+    const std::string& value) const {
+  sync_pb::EntitySpecifics specifics;
+  specifics.mutable_preference()->set_name(tag);
+  specifics.mutable_preference()->set_value(value);
+  return specifics;
+}
+
+size_t NonBlockingTypeProcessorTest::GetNumCommitRequestLists() {
+  return mock_processor_core_->commit_request_lists_.size();
+}
+
+CommitRequestDataList NonBlockingTypeProcessorTest::GetNthCommitRequestList(
+    size_t n) {
+  DCHECK_LT(n, GetNumCommitRequestLists());
+  return mock_processor_core_->commit_request_lists_[n];
+}
+
+bool NonBlockingTypeProcessorTest::HasCommitRequestForTag(
+    const std::string& tag) {
+  const std::vector<CommitRequestDataList>& lists =
+      mock_processor_core_->commit_request_lists_;
+
+  // Iterate backward through the sets of commit requests to find the most
+  // recent one that applies to the specified tag.
+  for (std::vector<CommitRequestDataList>::const_reverse_iterator lists_it =
+           lists.rbegin();
+       lists_it != lists.rend();
+       ++lists_it) {
+    for (CommitRequestDataList::const_iterator it = lists_it->begin();
+         it != lists_it->end();
+         ++it) {
+      if (it->client_tag == tag) {
+        return true;
+      }
+    }
+  }
+
+  return false;
+}
+
+CommitRequestData NonBlockingTypeProcessorTest::GetLatestCommitRequestForTag(
+    const std::string& tag) {
+  const std::vector<CommitRequestDataList>& lists =
+      mock_processor_core_->commit_request_lists_;
+
+  // Iterate backward through the sets of commit requests to find the most
+  // recent one that applies to the specified tag.
+  for (std::vector<CommitRequestDataList>::const_reverse_iterator lists_it =
+           lists.rbegin();
+       lists_it != lists.rend();
+       ++lists_it) {
+    for (CommitRequestDataList::const_iterator it = lists_it->begin();
+         it != lists_it->end();
+         ++it) {
+      if (it->client_tag == tag) {
+        return *it;
+      }
+    }
+  }
+
+  NOTREACHED() << "Could not find any commits for given tag " << tag << ". "
+               << "Test should have checked HasCommitRequestForTag() first.";
+  return CommitRequestData();
+}
+
+// Creates a new item locally.
+// Thoroughly tests the data generated by a local item creation.
+TEST_F(NonBlockingTypeProcessorTest, CreateLocalItem) {
+  Initialize();
+  EXPECT_EQ(0U, GetNumCommitRequestLists());
+
+  WriteItem("tag1", "value1");
+
+  // Verify the commit request this operation has triggered.
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
+  const CommitRequestData& tag1_data = GetLatestCommitRequestForTag("tag1");
+
+  EXPECT_TRUE(tag1_data.id.empty());
+  EXPECT_EQ("tag1", tag1_data.client_tag);
+  EXPECT_EQ(0, tag1_data.base_version);
+  EXPECT_FALSE(tag1_data.ctime.is_null());
+  EXPECT_FALSE(tag1_data.mtime.is_null());
+  EXPECT_FALSE(tag1_data.deleted);
+  EXPECT_EQ("tag1", tag1_data.specifics.preference().name());
+  EXPECT_EQ("value1", tag1_data.specifics.preference().value());
+}
+
+// Creates a new local item then modifies it.
+// Thoroughly tests data generated by modification of server-unknown item.
+TEST_F(NonBlockingTypeProcessorTest, CreateAndModifyLocalItem) {
+  Initialize();
+  EXPECT_EQ(0U, GetNumCommitRequestLists());
+
+  WriteItem("tag1", "value1");
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
+  const CommitRequestData& tag1_v1_data = GetLatestCommitRequestForTag("tag1");
+
+  WriteItem("tag1", "value2");
+  EXPECT_EQ(2U, GetNumCommitRequestLists());
+
+  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
+  const CommitRequestData& tag1_v2_data = GetLatestCommitRequestForTag("tag1");
+
+  // Test some of the relations between old and new commit requests.
+  EXPECT_EQ(tag1_v1_data.specifics.preference().value(), "value1");
+  EXPECT_GT(tag1_v2_data.sequence_number, tag1_v1_data.sequence_number);
+
+  // Perform a thorough examination of the update-generated request.
+  EXPECT_TRUE(tag1_v2_data.id.empty());
+  EXPECT_EQ("tag1", tag1_v2_data.client_tag);
+  EXPECT_EQ(0, tag1_v2_data.base_version);
+  EXPECT_FALSE(tag1_v2_data.ctime.is_null());
+  EXPECT_FALSE(tag1_v2_data.mtime.is_null());
+  EXPECT_FALSE(tag1_v2_data.deleted);
+  EXPECT_EQ("tag1", tag1_v2_data.specifics.preference().name());
+  EXPECT_EQ("value2", tag1_v2_data.specifics.preference().value());
+}
+
+// Deletes an item we've never seen before.
+// Should have no effect and not crash.
+TEST_F(NonBlockingTypeProcessorTest, DeleteUnknown) {
+  Initialize();
+
+  DeleteItem("tag1");
+  EXPECT_EQ(0U, GetNumCommitRequestLists());
+}
+
+// Creates an item locally then deletes it.
+//
+// In this test, no commit responses are received, so the deleted item is
+// server-unknown as far as the model thread is concerned.  That behavior
+// is race-dependent; other tests are used to test other races.
+TEST_F(NonBlockingTypeProcessorTest, DeleteServerUnknown) {
+  Initialize();
+
+  WriteItem("tag1", "value1");
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
+  const CommitRequestData& tag1_v1_data = GetLatestCommitRequestForTag("tag1");
+
+  DeleteItem("tag1");
+  EXPECT_EQ(2U, GetNumCommitRequestLists());
+  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
+  const CommitRequestData& tag1_v2_data = GetLatestCommitRequestForTag("tag1");
+
+  EXPECT_GT(tag1_v2_data.sequence_number, tag1_v1_data.sequence_number);
+
+  EXPECT_TRUE(tag1_v2_data.id.empty());
+  EXPECT_EQ("tag1", tag1_v2_data.client_tag);
+  EXPECT_EQ(0, tag1_v2_data.base_version);
+  EXPECT_TRUE(tag1_v2_data.deleted);
+}
+
+// Creates an item locally then deletes it.
+//
+// The item is created locally then enqueued for commit.  The sync thread
+// successfully commits it, but, before the commit response is picked up
+// by the model thread, the item is deleted by the model thread.
+TEST_F(NonBlockingTypeProcessorTest, DeleteServerUnknown_RacyCommitResponse) {

[Nicolas Zea, 2014/05/19 23:54:15]

Is this actually testing anything in its current state?

[rlarocque, 2014/05/20 01:39:47]

It's exercising the internals of this code in an unusual way, though you'd need
a white-box perspective to know that.  It proves the syncer won't crash or leak
memory when this happens.

I'd prefer to test externally observable side effects, like whether or not this
scenario generates a spurious delete notification.  That testing will have to
wait until we have an interface that emits notifications when deletions are
received.

+  Initialize();
+
+  WriteItem("tag1", "value1");
+  EXPECT_EQ(1U, GetNumCommitRequestLists());
+  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
+  const CommitRequestData& tag1_v1_data = GetLatestCommitRequestForTag("tag1");
+
+  DeleteItem("tag1");
+  EXPECT_EQ(2U, GetNumCommitRequestLists());
+  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
+
+  // This commit happened while the deletion was in progress, but the commit
+  // response didn't arrive on our thread until after the delete was issued to
+  // the sync thread.  It will update some metadata, but won't do much else.
+  SuccessfulCommitResponse(tag1_v1_data);
+
+  // TODO(rlarocque): Verify the state of the item is correct once we get
+  // storage hooked up in these tests.  For example, verify the item is still
+  // marked as deleted.
+}
+
+// Creates two different sync items.
+// Verifies that the second has no effect on the first.
+TEST_F(NonBlockingTypeProcessorTest, TwoIndependentItems) {
+  Initialize();
+  EXPECT_EQ(0U, GetNumCommitRequestLists());
+
+  WriteItem("tag1", "value1");
+
+  // There should be one commit request for this item only.
+  ASSERT_EQ(1U, GetNumCommitRequestLists());
+  EXPECT_EQ(1U, GetNthCommitRequestList(0).size());
+  ASSERT_TRUE(HasCommitRequestForTag("tag1"));
+
+  WriteItem("tag2", "value2");
+
+  // The second write should trigger another single-item commit request.
+  ASSERT_EQ(2U, GetNumCommitRequestLists());
+  EXPECT_EQ(1U, GetNthCommitRequestList(1).size());
+  ASSERT_TRUE(HasCommitRequestForTag("tag2"));
+}
+
+}  // namespace syncer