sync: Refactor NonBlockingTypeProcessorCore tests

Pulls MockNonBlockingTypeProcessor and SimpleMockServer classes out of
the NonBlockingTypeProcessorCore's test harness.

These two new mock classes are currently used in only one place, though
it's probable that they'll be reused elsewhere sooner or later.  Even if
no code is reused, this change is worthwhile because it leads to better
separation of responsibilities and improved documentation.

This is one of serveral follow-ups to the commits that introduced
NonBlockingTypeProcessor and NonBlockingTypeProcessorCore.

BUG=351005

Review URL: https://codereview.chromium.org/330523002

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@278114 0039d316-1c4b-4281-b951-d872f2087c98

7 files changed, 746 insertions, 435 deletions

diff --git a/sync/engine/non_blocking_type_processor_core_unittest.cc b/sync/engine/non_blocking_type_processor_core_unittest.cc
index 87b5647..c775a33 100644
--- a/sync/engine/non_blocking_type_processor_core_unittest.cc
+++ b/sync/engine/non_blocking_type_processor_core_unittest.cc
@@ -4,50 +4,23 @@
 
 #include "sync/engine/non_blocking_type_processor_core.h"
 
-#include "base/basictypes.h"
-#include "base/bind.h"
-#include "base/bind_helpers.h"
-#include "base/callback.h"
+#include "sync/engine/commit_contribution.h"
 #include "sync/engine/non_blocking_sync_common.h"
-#include "sync/engine/non_blocking_type_commit_contribution.h"
 #include "sync/engine/non_blocking_type_processor_interface.h"
+#include "sync/internal_api/public/base/model_type.h"
 #include "sync/protocol/sync.pb.h"
 #include "sync/sessions/status_controller.h"
 #include "sync/syncable/syncable_util.h"
-#include "sync/util/time.h"
+#include "sync/test/engine/mock_non_blocking_type_processor.h"
+#include "sync/test/engine/single_type_mock_server.h"
 
 #include "testing/gtest/include/gtest/gtest.h"
 
-using google::protobuf::RepeatedPtrField;
-
 static const std::string kTypeParentId = "PrefsRootNodeID";
+static const syncer::ModelType kModelType = syncer::PREFERENCES;
 
 namespace syncer {
 
-class NonBlockingTypeProcessorCoreTest;
-
-namespace {
-
-class MockNonBlockingTypeProcessor : public NonBlockingTypeProcessorInterface {
- public:
-  MockNonBlockingTypeProcessor(NonBlockingTypeProcessorCoreTest* parent);
-  virtual ~MockNonBlockingTypeProcessor();
-
-  virtual void ReceiveCommitResponse(
-      const DataTypeState& type_state,
-      const CommitResponseDataList& response_list) OVERRIDE;
-  virtual void ReceiveUpdateResponse(
-      const DataTypeState& type_state,
-      const UpdateResponseDataList& response_list) OVERRIDE;
-
- private:
-  NonBlockingTypeProcessorCoreTest* parent_;
-
-  DISALLOW_COPY_AND_ASSIGN(MockNonBlockingTypeProcessor);
-};
-
-}  // namespace
-
 // Tests the NonBlockingTypeProcessorCore.
 //
 // This class passes messages between the model thread and sync server.
@@ -105,15 +78,6 @@ class NonBlockingTypeProcessorCoreTest : public ::testing::Test {
                                const std::string& value);
   void TriggerTombstoneFromServer(int64 version_offset, const std::string& tag);
 
-  // Callbacks from the mock processor.  Called when the |core_| tries to send
-  // messages to its associated processor on the model thread.
-  void OnModelThreadReceivedCommitResponse(
-      const DataTypeState& type_state,
-      const CommitResponseDataList& response_list);
-  void OnModelThreadReceivedUpdateResponse(
-      const DataTypeState& type_state,
-      const UpdateResponseDataList& response_list);
-
   // By default, this harness behaves as if all tasks posted to the model
   // thread are executed immediately.  However, this is not necessarily true.
   // The model's TaskRunner has a queue, and the tasks we post to it could
@@ -169,97 +133,26 @@ class NonBlockingTypeProcessorCoreTest : public ::testing::Test {
       const std::string& tag) const;
 
   // Helpers for building various messages and structures.
-  static std::string GenerateId(const std::string& tag_hash);
   static std::string GenerateTagHash(const std::string& tag);
   static sync_pb::EntitySpecifics GenerateSpecifics(const std::string& tag,
                                                     const std::string& value);
 
  private:
-  // Get and set our emulated server state.
-  int64 GetServerVersion(const std::string& tag_hash);
-  void SetServerVersion(const std::string& tag_hash, int64 version);
-
-  // Get and set our emulated model thread state.
-  int64 GetCurrentSequenceNumber(const std::string& tag_hash) const;
-  int64 GetNextSequenceNumber(const std::string& tag_hash);
-  int64 GetModelVersion(const std::string& tag_hash) const;
-  void SetModelVersion(const std::string& tag_hash, int64 version);
-
-  // Receive a commit response in the emulated model thread.
-  //
-  // Kept in a separate Impl method so we can emulate deferred task processing.
-  // See SetModelThreadIsSynchronous() for details.
-  void ModelThreadReceiveCommitResponseImpl(
-      const DataTypeState& type_state,
-      const CommitResponseDataList& response_list);
-
-  // Receive an update response in the emulated model thread.
-  //
-  // Kept in a separate Impl method so we can emulate deferred task processing.
-  // See SetModelThreadIsSynchronous() for details.
-  void ModelThreadReceiveUpdateResponseImpl(
-      const DataTypeState& type_state,
-      const UpdateResponseDataList& response_list);
-
-  // Builds a fake progress marker for our response.
-  sync_pb::DataTypeProgressMarker GenerateResponseProgressMarker() const;
-
+  // The NonBlockingTypeProcessorCore being tested.
   scoped_ptr<NonBlockingTypeProcessorCore> core_;
+
+  // Non-owned, possibly NULL pointer.  This object belongs to the
+  // NonBlockingTypeProcessorCore under test.
   MockNonBlockingTypeProcessor* mock_processor_;
 
-  // Model thread state maps.
-  std::map<const std::string, int64> model_sequence_numbers_;
-  std::map<const std::string, int64> model_base_versions_;
-
-  // Server state maps.
-  std::map<const std::string, int64> server_versions_;
-
-  // Logs of messages sent to the server.  Used in assertions.
-  std::map<const std::string, sync_pb::SyncEntity> committed_items_;
-  std::vector<sync_pb::ClientToServerMessage> commit_messages_;
-
-  // State related to emulation of the model thread's task queue.  Used to
-  // defer model thread work to simulate a full model thread task runner queue.
-  bool model_thread_is_synchronous_;
-  std::vector<base::Closure> model_thread_tasks_;
-
-  // A cache of messages sent to the model thread.
-  std::vector<CommitResponseDataList> commit_responses_to_model_thread_;
-  std::vector<UpdateResponseDataList> updates_responses_to_model_thread_;
-  std::vector<DataTypeState> updates_states_to_model_thread_;
-  std::vector<DataTypeState> commit_states_to_model_thread_;
-
-  // A cache of the latest responses on the model thread, by client tag.
-  std::map<const std::string, CommitResponseData>
-      model_thread_commit_response_items_;
-  std::map<const std::string, UpdateResponseData>
-      model_thread_update_response_items_;
+  // A mock that emulates enough of the sync server that it can be used
+  // a single UpdateHandler and CommitContributor pair.  In this test
+  // harness, the |core_| is both of them.
+  SingleTypeMockServer mock_server_;
 };
 
-// These had to wait until the class definition of
-// NonBlockingTypeProcessorCoreTest
-MockNonBlockingTypeProcessor::MockNonBlockingTypeProcessor(
-    NonBlockingTypeProcessorCoreTest* parent)
-    : parent_(parent) {
-}
-
-MockNonBlockingTypeProcessor::~MockNonBlockingTypeProcessor() {
-}
-
-void MockNonBlockingTypeProcessor::ReceiveCommitResponse(
-    const DataTypeState& type_state,
-    const CommitResponseDataList& response_list) {
-  parent_->OnModelThreadReceivedCommitResponse(type_state, response_list);
-}
-
-void MockNonBlockingTypeProcessor::ReceiveUpdateResponse(
-    const DataTypeState& type_state,
-    const UpdateResponseDataList& response_list) {
-  parent_->OnModelThreadReceivedUpdateResponse(type_state, response_list);
-}
-
 NonBlockingTypeProcessorCoreTest::NonBlockingTypeProcessorCoreTest()
-    : model_thread_is_synchronous_(true) {
+    : mock_processor_(NULL), mock_server_(kModelType) {
 }
 
 NonBlockingTypeProcessorCoreTest::~NonBlockingTypeProcessorCoreTest() {
@@ -268,7 +161,7 @@ NonBlockingTypeProcessorCoreTest::~NonBlockingTypeProcessorCoreTest() {
 void NonBlockingTypeProcessorCoreTest::FirstInitialize() {
   DataTypeState initial_state;
   initial_state.progress_marker.set_data_type_id(
-      GetSpecificsFieldNumberFromModelType(PREFERENCES));
+      GetSpecificsFieldNumberFromModelType(kModelType));
   initial_state.next_client_id = 0;
 
   InitializeWithState(initial_state);
@@ -277,7 +170,7 @@ void NonBlockingTypeProcessorCoreTest::FirstInitialize() {
 void NonBlockingTypeProcessorCoreTest::NormalInitialize() {
   DataTypeState initial_state;
   initial_state.progress_marker.set_data_type_id(
-      GetSpecificsFieldNumberFromModelType(PREFERENCES));
+      GetSpecificsFieldNumberFromModelType(kModelType));
   initial_state.progress_marker.set_token("some_saved_progress_token");
 
   initial_state.next_client_id = 10;
@@ -292,91 +185,42 @@ void NonBlockingTypeProcessorCoreTest::InitializeWithState(
   DCHECK(!core_);
 
   // We don't get to own this interace.  The |core_| keeps a scoped_ptr to it.
-  mock_processor_ = new MockNonBlockingTypeProcessor(this);
+  mock_processor_ = new MockNonBlockingTypeProcessor();
   scoped_ptr<NonBlockingTypeProcessorInterface> interface(mock_processor_);
 
   core_.reset(
-      new NonBlockingTypeProcessorCore(PREFERENCES, state, interface.Pass()));
+      new NonBlockingTypeProcessorCore(kModelType, state, interface.Pass()));
 }
 
-void NonBlockingTypeProcessorCoreTest::CommitRequest(const std::string& tag,
+void NonBlockingTypeProcessorCoreTest::CommitRequest(const std::string& name,
                                                      const std::string& value) {
-  const std::string tag_hash = GenerateTagHash(tag);
-  const int64 base_version = GetModelVersion(tag_hash);
-
-  CommitRequestData data;
-
-  // Initial commits don't have IDs.  Everything else does.
-  if (base_version > kUncommittedVersion) {
-    data.id = GenerateId(tag_hash);
-  }
-
-  data.client_tag_hash = tag_hash;
-  data.sequence_number = GetNextSequenceNumber(tag_hash);
-
-  data.base_version = base_version;
-  data.ctime = base::Time::UnixEpoch() + base::TimeDelta::FromDays(1);
-  data.mtime = data.ctime + base::TimeDelta::FromSeconds(base_version);
-  data.non_unique_name = tag;
-
-  data.deleted = false;
-  data.specifics = GenerateSpecifics(tag, value);
-
+  const std::string tag_hash = GenerateTagHash(name);
+  CommitRequestData data =
+      mock_processor_->CommitRequest(tag_hash, GenerateSpecifics(name, value));
   CommitRequestDataList list;
   list.push_back(data);
-
   core_->EnqueueForCommit(list);
 }
 
 void NonBlockingTypeProcessorCoreTest::DeleteRequest(const std::string& tag) {
   const std::string tag_hash = GenerateTagHash(tag);
-  const int64 base_version = GetModelVersion(tag_hash);
-  CommitRequestData data;
-
-  // Requests to commit server-unknown items don't have IDs.
-  // We'll never send a deletion for a server-unknown item, but the model is
-  // allowed to request that we do.
-  if (base_version > kUncommittedVersion) {
-    data.id = GenerateId(tag_hash);
-  }
-
-  data.client_tag_hash = tag_hash;
-  data.sequence_number = GetNextSequenceNumber(tag_hash);
-
-  data.base_version = base_version;
-  data.ctime = base::Time::UnixEpoch() + base::TimeDelta::FromDays(1);
-  data.client_tag_hash = tag_hash;
-  data.mtime = data.ctime + base::TimeDelta::FromSeconds(base_version);
-  data.deleted = true;
-
+  CommitRequestData data = mock_processor_->DeleteRequest(tag_hash);
   CommitRequestDataList list;
   list.push_back(data);
-
   core_->EnqueueForCommit(list);
 }
 
 void NonBlockingTypeProcessorCoreTest::TriggerTypeRootUpdateFromServer() {
-  sync_pb::SyncEntity entity;
-
-  entity.set_id_string(kTypeParentId);
-  entity.set_parent_id_string("r");
-  entity.set_version(1000);
-  entity.set_ctime(TimeToProtoTime(base::Time::UnixEpoch()));
-  entity.set_mtime(TimeToProtoTime(base::Time::UnixEpoch()));
-  entity.set_server_defined_unique_tag(ModelTypeToRootTag(PREFERENCES));
-  entity.set_deleted(false);
-  AddDefaultFieldValue(PREFERENCES, entity.mutable_specifics());
-
-  const sync_pb::DataTypeProgressMarker& progress =
-      GenerateResponseProgressMarker();
-  const sync_pb::DataTypeContext blank_context;
-  sessions::StatusController dummy_status;
-
+  sync_pb::SyncEntity entity = mock_server_.TypeRootUpdate();
   SyncEntityList entity_list;
   entity_list.push_back(&entity);
 
-  core_->ProcessGetUpdatesResponse(
-      progress, blank_context, entity_list, &dummy_status);
+  sessions::StatusController dummy_status;
+
+  core_->ProcessGetUpdatesResponse(mock_server_.GetProgress(),
+                                   mock_server_.GetContext(),
+                                   entity_list,
+                                   &dummy_status);
   core_->ApplyUpdates(&dummy_status);
 }
 
@@ -384,102 +228,44 @@ void NonBlockingTypeProcessorCoreTest::TriggerUpdateFromServer(
     int64 version_offset,
     const std::string& tag,
     const std::string& value) {
-  const std::string tag_hash = GenerateTagHash(tag);
-
-  int64 old_version = GetServerVersion(tag_hash);
-  int64 version = old_version + version_offset;
-  if (version > old_version) {
-    SetServerVersion(tag_hash, version);
-  }
-
-  sync_pb::SyncEntity entity;
-
-  entity.set_id_string(GenerateId(tag_hash));
-  entity.set_parent_id_string(kTypeParentId);
-  entity.set_version(version);
-
-  base::Time ctime = base::Time::UnixEpoch() + base::TimeDelta::FromDays(1);
-  base::Time mtime = ctime + base::TimeDelta::FromSeconds(version);
-  entity.set_ctime(TimeToProtoTime(ctime));
-  entity.set_mtime(TimeToProtoTime(mtime));
-
-  entity.set_name(tag);
-  entity.set_client_defined_unique_tag(GenerateTagHash(tag));
-  entity.set_deleted(false);
-  entity.mutable_specifics()->CopyFrom(GenerateSpecifics(tag, value));
-
+  sync_pb::SyncEntity entity = mock_server_.UpdateFromServer(
+      version_offset, GenerateTagHash(tag), GenerateSpecifics(tag, value));
   SyncEntityList entity_list;
   entity_list.push_back(&entity);
 
-  const sync_pb::DataTypeProgressMarker& progress =
-      GenerateResponseProgressMarker();
-  const sync_pb::DataTypeContext blank_context;
   sessions::StatusController dummy_status;
 
-  core_->ProcessGetUpdatesResponse(
-      progress, blank_context, entity_list, &dummy_status);
+  core_->ProcessGetUpdatesResponse(mock_server_.GetProgress(),
+                                   mock_server_.GetContext(),
+                                   entity_list,
+                                   &dummy_status);
   core_->ApplyUpdates(&dummy_status);
 }
 
 void NonBlockingTypeProcessorCoreTest::TriggerTombstoneFromServer(
     int64 version_offset,
     const std::string& tag) {
-  const std::string tag_hash = GenerateTagHash(tag);
-  int64 old_version = GetServerVersion(tag_hash);
-  int64 version = old_version + version_offset;
-  if (version > old_version) {
-    SetServerVersion(tag_hash, version);
-  }
+  sync_pb::SyncEntity entity =
+      mock_server_.TombstoneFromServer(version_offset, GenerateTagHash(tag));
+  SyncEntityList entity_list;
+  entity_list.push_back(&entity);
+
+  sessions::StatusController dummy_status;
 
-  UpdateResponseData data;
-
-  data.id = GenerateId(tag_hash);
-  data.client_tag_hash = tag_hash;
-  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;
-}
-
-void NonBlockingTypeProcessorCoreTest::OnModelThreadReceivedCommitResponse(
-    const DataTypeState& type_state,
-    const CommitResponseDataList& response_list) {
-  base::Closure task = base::Bind(
-      &NonBlockingTypeProcessorCoreTest::ModelThreadReceiveCommitResponseImpl,
-      base::Unretained(this),
-      type_state,
-      response_list);
-  model_thread_tasks_.push_back(task);
-  if (model_thread_is_synchronous_)
-    PumpModelThread();
-}
-
-void NonBlockingTypeProcessorCoreTest::OnModelThreadReceivedUpdateResponse(
-    const DataTypeState& type_state,
-    const UpdateResponseDataList& response_list) {
-  base::Closure task = base::Bind(
-      &NonBlockingTypeProcessorCoreTest::ModelThreadReceiveUpdateResponseImpl,
-      base::Unretained(this),
-      type_state,
-      response_list);
-  model_thread_tasks_.push_back(task);
-  if (model_thread_is_synchronous_)
-    PumpModelThread();
+  core_->ProcessGetUpdatesResponse(mock_server_.GetProgress(),
+                                   mock_server_.GetContext(),
+                                   entity_list,
+                                   &dummy_status);
+  core_->ApplyUpdates(&dummy_status);
 }
 
 void NonBlockingTypeProcessorCoreTest::SetModelThreadIsSynchronous(
     bool is_synchronous) {
-  model_thread_is_synchronous_ = is_synchronous;
+  mock_processor_->SetSynchronousExecution(is_synchronous);
 }
 
 void NonBlockingTypeProcessorCoreTest::PumpModelThread() {
-  for (std::vector<base::Closure>::iterator it = model_thread_tasks_.begin();
-       it != model_thread_tasks_.end();
-       ++it) {
-    it->Run();
-  }
-  model_thread_tasks_.clear();
+  mock_processor_->RunQueuedTasks();
 }
 
 bool NonBlockingTypeProcessorCoreTest::WillCommit() {
@@ -503,35 +289,9 @@ void NonBlockingTypeProcessorCoreTest::DoSuccessfulCommit() {
 
   sync_pb::ClientToServerMessage message;
   contribution->AddToCommitMessage(&message);
-  commit_messages_.push_back(message);
-
-  sync_pb::ClientToServerResponse response;
-  sync_pb::CommitResponse* commit_response = response.mutable_commit();
-
-  const RepeatedPtrField<sync_pb::SyncEntity>& entries =
-      message.commit().entries();
-  for (RepeatedPtrField<sync_pb::SyncEntity>::const_iterator it =
-           entries.begin();
-       it != entries.end();
-       ++it) {
-    const std::string tag_hash = it->client_defined_unique_tag();
-
-    committed_items_[tag_hash] = *it;
-
-    // Every commit increments the version number.
-    int64 version = GetServerVersion(tag_hash);
-    version++;
-    SetServerVersion(tag_hash, version);
-
-    sync_pb::CommitResponse_EntryResponse* entryresponse =
-        commit_response->add_entryresponse();
-    entryresponse->set_response_type(sync_pb::CommitResponse::SUCCESS);
-    entryresponse->set_id_string(GenerateId(tag_hash));
-    entryresponse->set_parent_id_string(it->parent_id_string());
-    entryresponse->set_version(version);
-    entryresponse->set_name(it->name());
-    entryresponse->set_mtime(it->mtime());
-  }
+
+  sync_pb::ClientToServerResponse response =
+      mock_server_.DoSuccessfulCommit(message);
 
   sessions::StatusController dummy_status;
   contribution->ProcessCommitResponse(response, &dummy_status);
@@ -539,21 +299,19 @@ void NonBlockingTypeProcessorCoreTest::DoSuccessfulCommit() {
 }
 
 size_t NonBlockingTypeProcessorCoreTest::GetNumCommitMessagesOnServer() const {
-  return commit_messages_.size();
+  return mock_server_.GetNumCommitMessages();
 }
 
 sync_pb::ClientToServerMessage
 NonBlockingTypeProcessorCoreTest::GetNthCommitMessageOnServer(size_t n) const {
   DCHECK_LT(n, GetNumCommitMessagesOnServer());
-  return commit_messages_[n];
+  return mock_server_.GetNthCommitMessage(n);
 }
 
 bool NonBlockingTypeProcessorCoreTest::HasCommitEntityOnServer(
     const std::string& tag) const {
   const std::string tag_hash = GenerateTagHash(tag);
-  std::map<const std::string, sync_pb::SyncEntity>::const_iterator it =
-      committed_items_.find(tag_hash);
-  return it != committed_items_.end();
+  return mock_server_.HasCommitEntity(tag_hash);
 }
 
 sync_pb::SyncEntity
@@ -561,71 +319,62 @@ NonBlockingTypeProcessorCoreTest::GetLatestCommitEntityOnServer(
     const std::string& tag) const {
   DCHECK(HasCommitEntityOnServer(tag));
   const std::string tag_hash = GenerateTagHash(tag);
-  std::map<const std::string, sync_pb::SyncEntity>::const_iterator it =
-      committed_items_.find(tag_hash);
-  return it->second;
+  return mock_server_.GetLastCommittedEntity(tag_hash);
 }
 
 size_t NonBlockingTypeProcessorCoreTest::GetNumModelThreadUpdateResponses()
     const {
-  return updates_responses_to_model_thread_.size();
+  return mock_processor_->GetNumUpdateResponses();
 }
 
 UpdateResponseDataList
 NonBlockingTypeProcessorCoreTest::GetNthModelThreadUpdateResponse(
     size_t n) const {
-  DCHECK(GetNumModelThreadUpdateResponses());
-  return updates_responses_to_model_thread_[n];
+  DCHECK_LT(n, GetNumModelThreadUpdateResponses());
+  return mock_processor_->GetNthUpdateResponse(n);
 }
 
 DataTypeState NonBlockingTypeProcessorCoreTest::GetNthModelThreadUpdateState(
     size_t n) const {
-  DCHECK(GetNumModelThreadUpdateResponses());
-  return updates_states_to_model_thread_[n];
+  DCHECK_LT(n, GetNumModelThreadUpdateResponses());
+  return mock_processor_->GetNthTypeStateReceivedInUpdateResponse(n);
 }
 
 bool NonBlockingTypeProcessorCoreTest::HasUpdateResponseOnModelThread(
     const std::string& tag) const {
   const std::string tag_hash = GenerateTagHash(tag);
-  std::map<const std::string, UpdateResponseData>::const_iterator it =
-      model_thread_update_response_items_.find(tag_hash);
-  return it != model_thread_update_response_items_.end();
+  return mock_processor_->HasUpdateResponse(tag_hash);
 }
 
 UpdateResponseData
 NonBlockingTypeProcessorCoreTest::GetUpdateResponseOnModelThread(
     const std::string& tag) const {
   const std::string tag_hash = GenerateTagHash(tag);
-  DCHECK(HasUpdateResponseOnModelThread(tag));
-  std::map<const std::string, UpdateResponseData>::const_iterator it =
-      model_thread_update_response_items_.find(tag_hash);
-  return it->second;
+  return mock_processor_->GetUpdateResponse(tag_hash);
 }
 
 size_t NonBlockingTypeProcessorCoreTest::GetNumModelThreadCommitResponses()
     const {
-  return commit_responses_to_model_thread_.size();
+  return mock_processor_->GetNumCommitResponses();
 }
 
 CommitResponseDataList
 NonBlockingTypeProcessorCoreTest::GetNthModelThreadCommitResponse(
     size_t n) const {
-  DCHECK(GetNumModelThreadCommitResponses());
-  return commit_responses_to_model_thread_[n];
+  DCHECK_LT(n, GetNumModelThreadCommitResponses());
+  return mock_processor_->GetNthCommitResponse(n);
 }
 
 DataTypeState NonBlockingTypeProcessorCoreTest::GetNthModelThreadCommitState(
     size_t n) const {
-  DCHECK(GetNumModelThreadCommitResponses());
-  return commit_states_to_model_thread_[n];
+  DCHECK_LT(n, GetNumModelThreadCommitResponses());
+  return mock_processor_->GetNthTypeStateReceivedInCommitResponse(n);
 }
 
 bool NonBlockingTypeProcessorCoreTest::HasCommitResponseOnModelThread(
     const std::string& tag) const {
   const std::string tag_hash = GenerateTagHash(tag);
-  std::map<const std::string, CommitResponseData>::const_iterator it =
-      model_thread_commit_response_items_.find(tag_hash);
-  return it != model_thread_commit_response_items_.end();
+  return mock_processor_->HasCommitResponse(tag_hash);
 }
 
 CommitResponseData
@@ -633,21 +382,13 @@ NonBlockingTypeProcessorCoreTest::GetCommitResponseOnModelThread(
     const std::string& tag) const {
   DCHECK(HasCommitResponseOnModelThread(tag));
   const std::string tag_hash = GenerateTagHash(tag);
-  std::map<const std::string, CommitResponseData>::const_iterator it =
-      model_thread_commit_response_items_.find(tag_hash);
-  return it->second;
-}
-
-std::string NonBlockingTypeProcessorCoreTest::GenerateId(
-    const std::string& tag_hash) {
-  return "FakeId:" + tag_hash;
+  return mock_processor_->GetCommitResponse(tag_hash);
 }
 
 std::string NonBlockingTypeProcessorCoreTest::GenerateTagHash(
     const std::string& tag) {
   const std::string& client_tag_hash =
-      syncable::GenerateSyncableHash(PREFERENCES, tag);
-
+      syncable::GenerateSyncableHash(kModelType, tag);
   return client_tag_hash;
 }
 
@@ -660,104 +401,6 @@ sync_pb::EntitySpecifics NonBlockingTypeProcessorCoreTest::GenerateSpecifics(
   return specifics;
 }
 
-int64 NonBlockingTypeProcessorCoreTest::GetServerVersion(
-    const std::string& tag_hash) {
-  std::map<const std::string, int64>::const_iterator it;
-  it = server_versions_.find(tag_hash);
-  // Server versions do not necessarily start at 1 or 0.
-  if (it == server_versions_.end()) {
-    return 2048;
-  } else {
-    return it->second;
-  }
-}
-
-void NonBlockingTypeProcessorCoreTest::SetServerVersion(
-    const std::string& tag_hash,
-    int64 version) {
-  server_versions_[tag_hash] = version;
-}
-
-// Fetches the sequence number as of the most recent update request.
-int64 NonBlockingTypeProcessorCoreTest::GetCurrentSequenceNumber(
-    const std::string& tag_hash) const {
-  std::map<const std::string, int64>::const_iterator it =
-      model_sequence_numbers_.find(tag_hash);
-  if (it == model_sequence_numbers_.end()) {
-    return 0;
-  } else {
-    return it->second;
-  }
-}
-
-// The model thread should be sending us items with strictly increasing
-// sequence numbers.  Here's where we emulate that behavior.
-int64 NonBlockingTypeProcessorCoreTest::GetNextSequenceNumber(
-    const std::string& tag_hash) {
-  int64 sequence_number = GetCurrentSequenceNumber(tag_hash);
-  sequence_number++;
-  model_sequence_numbers_[tag_hash] = sequence_number;
-  return sequence_number;
-}
-
-// Fetches the model's base version.
-int64 NonBlockingTypeProcessorCoreTest::GetModelVersion(
-    const std::string& tag_hash) const {
-  std::map<const std::string, int64>::const_iterator it =
-      model_base_versions_.find(tag_hash);
-  if (it == model_base_versions_.end()) {
-    return kUncommittedVersion;
-  } else {
-    return it->second;
-  }
-}
-
-void NonBlockingTypeProcessorCoreTest::SetModelVersion(
-    const std::string& tag_hash,
-    int64 version) {
-  model_base_versions_[tag_hash] = version;
-}
-
-void NonBlockingTypeProcessorCoreTest::ModelThreadReceiveCommitResponseImpl(
-    const DataTypeState& type_state,
-    const CommitResponseDataList& response_list) {
-  commit_responses_to_model_thread_.push_back(response_list);
-  commit_states_to_model_thread_.push_back(type_state);
-  for (CommitResponseDataList::const_iterator it = response_list.begin();
-       it != response_list.end();
-       ++it) {
-    model_thread_commit_response_items_.insert(
-        std::make_pair(it->client_tag_hash, *it));
-
-    // Server wins.  Set the model's base version.
-    SetModelVersion(it->client_tag_hash, it->response_version);
-  }
-}
-
-void NonBlockingTypeProcessorCoreTest::ModelThreadReceiveUpdateResponseImpl(
-    const DataTypeState& type_state,
-    const UpdateResponseDataList& response_list) {
-  updates_responses_to_model_thread_.push_back(response_list);
-  updates_states_to_model_thread_.push_back(type_state);
-  for (UpdateResponseDataList::const_iterator it = response_list.begin();
-       it != response_list.end();
-       ++it) {
-    model_thread_update_response_items_.insert(
-        std::make_pair(it->client_tag_hash, *it));
-
-    // Server wins.  Set the model's base version.
-    SetModelVersion(it->client_tag_hash, it->response_version);
-  }
-}
-
-sync_pb::DataTypeProgressMarker
-NonBlockingTypeProcessorCoreTest::GenerateResponseProgressMarker() const {
-  sync_pb::DataTypeProgressMarker progress;
-  progress.set_data_type_id(PREFERENCES);
-  progress.set_token("non_null_progress_token");
-  return progress;
-}
-
 // Requests a commit and verifies the messages sent to the client and server as
 // a result.
 //
@@ -790,7 +433,7 @@ TEST_F(NonBlockingTypeProcessorCoreTest, SimpleCommit) {
   EXPECT_EQ(kUncommittedVersion, entity.version());
   EXPECT_NE(0, entity.mtime());
   EXPECT_NE(0, entity.ctime());
-  EXPECT_EQ("tag1", entity.name());
+  EXPECT_FALSE(entity.name().empty());
   EXPECT_EQ(client_tag_hash, entity.client_defined_unique_tag());
   EXPECT_EQ("tag1", entity.specifics().preference().name());
   EXPECT_FALSE(entity.deleted());
@@ -842,7 +485,7 @@ TEST_F(NonBlockingTypeProcessorCoreTest, SimpleDelete) {
 
   // Deletions should contain enough specifics to identify the type.
   EXPECT_TRUE(entity.has_specifics());
-  EXPECT_EQ(PREFERENCES, GetModelTypeFromSpecifics(entity.specifics()));
+  EXPECT_EQ(kModelType, GetModelTypeFromSpecifics(entity.specifics()));
 
   // Verify the commit response returned to the model thread.
   ASSERT_EQ(2U, GetNumModelThreadCommitResponses());
@@ -952,7 +595,7 @@ TEST_F(NonBlockingTypeProcessorCoreTest, ReceiveUpdates) {
   EXPECT_LT(0, update.response_version);
   EXPECT_FALSE(update.ctime.is_null());
   EXPECT_FALSE(update.mtime.is_null());
-  EXPECT_EQ("tag1", update.non_unique_name);
+  EXPECT_FALSE(update.non_unique_name.empty());
   EXPECT_FALSE(update.deleted);
   EXPECT_EQ("tag1", update.specifics.preference().name());
   EXPECT_EQ("value1", update.specifics.preference().value());
diff --git a/sync/engine/non_blocking_type_processor_interface.h b/sync/engine/non_blocking_type_processor_interface.h
index ccc504b..bafd141 100644
--- a/sync/engine/non_blocking_type_processor_interface.h
+++ b/sync/engine/non_blocking_type_processor_interface.h
@@ -6,10 +6,7 @@
 #define SYNC_ENGINE_NON_BLOCKING_TYPE_PROCESSOR_INTERFACE_H_
 
 #include "sync/base/sync_export.h"
-
-struct CommitResponseDataList;
-struct DataTypeState;
-struct UpdateResponseDataList;
+#include "sync/engine/non_blocking_sync_common.h"
 
 namespace syncer {
 
diff --git a/sync/sync_tests.gypi b/sync/sync_tests.gypi
index 4cb5d17..6212dd4 100644
--- a/sync/sync_tests.gypi
+++ b/sync/sync_tests.gypi
@@ -42,8 +42,12 @@
         'test/engine/mock_connection_manager.h',
         'test/engine/mock_non_blocking_type_processor_core.cc',
         'test/engine/mock_non_blocking_type_processor_core.h',
+        'test/engine/mock_non_blocking_type_processor.cc',
+        'test/engine/mock_non_blocking_type_processor.h',
         'test/engine/mock_update_handler.cc',
         'test/engine/mock_update_handler.h',
+        'test/engine/single_type_mock_server.cc',
+        'test/engine/single_type_mock_server.h',
         'test/engine/test_directory_setter_upper.cc',
         'test/engine/test_directory_setter_upper.h',
         'test/engine/test_id_factory.h',
diff --git a/sync/test/engine/mock_non_blocking_type_processor.cc b/sync/test/engine/mock_non_blocking_type_processor.cc
new file mode 100644
index 0000000..0c3d8c9
--- /dev/null
+++ b/sync/test/engine/mock_non_blocking_type_processor.cc
@@ -0,0 +1,258 @@
+// 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/test/engine/mock_non_blocking_type_processor.h"
+
+#include "base/bind.h"
+
+namespace syncer {
+
+MockNonBlockingTypeProcessor::MockNonBlockingTypeProcessor()
+  : is_synchronous_(true) {
+}
+
+MockNonBlockingTypeProcessor::~MockNonBlockingTypeProcessor() {
+}
+
+void MockNonBlockingTypeProcessor::ReceiveCommitResponse(
+    const DataTypeState& type_state,
+    const CommitResponseDataList& response_list) {
+  base::Closure task =
+      base::Bind(&MockNonBlockingTypeProcessor::ReceiveCommitResponseImpl,
+                 base::Unretained(this),
+                 type_state,
+                 response_list);
+  pending_tasks_.push_back(task);
+  if (is_synchronous_)
+    RunQueuedTasks();
+}
+
+void MockNonBlockingTypeProcessor::ReceiveUpdateResponse(
+    const DataTypeState& type_state,
+    const UpdateResponseDataList& response_list) {
+  base::Closure task =
+      base::Bind(&MockNonBlockingTypeProcessor::ReceiveUpdateResponseImpl,
+                 base::Unretained(this),
+                 type_state,
+                 response_list);
+  pending_tasks_.push_back(task);
+  if (is_synchronous_)
+    RunQueuedTasks();
+}
+
+void MockNonBlockingTypeProcessor::SetSynchronousExecution(
+    bool is_synchronous) {
+  is_synchronous_ = is_synchronous;
+}
+
+void MockNonBlockingTypeProcessor::RunQueuedTasks() {
+  for (std::vector<base::Closure>::iterator it = pending_tasks_.begin();
+       it != pending_tasks_.end();
+       ++it) {
+    it->Run();
+  }
+  pending_tasks_.clear();
+}
+
+CommitRequestData MockNonBlockingTypeProcessor::CommitRequest(
+    const std::string& tag_hash,
+    const sync_pb::EntitySpecifics& specifics) {
+  const int64 base_version = GetBaseVersion(tag_hash);
+
+  CommitRequestData data;
+
+  if (HasServerAssignedId(tag_hash)) {
+    data.id = GetServerAssignedId(tag_hash);
+  }
+
+  data.client_tag_hash = tag_hash;
+  data.sequence_number = GetNextSequenceNumber(tag_hash);
+  data.deleted = false;
+  data.specifics = specifics;
+  data.base_version = base_version;
+
+  // These fields are not really used for much, but we set them anyway
+  // to make this item look more realistic.
+  data.ctime = base::Time::UnixEpoch() + base::TimeDelta::FromDays(1);
+  data.mtime = data.ctime + base::TimeDelta::FromSeconds(base_version);
+  data.non_unique_name = "Name: " + tag_hash;
+
+  return data;
+}
+
+CommitRequestData MockNonBlockingTypeProcessor::DeleteRequest(
+    const std::string& tag_hash) {
+  const int64 base_version = GetBaseVersion(tag_hash);
+  CommitRequestData data;
+
+  if (HasServerAssignedId(tag_hash)) {
+    data.id = GetServerAssignedId(tag_hash);
+  }
+
+  data.client_tag_hash = tag_hash;
+  data.sequence_number = GetNextSequenceNumber(tag_hash);
+  data.base_version = base_version;
+  data.mtime = data.ctime + base::TimeDelta::FromSeconds(base_version);
+  data.deleted = true;
+
+  // These fields have little or no effect on behavior.  We set them anyway to
+  // make the test more realistic.
+  data.ctime = base::Time::UnixEpoch() + base::TimeDelta::FromDays(1);
+  data.non_unique_name = "Name deleted";
+
+  return data;
+}
+
+size_t MockNonBlockingTypeProcessor::GetNumUpdateResponses() const {
+  return received_update_responses_.size();
+}
+
+UpdateResponseDataList MockNonBlockingTypeProcessor::GetNthUpdateResponse(
+    size_t n) const {
+  DCHECK_LT(n, GetNumUpdateResponses());
+  return received_update_responses_[n];
+}
+
+DataTypeState
+MockNonBlockingTypeProcessor::GetNthTypeStateReceivedInUpdateResponse(
+    size_t n) const {
+  DCHECK_LT(n, GetNumUpdateResponses());
+  return type_states_received_on_update_[n];
+}
+
+size_t MockNonBlockingTypeProcessor::GetNumCommitResponses() const {
+  return received_commit_responses_.size();
+}
+
+CommitResponseDataList MockNonBlockingTypeProcessor::GetNthCommitResponse(
+    size_t n) const {
+  DCHECK_LT(n, GetNumCommitResponses());
+  return received_commit_responses_[n];
+}
+
+DataTypeState
+MockNonBlockingTypeProcessor::GetNthTypeStateReceivedInCommitResponse(
+    size_t n) const {
+  DCHECK_LT(n, GetNumCommitResponses());
+  return type_states_received_on_commit_[n];
+}
+
+bool MockNonBlockingTypeProcessor::HasUpdateResponse(
+    const std::string& tag_hash) const {
+  std::map<const std::string, UpdateResponseData>::const_iterator it =
+      update_response_items_.find(tag_hash);
+  return it != update_response_items_.end();
+}
+
+UpdateResponseData MockNonBlockingTypeProcessor::GetUpdateResponse(
+    const std::string& tag_hash) const {
+  DCHECK(HasUpdateResponse(tag_hash));
+  std::map<const std::string, UpdateResponseData>::const_iterator it =
+      update_response_items_.find(tag_hash);
+  return it->second;
+}
+
+bool MockNonBlockingTypeProcessor::HasCommitResponse(
+    const std::string& tag_hash) const {
+  std::map<const std::string, CommitResponseData>::const_iterator it =
+      commit_response_items_.find(tag_hash);
+  return it != commit_response_items_.end();
+}
+
+CommitResponseData MockNonBlockingTypeProcessor::GetCommitResponse(
+    const std::string& tag_hash) const {
+  DCHECK(HasCommitResponse(tag_hash));
+  std::map<const std::string, CommitResponseData>::const_iterator it =
+      commit_response_items_.find(tag_hash);
+  return it->second;
+}
+
+void MockNonBlockingTypeProcessor::ReceiveCommitResponseImpl(
+    const DataTypeState& type_state,
+    const CommitResponseDataList& response_list) {
+  received_commit_responses_.push_back(response_list);
+  type_states_received_on_commit_.push_back(type_state);
+  for (CommitResponseDataList::const_iterator it = response_list.begin();
+       it != response_list.end();
+       ++it) {
+    commit_response_items_.insert(std::make_pair(it->client_tag_hash, *it));
+
+    // Server wins.  Set the model's base version.
+    SetBaseVersion(it->client_tag_hash, it->response_version);
+    SetServerAssignedId(it->client_tag_hash, it->id);
+  }
+}
+
+void MockNonBlockingTypeProcessor::ReceiveUpdateResponseImpl(
+    const DataTypeState& type_state,
+    const UpdateResponseDataList& response_list) {
+  received_update_responses_.push_back(response_list);
+  type_states_received_on_update_.push_back(type_state);
+  for (UpdateResponseDataList::const_iterator it = response_list.begin();
+       it != response_list.end();
+       ++it) {
+    update_response_items_.insert(std::make_pair(it->client_tag_hash, *it));
+
+    // Server wins.  Set the model's base version.
+    SetBaseVersion(it->client_tag_hash, it->response_version);
+    SetServerAssignedId(it->client_tag_hash, it->id);
+  }
+}
+
+// Fetches the sequence number as of the most recent update request.
+int64 MockNonBlockingTypeProcessor::GetCurrentSequenceNumber(
+    const std::string& tag_hash) const {
+  std::map<const std::string, int64>::const_iterator it =
+      sequence_numbers_.find(tag_hash);
+  if (it == sequence_numbers_.end()) {
+    return 0;
+  } else {
+    return it->second;
+  }
+}
+
+// The model thread should be sending us items with strictly increasing
+// sequence numbers.  Here's where we emulate that behavior.
+int64 MockNonBlockingTypeProcessor::GetNextSequenceNumber(
+    const std::string& tag_hash) {
+  int64 sequence_number = GetCurrentSequenceNumber(tag_hash);
+  sequence_number++;
+  sequence_numbers_[tag_hash] = sequence_number;
+  return sequence_number;
+}
+
+int64 MockNonBlockingTypeProcessor::GetBaseVersion(
+    const std::string& tag_hash) const {
+  std::map<const std::string, int64>::const_iterator it =
+      base_versions_.find(tag_hash);
+  if (it == base_versions_.end()) {
+    return kUncommittedVersion;
+  } else {
+    return it->second;
+  }
+}
+
+void MockNonBlockingTypeProcessor::SetBaseVersion(const std::string& tag_hash,
+                                                  int64 version) {
+  base_versions_[tag_hash] = version;
+}
+
+bool MockNonBlockingTypeProcessor::HasServerAssignedId(
+    const std::string& tag_hash) const {
+  return assigned_ids_.find(tag_hash) != assigned_ids_.end();
+}
+
+const std::string& MockNonBlockingTypeProcessor::GetServerAssignedId(
+    const std::string& tag_hash) const {
+  DCHECK(HasServerAssignedId(tag_hash));
+  return assigned_ids_.find(tag_hash)->second;
+}
+
+void MockNonBlockingTypeProcessor::SetServerAssignedId(
+    const std::string& tag_hash,
+    const std::string& id) {
+  assigned_ids_[tag_hash] = id;
+}
+
+}  // namespace syncer
diff --git a/sync/test/engine/mock_non_blocking_type_processor.h b/sync/test/engine/mock_non_blocking_type_processor.h
new file mode 100644
index 0000000..30e7670
--- /dev/null
+++ b/sync/test/engine/mock_non_blocking_type_processor.h
@@ -0,0 +1,136 @@
+// 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.
+
+#ifndef SYNC_TEST_ENGINE_MOCK_NON_BLOCKING_TYPE_PROCESSOR_H_
+#define SYNC_TEST_ENGINE_MOCK_NON_BLOCKING_TYPE_PROCESSOR_H_
+
+#include <vector>
+
+#include "base/callback.h"
+#include "base/macros.h"
+#include "sync/engine/non_blocking_sync_common.h"
+#include "sync/engine/non_blocking_type_processor_interface.h"
+
+namespace syncer {
+
+// Mocks the NonBlockingTypeProcessor.
+//
+// This mock is made simpler by not using any threads.  It does still have the
+// ability to defer execution if we need to test race conditions, though.
+//
+// It maintains some state to try to make its behavior more realistic.  It
+// updates this state as it creates commit requests or receives update and
+// commit responses.
+//
+// It keeps a log of all received messages so tests can make assertions based
+// on their value.
+class MockNonBlockingTypeProcessor : public NonBlockingTypeProcessorInterface {
+ public:
+  MockNonBlockingTypeProcessor();
+  virtual ~MockNonBlockingTypeProcessor();
+
+  // Implementation of NonBlockingTypeProcessorInterface.
+  virtual void ReceiveCommitResponse(
+      const DataTypeState& type_state,
+      const CommitResponseDataList& response_list) OVERRIDE;
+  virtual void ReceiveUpdateResponse(
+      const DataTypeState& type_state,
+      const UpdateResponseDataList& response_list) OVERRIDE;
+
+  // By default, this object behaves as if all messages are processed
+  // immediately.  Sometimes it is useful to defer work until later, as might
+  // happen in the real world if the model thread's task queue gets backed up.
+  void SetSynchronousExecution(bool is_synchronous);
+
+  // Runs any work that was deferred while this class was in asynchronous mode.
+  //
+  // This function is not useful unless this object is set to synchronous
+  // execution mode, which is the default.
+  void RunQueuedTasks();
+
+  // Generate commit or deletion requests to be sent to the server.
+  // These functions update local state to keep sequence numbers consistent.
+  //
+  // A real NonBlockingTypeProcessor would forward these kinds of messages
+  // directly to its attached NonBlockingTypeProcessorCore.  These methods
+  // return the value to the caller so the test framework can handle them as it
+  // sees fit.
+  CommitRequestData CommitRequest(const std::string& tag_hash,
+                                  const sync_pb::EntitySpecifics& specifics);
+  CommitRequestData DeleteRequest(const std::string& tag_hash);
+
+  // Getters to access the log of received update responses.
+  //
+  // Does not includes repsonses that are in pending tasks.
+  size_t GetNumUpdateResponses() const;
+  UpdateResponseDataList GetNthUpdateResponse(size_t n) const;
+  DataTypeState GetNthTypeStateReceivedInUpdateResponse(size_t n) const;
+
+  // Getters to access the log of received commit responses.
+  //
+  // Does not includes repsonses that are in pending tasks.
+  size_t GetNumCommitResponses() const;
+  CommitResponseDataList GetNthCommitResponse(size_t n) const;
+  DataTypeState GetNthTypeStateReceivedInCommitResponse(size_t n) const;
+
+  // Getters to access the lastest update response for a given tag_hash.
+  bool HasUpdateResponse(const std::string& tag_hash) const;
+  UpdateResponseData GetUpdateResponse(const std::string& tag_hash) const;
+
+  // Getters to access the lastest commit response for a given tag_hash.
+  bool HasCommitResponse(const std::string& tag_hash) const;
+  CommitResponseData GetCommitResponse(const std::string& tag_hash) const;
+
+ private:
+  // Process a received commit response.
+  //
+  // Implemented as an Impl method so we can defer its execution in some cases.
+  void ReceiveCommitResponseImpl(const DataTypeState& type_state,
+                                 const CommitResponseDataList& response_list);
+
+  // Process a received update response.
+  //
+  // Implemented as an Impl method so we can defer its execution in some cases.
+  void ReceiveUpdateResponseImpl(const DataTypeState& type_state,
+                                 const UpdateResponseDataList& response_list);
+
+  // Getter and setter for per-item sequence number tracking.
+  int64 GetCurrentSequenceNumber(const std::string& tag_hash) const;
+  int64 GetNextSequenceNumber(const std::string& tag_hash);
+
+  // Getter and setter for per-item base version tracking.
+  int64 GetBaseVersion(const std::string& tag_hash) const;
+  void SetBaseVersion(const std::string& tag_hash, int64 version);
+
+  // Getters and setter for server-assigned ID values.
+  bool HasServerAssignedId(const std::string& tag_hash) const;
+  const std::string& GetServerAssignedId(const std::string& tag_hash) const;
+  void SetServerAssignedId(const std::string& tag_hash, const std::string& id);
+
+  // State related to the implementation of deferred work.
+  // See SetSynchronousExecution() for details.
+  bool is_synchronous_;
+  std::vector<base::Closure> pending_tasks_;
+
+  // A log of messages received by this object.
+  std::vector<CommitResponseDataList> received_commit_responses_;
+  std::vector<UpdateResponseDataList> received_update_responses_;
+  std::vector<DataTypeState> type_states_received_on_update_;
+  std::vector<DataTypeState> type_states_received_on_commit_;
+
+  // Latest responses received, indexed by tag_hash.
+  std::map<const std::string, CommitResponseData> commit_response_items_;
+  std::map<const std::string, UpdateResponseData> update_response_items_;
+
+  // The per-item state maps.
+  std::map<const std::string, int64> sequence_numbers_;
+  std::map<const std::string, int64> base_versions_;
+  std::map<const std::string, std::string> assigned_ids_;
+
+  DISALLOW_COPY_AND_ASSIGN(MockNonBlockingTypeProcessor);
+};
+
+}  // namespace syncer
+
+#endif  // SYNC_TEST_ENGINE_MOCK_NON_BLOCKING_TYPE_PROCESSOR_H_
diff --git a/sync/test/engine/single_type_mock_server.cc b/sync/test/engine/single_type_mock_server.cc
new file mode 100644
index 0000000..6f828fb
--- /dev/null
+++ b/sync/test/engine/single_type_mock_server.cc
@@ -0,0 +1,179 @@
+// 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/test/engine/single_type_mock_server.h"
+
+#include "sync/util/time.h"
+
+using google::protobuf::RepeatedPtrField;
+
+namespace syncer {
+
+SingleTypeMockServer::SingleTypeMockServer(syncer::ModelType type)
+    : type_(type), type_root_id_(ModelTypeToRootTag(type)) {
+}
+
+SingleTypeMockServer::~SingleTypeMockServer() {
+}
+
+sync_pb::SyncEntity SingleTypeMockServer::TypeRootUpdate() {
+  sync_pb::SyncEntity entity;
+
+  entity.set_id_string(type_root_id_);
+  entity.set_parent_id_string("r");
+  entity.set_version(1000);
+  entity.set_ctime(TimeToProtoTime(base::Time::UnixEpoch()));
+  entity.set_mtime(TimeToProtoTime(base::Time::UnixEpoch()));
+  entity.set_server_defined_unique_tag(ModelTypeToRootTag(type_));
+  entity.set_deleted(false);
+  AddDefaultFieldValue(type_, entity.mutable_specifics());
+
+  return entity;
+}
+
+sync_pb::SyncEntity SingleTypeMockServer::UpdateFromServer(
+    int64 version_offset,
+    const std::string& tag_hash,
+    const sync_pb::EntitySpecifics& specifics) {
+  int64 old_version = GetServerVersion(tag_hash);
+  int64 version = old_version + version_offset;
+  if (version > old_version) {
+    SetServerVersion(tag_hash, version);
+  }
+
+  sync_pb::SyncEntity entity;
+
+  entity.set_id_string(GenerateId(tag_hash));
+  entity.set_parent_id_string(type_root_id_);
+  entity.set_version(version);
+  entity.set_client_defined_unique_tag(tag_hash);
+  entity.set_deleted(false);
+  entity.mutable_specifics()->CopyFrom(specifics);
+
+  // Unimportant fields, set for completeness only.
+  base::Time ctime = base::Time::UnixEpoch() + base::TimeDelta::FromDays(1);
+  base::Time mtime = ctime + base::TimeDelta::FromSeconds(version);
+  entity.set_ctime(TimeToProtoTime(ctime));
+  entity.set_mtime(TimeToProtoTime(mtime));
+  entity.set_name("Name: " + tag_hash);
+
+  return entity;
+}
+
+sync_pb::SyncEntity SingleTypeMockServer::TombstoneFromServer(
+    int64 version_offset,
+    const std::string& tag_hash) {
+  int64 old_version = GetServerVersion(tag_hash);
+  int64 version = old_version + version_offset;
+  if (version > old_version) {
+    SetServerVersion(tag_hash, version);
+  }
+
+  sync_pb::SyncEntity entity;
+
+  entity.set_id_string(GenerateId(tag_hash));
+  entity.set_parent_id_string(type_root_id_);
+  entity.set_version(version);
+  entity.set_client_defined_unique_tag(tag_hash);
+  entity.set_deleted(false);
+  AddDefaultFieldValue(type_, entity.mutable_specifics());
+
+  // Unimportant fields, set for completeness only.
+  base::Time ctime = base::Time::UnixEpoch() + base::TimeDelta::FromDays(1);
+  base::Time mtime = ctime + base::TimeDelta::FromSeconds(version);
+  entity.set_ctime(TimeToProtoTime(ctime));
+  entity.set_mtime(TimeToProtoTime(mtime));
+  entity.set_name("Tombstone");
+
+  return entity;
+}
+
+sync_pb::ClientToServerResponse SingleTypeMockServer::DoSuccessfulCommit(
+    const sync_pb::ClientToServerMessage& message) {
+  commit_messages_.push_back(message);
+
+  sync_pb::ClientToServerResponse response;
+  sync_pb::CommitResponse* commit_response = response.mutable_commit();
+
+  const RepeatedPtrField<sync_pb::SyncEntity>& entries =
+      message.commit().entries();
+  for (RepeatedPtrField<sync_pb::SyncEntity>::const_iterator it =
+           entries.begin();
+       it != entries.end();
+       ++it) {
+    const std::string tag_hash = it->client_defined_unique_tag();
+
+    committed_items_[tag_hash] = *it;
+
+    // Every commit increments the version number.
+    int64 version = GetServerVersion(tag_hash);
+    version++;
+    SetServerVersion(tag_hash, version);
+
+    sync_pb::CommitResponse_EntryResponse* entryresponse =
+        commit_response->add_entryresponse();
+    entryresponse->set_response_type(sync_pb::CommitResponse::SUCCESS);
+    entryresponse->set_id_string(GenerateId(tag_hash));
+    entryresponse->set_parent_id_string(it->parent_id_string());
+    entryresponse->set_version(version);
+    entryresponse->set_name(it->name());
+    entryresponse->set_mtime(it->mtime());
+  }
+
+  return response;
+}
+
+size_t SingleTypeMockServer::GetNumCommitMessages() const {
+  return commit_messages_.size();
+}
+
+sync_pb::ClientToServerMessage SingleTypeMockServer::GetNthCommitMessage(
+    size_t n) const {
+  DCHECK_LT(n, GetNumCommitMessages());
+  return commit_messages_[n];
+}
+
+bool SingleTypeMockServer::HasCommitEntity(const std::string& tag_hash) const {
+  return committed_items_.find(tag_hash) != committed_items_.end();
+}
+
+sync_pb::SyncEntity SingleTypeMockServer::GetLastCommittedEntity(
+    const std::string& tag_hash) const {
+  DCHECK(HasCommitEntity(tag_hash));
+  return committed_items_.find(tag_hash)->second;
+}
+
+sync_pb::DataTypeProgressMarker SingleTypeMockServer::GetProgress() const {
+  sync_pb::DataTypeProgressMarker progress;
+  progress.set_data_type_id(type_);
+  progress.set_token("non_null_progress_token");
+  return progress;
+}
+
+sync_pb::DataTypeContext SingleTypeMockServer::GetContext() const {
+  return sync_pb::DataTypeContext();
+}
+
+std::string SingleTypeMockServer::GenerateId(const std::string& tag_hash) {
+  return "FakeId:" + tag_hash;
+}
+
+int64 SingleTypeMockServer::GetServerVersion(
+    const std::string& tag_hash) const {
+  std::map<const std::string, int64>::const_iterator it;
+  it = server_versions_.find(tag_hash);
+  // Server versions do not necessarily start at 1 or 0.
+  if (it == server_versions_.end()) {
+    return 2048;
+  } else {
+    return it->second;
+  }
+}
+
+void SingleTypeMockServer::SetServerVersion(const std::string& tag_hash,
+                                            int64 version) {
+  server_versions_[tag_hash] = version;
+}
+
+}  // namespace syncer
diff --git a/sync/test/engine/single_type_mock_server.h b/sync/test/engine/single_type_mock_server.h
new file mode 100644
index 0000000..39ad35c
--- /dev/null
+++ b/sync/test/engine/single_type_mock_server.h
@@ -0,0 +1,94 @@
+// 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.
+
+#ifndef SYNC_TEST_ENGINE_SINGLE_TYPE_MOCK_SERVER_H_
+#define SYNC_TEST_ENGINE_SINGLE_TYPE_MOCK_SERVER_H_
+
+#include "sync/engine/non_blocking_sync_common.h"
+
+#include "sync/internal_api/public/base/model_type.h"
+
+namespace syncer {
+
+// A mock server used to test of happy-path update and commit logic.
+//
+// This object supports only one ModelType, which must be specified at
+// initialization time.  It does not support GetUpdates messages.  It does not
+// support simulated errors.
+//
+// This class is useful for testing UpdateHandlers and CommitContributors.
+class SingleTypeMockServer {
+ public:
+  explicit SingleTypeMockServer(syncer::ModelType type);
+  ~SingleTypeMockServer();
+
+  // Generates a SyncEntity representing a server-delivered update containing
+  // the root node for this SingleTypeMockServer's type.
+  sync_pb::SyncEntity TypeRootUpdate();
+
+  // Generates a SyncEntity representing a server-delivered update.
+  //
+  // The |version_offset| parameter allows the caller to simulate reflected
+  // updates, redeliveries, and genuine updates.
+  sync_pb::SyncEntity UpdateFromServer(
+      int64 version_offset,
+      const std::string& tag_hash,
+      const sync_pb::EntitySpecifics& specifics);
+
+  // Generates a SyncEntity representing a server-delivered update to delete
+  // an item.
+  sync_pb::SyncEntity TombstoneFromServer(int64 version_offset,
+                                          const std::string& tag_hash);
+
+  // Generates a response to the specified commit message.
+  //
+  // This does not perform any exhausive testing of the sync protocol.  Many of
+  // the request's fields may safely be left blank, and much of the returned
+  // response will be empty, too.
+  //
+  // This is useful mainly for testing objects that implement the
+  // CommitContributor interface.
+  sync_pb::ClientToServerResponse DoSuccessfulCommit(
+      const sync_pb::ClientToServerMessage& message);
+
+  // Getters to return the commit messages sent to the server through
+  // DoSuccessfulCommit().
+  size_t GetNumCommitMessages() const;
+  sync_pb::ClientToServerMessage GetNthCommitMessage(size_t n) const;
+
+  // Getters to return the most recently committed entities for a given
+  // unique_client_tag hash.
+  bool HasCommitEntity(const std::string& tag_hash) const;
+  sync_pb::SyncEntity GetLastCommittedEntity(const std::string& tag_hash) const;
+
+  // Getters that create realistic-looking progress markers and data type
+  // context.
+  sync_pb::DataTypeProgressMarker GetProgress() const;
+  sync_pb::DataTypeContext GetContext() const;
+
+ private:
+  static std::string GenerateId(const std::string& tag_hash);
+
+  // Get and set our emulated server state.
+  int64 GetServerVersion(const std::string& tag_hash) const;
+  void SetServerVersion(const std::string& tag_hash, int64 version);
+
+  const ModelType type_;
+  const std::string type_root_id_;
+
+  // Server version state maps.
+  std::map<const std::string, int64> server_versions_;
+
+  // Log of messages sent to the server.
+  std::vector<sync_pb::ClientToServerMessage> commit_messages_;
+
+  // Map of most recent commits by tag_hash.
+  std::map<const std::string, sync_pb::SyncEntity> committed_items_;
+
+  DISALLOW_COPY_AND_ASSIGN(SingleTypeMockServer);
+};
+
+}  // namespace syncer
+
+#endif  // SYNC_TEST_ENGINE_SINGLE_TYPE_MOCK_SERVER_H_