public void TestFailedBulkGetDoesntChangeLastSequence()
        {
            if (!Boolean.Parse((string)Runtime.Properties["replicationTestsEnabled"]))
            {
                Assert.Inconclusive("Replication tests disabled.");
                return;
            }
                
            var fakeFactory = new MockHttpClientFactory(false);
            fakeFactory.HttpHandler.SetResponder("_bulk_get", (request) =>
            {
                var str = request.Content.ReadAsStringAsync().Result;
                if(str.IndexOf("doc2") != -1) {
                    Log.D(Tag, "Rejecting this bulk get because it looks like the first batch");
                    throw new OperationCanceledException();
                }

                Log.D(Tag, "Letting this bulk get through");
                return new RequestCorrectHttpMessage();
            });
            manager.DefaultHttpClientFactory = fakeFactory;
            Manager.DefaultOptions.MaxRevsToGetInBulk = 10;
            Manager.DefaultOptions.MaxOpenHttpConnections = 8;
            Manager.DefaultOptions.RequestTimeout = TimeSpan.FromSeconds(5);

            CreatePullAndTest((int)(Manager.DefaultOptions.MaxRevsToGetInBulk * 1.5), repl => {
                Log.D(Tag, "Document count increased to {0} with last sequence '{1}'", database.DocumentCount, repl.LastSequence);
                Assert.IsTrue(database.DocumentCount > 0, "Didn't get docs from second bulk get batch");
                Assert.AreEqual("0", repl.LastSequence, "LastSequence was advanced");
            });

            fakeFactory.HttpHandler.ClearResponders();
            var pull = database.CreatePullReplication(GetReplicationURL());
            RunReplication(pull);
            Assert.AreEqual(pull.ChangesCount, pull.CompletedChangesCount);
            Assert.AreNotEqual(pull.LastSequence, "0");
        }

        public void TestHeaders()
        {
            if (!Boolean.Parse((string)GetProperty("replicationTestsEnabled")))
            {
                Assert.Inconclusive("Replication tests disabled.");
                return;
            }

            var mockHttpClientFactory = new MockHttpClientFactory(false);
            manager.DefaultHttpClientFactory = mockHttpClientFactory;

            var mockHttpHandler = mockHttpClientFactory.HttpHandler;

            using (var remoteDb = _sg.CreateDatabase(TempDbName())) {
                var remote = remoteDb.RemoteUri;
                var puller = database.CreatePullReplication(remote);
                var headers = new Dictionary<string, string>();
                headers["foo"] = "bar";

                var pusher = database.CreatePushReplication (remote);
                pusher.Headers = headers;
                pusher.Start ();
                Sleep (5000);
                pusher.Stop ();

                ValidateHttpHeaders (mockHttpHandler);
                mockHttpHandler.ClearCapturedRequests ();

                puller.Headers = headers;
                puller.Start();
                Sleep(5000);
                puller.Stop();

                ValidateHttpHeaders (mockHttpHandler);
            }
        }

        public void TestServerDoesNotSupportMultipart() {
            if (!Boolean.Parse((string)GetProperty("replicationTestsEnabled")))
            {
                Assert.Inconclusive("Replication tests disabled.");
                return;
            }

            Assert.AreEqual(0, database.GetLastSequenceNumber());

            var properties1 = new Dictionary<string, object>() {
                { "dynamic", 1 }
            };

            var doc = CreateDocumentWithProperties(database, properties1);
            var rev1 = doc.CurrentRevision;

            var unsavedRev2 = doc.CreateRevision();
            var attachmentStream = GetAsset("attachment.png");
            unsavedRev2.SetAttachment("attachment.png", "image/png", attachmentStream);
            var rev2 = unsavedRev2.Save();
            attachmentStream.Dispose();
            Assert.IsNotNull(rev2);
            unsavedRev2.Dispose();

            var httpClientFactory = new MockHttpClientFactory();
            var httpHandler = httpClientFactory.HttpHandler; 
            httpHandler.AddResponderFakeLocalDocumentUpdate404();

            var responders = new List<MockHttpRequestHandler.HttpResponseDelegate>();
            responders.Add((request) => 
            {
                var json = "{\"error\":\"Unsupported Media Type\",\"reason\":\"missing\"}";
                return MockHttpRequestHandler.GenerateHttpResponseMessage(HttpStatusCode.UnsupportedMediaType, 
                    "Unsupported Media Type", json);
            });

            responders.Add((request) => 
            {
                var props = new Dictionary<string, object>()
                {
                    {"id", doc.Id},
                    {"ok", true},
                    {"rev", doc.CurrentRevisionId}

                };
                return MockHttpRequestHandler.GenerateHttpResponseMessage(props);
            });

            MockHttpRequestHandler.HttpResponseDelegate chainResponder = (request) =>
            {
                if (responders.Count > 0) {
                    var responder = responders[0];
                    responders.RemoveAt(0);
                    return responder(request);
                }
                return null;
            };

            httpHandler.SetResponder(doc.Id, chainResponder);
            manager.DefaultHttpClientFactory = httpClientFactory;

            using (var remoteDb = _sg.CreateDatabase(TempDbName())) {
                var pusher = database.CreatePushReplication(remoteDb.RemoteUri);
                RunReplication(pusher);

                var count = 0;
                foreach (var request in httpHandler.CapturedRequests) {
                    if (request.Method == HttpMethod.Put && request.RequestUri.PathAndQuery.Contains(doc.Id)) {
                        var isMultipartContent = (request.Content is MultipartContent);
                        if (count == 0) {
                            Assert.IsTrue(isMultipartContent);
                        }
                        else {
                            Assert.IsFalse(isMultipartContent);
                        }
                        count++;
                    }
                }
            }
        }

        public void TestBulkPullPermanentExceptionSurrender() {
            if (!Boolean.Parse((string)GetProperty("replicationTestsEnabled")))
            {
                Assert.Inconclusive("Replication tests disabled.");
                return;
            }

            Log.Domains.Sync.Level = Log.LogLevel.Debug;
            var fakeFactory = new MockHttpClientFactory(false);
            FlowControl flow = new FlowControl(new FlowItem[]
            {
                new ExceptionThrower(new SocketException()) { ExecutionCount = -1 },
            });

            fakeFactory.HttpHandler.SetResponder("_bulk_get", (request) => 
                flow.ExecuteNext<HttpResponseMessage>());
            manager.DefaultHttpClientFactory = fakeFactory;

            using (var remoteDb = _sg.CreateDatabase(TempDbName())) {
                CreatePullAndTest(20, remoteDb, repl => Assert.IsTrue(database.GetDocumentCount() < 20, "Somehow got all the docs"));
            }
        }

        public void TestContinuousReplicationErrorNotification() {
            if (!Boolean.Parse((string)GetProperty("replicationTestsEnabled")))
            {
                Assert.Inconclusive("Replication tests disabled.");
                return;
            }

            var httpClientFactory = new MockHttpClientFactory();
            manager.DefaultHttpClientFactory = httpClientFactory;

            var httpHandler = httpClientFactory.HttpHandler; 
            httpHandler.AddResponderThrowExceptionAllRequests();

            using (var remoteDb = _sg.CreateDatabase(TempDbName())) {
                var pusher = database.CreatePushReplication(remoteDb.RemoteUri);
                pusher.Continuous = true;

                var signal = new CountdownEvent(1);
                var observer = new ReplicationErrorObserver(signal);
                pusher.Changed += observer.Changed;
                pusher.Start();

                CreateDocuments(database, 10);

                var success = signal.Wait(TimeSpan.FromSeconds(30));
                Assert.IsTrue(success);

                StopReplication(pusher);
            }
        }

        private void RunPushReplicationWithTransientError(Int32 errorCode, string statusMessage, Boolean expectError) 
        {
            var properties1 = new Dictionary<string, object>() 
            {
                {"doc1", "testPushReplicationTransientError"}
            };
            CreateDocumentWithProperties(database, properties1);

            var httpClientFactory = new MockHttpClientFactory(false);
            var httpHandler = httpClientFactory.HttpHandler; 
            manager.DefaultHttpClientFactory = httpClientFactory;

            MockHttpRequestHandler.HttpResponseDelegate sentinal = MockHttpRequestHandler.FakeBulkDocs;

            var responders = new List<MockHttpRequestHandler.HttpResponseDelegate>();
            responders.Add(MockHttpRequestHandler.TransientErrorResponder(errorCode, statusMessage));
            MockHttpRequestHandler.HttpResponseDelegate chainResponder = (request) =>
            {
                if (responders.Count > 0) {
                    var responder = responders[0];
                    responders.RemoveAt(0);
                    return responder(request);
                }

                return sentinal(request);
            };

            httpHandler.SetResponder("_bulk_docs", chainResponder);

            // Create a replication observer to wait until replication finishes
            var replicationDoneSignal = new CountdownEvent(1);
            var replicationFinishedObserver = new ReplicationObserver(replicationDoneSignal);
            using (var remoteDb = _sg.CreateDatabase(TempDbName())) {
                var pusher = database.CreatePushReplication(remoteDb.RemoteUri);
                pusher.Changed += replicationFinishedObserver.Changed;

                // save the checkpoint id for later usage
                var checkpointId = pusher.RemoteCheckpointDocID();

                // kick off the replication
                pusher.Start();

                // wait for it to finish
                var success = replicationDoneSignal.Wait(TimeSpan.FromSeconds(30));
                Assert.IsTrue(success);

                if (expectError) {
                    Assert.IsNotNull(pusher.LastError);
                }
                else {
                    Assert.IsNull(pusher.LastError);
                }

                // workaround for the fact that the replicationDoneSignal.Await() call could unblock before all
                // the statements in Replication.Stopped() have even had a chance to execute.
                Sleep(500);

                var localLastSequence = database.LastSequenceWithCheckpointId(checkpointId);
                if (expectError) {
                    Assert.Null(localLastSequence);
                }
                else {
                    Assert.IsNotNull(localLastSequence);
                }
            }
        }

        public void TestAllLeafRevisionsArePushed()
        {
            var httpClientFactory = new MockHttpClientFactory();
            var httpHandler = httpClientFactory.HttpHandler; 
            httpHandler.AddResponderRevDiffsAllMissing();
            httpHandler.AddResponderFakeLocalDocumentUpdate404();
            httpHandler.ResponseDelayMilliseconds = 250;
            manager.DefaultHttpClientFactory = httpClientFactory;

            var doc = database.CreateDocument();
            var rev1a = doc.CreateRevision().Save();
            var rev2a = rev1a.CreateRevision().Save();
            var rev3a = rev2a.CreateRevision().Save();

            // delete the branch we've been using, then create a new one to replace it
            var rev4a = rev3a.DeleteDocument();
            var rev2b = rev1a.CreateRevision().Save(true);

            Assert.AreEqual(rev2b.Id, doc.CurrentRevisionId);

            // sync with remote DB -- should push both leaf revisions
            var pusher = database.CreatePushReplication(GetReplicationURL());
            RunReplication(pusher);
            Assert.IsNull(pusher.LastError);

            var foundRevsDiff = false;
            var capturedRequests = httpHandler.CapturedRequests;
            foreach (var httpRequest in capturedRequests) 
            {
                var uriString = httpRequest.RequestUri.ToString();
                if (uriString.EndsWith("_revs_diff", StringComparison.Ordinal))
                {
                    foundRevsDiff = true;
                    var jsonMap = MockHttpRequestHandler.GetJsonMapFromRequest(httpRequest);
                    var revisionIds = ((JArray)jsonMap.Get(doc.Id)).Values<string>().ToList();
                    Assert.AreEqual(2, revisionIds.Count);
                    Assert.IsTrue(revisionIds.Contains(rev4a.Id));
                    Assert.IsTrue(revisionIds.Contains(rev2b.Id));
                }
            }

            Assert.IsTrue(foundRevsDiff);
        }

        [Test] // This test takes nearly 5 minutes to run, so only run when needed
        #endif
        public void TestLongRemovedChangesFeed()
        {
            var random = new Random();
            var changesFeed = new StringBuilder("{\"results\":[");
            const int limit = 100000;
            HashSet<string> removedIDSet = new HashSet<string>();
            for (var i = 1; i < limit; i++) {
                var removed = random.NextDouble() >= 0.5;
                if (removed) {
                    var removedID = Misc.CreateGUID();
                    changesFeed.AppendFormat("{{\"seq\":\"{0}\",\"id\":\"{1}\",\"removed\":[\"fake\"]," +
                        "\"changes\":[{{\"rev\":\"1-deadbeef\"}}]}},",
                        i, removedID);
                    removedIDSet.Add(removedID);
                } else {
                    changesFeed.AppendFormat("{{\"seq\":\"{0}\",\"id\":\"{1}\",\"changes\":[{{\"rev\":\"1-deadbeef\"}}]}},",
                        i, Misc.CreateGUID());
                }
            }

            changesFeed.AppendFormat("{{\"seq\":\"{0}\",\"id\":\"{1}\",\"changes\":[{{\"rev\":\"1-deadbeef\"}}]}}]," +
                "last_seq: \"{0}\"}}",
                limit, Misc.CreateGUID());

            var factory = new MockHttpClientFactory(false);
            factory.HttpHandler.SetResponder("_changes", req =>
            {
                var response = new HttpResponseMessage(HttpStatusCode.OK);
                var changesString = changesFeed.ToString();
                response.Content = new StringContent(changesString);

                return response;
            });

            factory.HttpHandler.SetResponder("_bulk_get", req =>
            {
                var contentStream = req.Content.ReadAsStreamAsync().Result;
                var content = Manager.GetObjectMapper().ReadValue<IDictionary<string, object>>(contentStream);

                var responseBody = new StringBuilder("--67aac1bcad803590b9a9e1999fc539438b3363fab35a24c17990188b222f\r\n");
                foreach(var obj in content["docs"] as IEnumerable) {
                    var dict = obj.AsDictionary<string, object>();
                    var nonexistent = removedIDSet.Contains(dict.GetCast<string>("id"));
                    if(nonexistent) {
                        return new HttpResponseMessage(HttpStatusCode.InternalServerError); // Just so we can know
                    } else {
                        responseBody.Append("Content-Type: application/json\r\n\r\n");
                        responseBody.AppendFormat("{{\"_id\":\"{0}\",\"_rev\":\"1-deadbeef\",\"foo\":\"bar\"}}\r\n", dict["id"]);
                    }

                    responseBody.Append("--67aac1bcad803590b9a9e1999fc539438b3363fab35a24c17990188b222f\r\n");
                }

                responseBody.Remove(responseBody.Length - 2, 2);
                responseBody.Append("--");

                var retVal = new HttpResponseMessage(HttpStatusCode.OK);
                var responseString = responseBody.ToString();
                retVal.Content = new StringContent(responseString);
                retVal.Content.Headers.Remove("Content-Type");
                retVal.Content.Headers.TryAddWithoutValidation("Content-Type", "multipart/mixed; boundary=\"67aac1bcad803590b9a9e1999fc539438b3363fab35a24c17990188b222f\"");

                return retVal;
            });

            manager.DefaultHttpClientFactory = factory;
            using (var remoteDb = _sg.CreateDatabase(TempDbName())) {
                var puller = database.CreatePullReplication(remoteDb.RemoteUri);
                RunReplication(puller);
                Assert.AreEqual(ReplicationStatus.Stopped, puller.Status);
                Assert.AreNotEqual(limit, puller.ChangesCount);
                Assert.AreNotEqual(limit, puller.CompletedChangesCount);
                Assert.AreEqual(limit.ToString(), puller.LastSequence);
            }

            Sleep(1000);
        }

 public ChangeTrackerTestClient(CountDownLatch stoppedSignal, CountDownLatch changedSignal)
 {
     this.stoppedSignal = stoppedSignal;
     this.changedSignal = changedSignal;
     HttpClientFactory = new MockHttpClientFactory();
 }

        public void TestHeaders()
        {
            var mockHttpClientFactory = new MockHttpClientFactory();
            manager.DefaultHttpClientFactory = mockHttpClientFactory;

            var mockHttpHandler = mockHttpClientFactory.HttpHandler;
            mockHttpHandler.AddResponderThrowExceptionAllRequests();

            Uri remote = GetReplicationURL();
            Replication puller = database.CreatePullReplication(remote);
            var headers = new Dictionary<string, string>();
            headers["foo"] = "bar";
            puller.Headers = headers;
            RunReplication(puller);
            Assert.IsNotNull(puller.LastError);

            var foundFooHeader = false;
            var requests = mockHttpHandler.CapturedRequests;

            foreach (var request in requests)
            {
                var requestHeaders = request.Headers.GetValues("foo");
                foreach (var requestHeader in requestHeaders)
                {
                    foundFooHeader = true;
                    Assert.AreEqual("bar", requestHeader);
                }
            }
            Assert.IsTrue(foundFooHeader);
        }

 public void TestChangeTrackerBackoffInvalidJson() 
 {
     var factory = new MockHttpClientFactory();
     var httpHandler = (MockHttpRequestHandler)factory.HttpHandler;
     httpHandler.AddResponderReturnInvalidChangesFeedJson();
     TestChangeTrackerBackoff(factory);
 }

 public void TestChangeTrackerBackoffExceptions()
 {
     var factory = new MockHttpClientFactory();
     var httpHandler = (MockHttpRequestHandler)factory.HttpHandler;
     httpHandler.AddResponderThrowExceptionAllRequests();
     TestChangeTrackerBackoff(factory);
 }

        private void TestChangeTrackerBackoff(MockHttpClientFactory httpClientFactory)
        {
            var changeTrackerFinishedSignal = new CountDownLatch(1);
            var client = new ChangeTrackerTestClient(changeTrackerFinishedSignal, null);
            client.HttpClientFactory = httpClientFactory;

            var testUrl = GetReplicationURL();
            var scheduler = new SingleTaskThreadpoolScheduler();
            var changeTracker = new ChangeTracker(testUrl, ChangeTrackerMode.LongPoll, 0, false, client, new TaskFactory(scheduler));

            changeTracker.UsePost = IsSyncGateway(testUrl);
            changeTracker.Start();

            // sleep for a few seconds
            Thread.Sleep(15 * 1000);

            // make sure we got less than 10 requests in those 10 seconds (if it was hammering, we'd get a lot more)
            var handler = client.HttpRequestHandler;
            Assert.IsTrue(handler.CapturedRequests.Count < 25);
            Assert.IsTrue(changeTracker.backoff.NumAttempts > 0, "Observed attempts: {0}".Fmt(changeTracker.backoff.NumAttempts));

            handler.ClearResponders();
            handler.AddResponderReturnEmptyChangesFeed();

            // at this point, the change tracker backoff should cause it to sleep for about 3 seconds
            // and so lets wait 3 seconds until it wakes up and starts getting valid responses
            Thread.Sleep(3 * 1000);

            // now find the delta in requests received in a 2s period
            int before = handler.CapturedRequests.Count;
            Thread.Sleep(2 * 1000);
            int after = handler.CapturedRequests.Count;

            // assert that the delta is high, because at this point the change tracker should
            // be hammering away
            Assert.IsTrue((after - before) > 25);

            // the backoff numAttempts should have been reset to 0
            Assert.IsTrue(changeTracker.backoff.NumAttempts == 0);

            changeTracker.Stop();

            var success = changeTrackerFinishedSignal.Await(TimeSpan.FromSeconds(30));
            Assert.IsTrue(success);
        }

        public void TestAllLeafRevisionsArePushed()
        {
            if (!Boolean.Parse((string)GetProperty("replicationTestsEnabled")))
            {
                Assert.Inconclusive("Replication tests disabled.");
                return;
            }

            var httpClientFactory = new MockHttpClientFactory();
            var httpHandler = httpClientFactory.HttpHandler; 
            httpHandler.AddResponderRevDiffsAllMissing();
            httpHandler.AddResponderFakeLocalDocumentUpdate404();
            httpHandler.ResponseDelayMilliseconds = 250;
            manager.DefaultHttpClientFactory = httpClientFactory;

            var doc = database.CreateDocument();
            var rev1a = doc.CreateRevision().Save();
            var rev2a = rev1a.CreateRevision().Save();
            var rev3a = rev2a.CreateRevision().Save();

            // delete the branch we've been using, then create a new one to replace it
            var rev4a = rev3a.DeleteDocument();
            var unsaved = rev1a.CreateRevision();
            unsaved.SetUserProperties(new Dictionary<string, object> { { "foo", "bar" } });
            var rev2b = unsaved.Save(true);

            Assert.AreEqual(rev2b.Id, doc.CurrentRevisionId);

            using (var remoteDb = _sg.CreateDatabase(TempDbName())) {
                // sync with remote DB -- should push both leaf revisions
                var pusher = database.CreatePushReplication(remoteDb.RemoteUri);
                RunReplication(pusher);
                Assert.IsNull(pusher.LastError);

                var foundRevsDiff = false;
                var capturedRequests = httpHandler.CapturedRequests;
                foreach (var httpRequest in capturedRequests) {
                    var uriString = httpRequest.RequestUri.ToString();
                    if (uriString.EndsWith("_revs_diff", StringComparison.Ordinal)) {
                        foundRevsDiff = true;
                        var jsonMap = MockHttpRequestHandler.GetJsonMapFromRequest(httpRequest);
                        var revisionIds = ((JArray)jsonMap.Get(doc.Id)).Values<string>().ToList();
                        Assert.AreEqual(2, revisionIds.Count);
                        Assert.IsTrue(revisionIds.Contains(rev4a.Id));
                        Assert.IsTrue(revisionIds.Contains(rev2b.Id));
                    }
                }

                Assert.IsTrue(foundRevsDiff);
            }
        }

        public void TestContinuousReplicationErrorNotification() {
            if (!Boolean.Parse((string)Runtime.Properties["replicationTestsEnabled"]))
            {
                Assert.Inconclusive("Replication tests disabled.");
                return;
            }

            var httpClientFactory = new MockHttpClientFactory();
            manager.DefaultHttpClientFactory = httpClientFactory;

            var httpHandler = httpClientFactory.HttpHandler; 
            httpHandler.AddResponderThrowExceptionAllRequests();

            var pusher = database.CreatePushReplication(GetReplicationURL());
            pusher.Continuous = true;

            var signal = new CountdownEvent(1);
            var observer = new ReplicationErrorObserver(signal);
            pusher.Changed += observer.Changed;
            pusher.Start();

            var success = signal.Wait(TimeSpan.FromSeconds(30));
            Assert.IsTrue(success);

            pusher.Stop();
        }

        // Note that this should not happen anymore but this test will remain just to verify
        // the correct behavior if it does
        public void TestBulkGet404()
        {
            var factory = new MockHttpClientFactory(false);
            factory.HttpHandler.SetResponder("_changes", req =>
            {
                var response = new HttpResponseMessage(HttpStatusCode.OK);
                response.Content = new StringContent(@"{""results"":[{""seq"":3,""id"":""somedoc"",""changes"":
                [{""rev"":""2-cafebabe""}]},{""seq"":4,""id"":""otherdoc"",""changes"":[{""rev"":""5-bedbedbe""}]},
                {""seq"":5,""id"":""realdoc"",""changes"":[{""rev"":""1-12345abc""}]}]}");

                return response;
            });

            factory.HttpHandler.SetResponder("_bulk_get", req =>
            {
                var response = new HttpResponseMessage(HttpStatusCode.OK);
                response.Content = new StringContent("--67aac1bcad803590b9a9e1999fc539438b3363fab35a24c17990188b222f\r\n" +
                    "Content-Type: application/json; error=\"true\"\r\n\r\n" +
                    "{\"error\":\"not_found\",\"id\":\"somedoc\",\"reason\":\"missing\",\"rev\":\"2-cafebabe\",\"status\":404}\r\n" +
                    "--67aac1bcad803590b9a9e1999fc539438b3363fab35a24c17990188b222f\r\n" +
                    "Content-Type: application/json; error=\"true\"\r\n\r\n" +
                    "{\"error\":\"not_found\",\"id\":\"otherdoc\",\"reason\":\"missing\",\"rev\":\"5-bedbedbe\",\"status\":404}\r\n" +
                    "--67aac1bcad803590b9a9e1999fc539438b3363fab35a24c17990188b222f\r\n" +
                    "Content-Type: application/json\r\n\r\n" +
                    "{\"_id\":\"realdoc\",\"_rev\":\"1-12345abc\",\"channels\":[\"unit_test\"],\"foo\":\"bar\"}\r\n" +
                    "--67aac1bcad803590b9a9e1999fc539438b3363fab35a24c17990188b222f--");

                response.Content.Headers.Remove("Content-Type");
                response.Content.Headers.TryAddWithoutValidation("Content-Type", "multipart/mixed; boundary=\"67aac1bcad803590b9a9e1999fc539438b3363fab35a24c17990188b222f\"");
                return response;
            });
            manager.DefaultHttpClientFactory = factory;

            using (var remoteDb = _sg.CreateDatabase(TempDbName())) {
                var puller = database.CreatePullReplication(remoteDb.RemoteUri);
                RunReplication(puller);
                Assert.IsNotNull(puller.LastError);
                Assert.AreEqual(3, puller.ChangesCount);
                Assert.AreEqual(3, puller.CompletedChangesCount);
                Assert.AreEqual("5", puller.LastSequence);
            }
        }

        public void TestHeaders()
        {
            if (!Boolean.Parse((string)Runtime.Properties["replicationTestsEnabled"]))
            {
                Assert.Inconclusive("Replication tests disabled.");
                return;
            }

            var mockHttpClientFactory = new MockHttpClientFactory();
            manager.DefaultHttpClientFactory = mockHttpClientFactory;

            var mockHttpHandler = mockHttpClientFactory.HttpHandler;
            mockHttpHandler.AddResponderThrowExceptionAllRequests();

            using (var remoteDb = _sg.CreateDatabase(TempDbName())) {
                var remote = remoteDb.RemoteUri;
                var puller = database.CreatePullReplication(remote);
                var headers = new Dictionary<string, string>();
                headers["foo"] = "bar";
                puller.Headers = headers;
                RunReplication(puller);
                Assert.IsNotNull(puller.LastError);

                var foundFooHeader = false;
                var requests = mockHttpHandler.CapturedRequests;

                foreach (var request in requests) {
                    var requestHeaders = request.Headers.GetValues("foo");
                    foreach (var requestHeader in requestHeaders) {
                        foundFooHeader = true;
                        Assert.AreEqual("bar", requestHeader);
                    }
                }
                Assert.IsTrue(foundFooHeader);
            }
        }

        public void TestContinuousPushReplicationGoesIdle() 
        {
            if (!Boolean.Parse((string)GetProperty("replicationTestsEnabled")))
            {
                Assert.Inconclusive("Replication tests disabled.");
                return;
            }

            // make sure we are starting empty
            Assert.AreEqual(0, database.GetLastSequenceNumber());

            // add docs
            var properties1 = new Dictionary<string, object>();
            properties1["doc1"] = "testContinuousPushReplicationGoesIdle";
            CreateDocumentWithProperties(database, properties1);

            var httpClientFactory = new MockHttpClientFactory();
            manager.DefaultHttpClientFactory = httpClientFactory;

            using (var remoteDb = _sg.CreateDatabase(TempDbName())) {
                var firstPusher = database.CreatePushReplication(remoteDb.RemoteUri);
                firstPusher.Continuous = true;
                var checkpointId = firstPusher.RemoteCheckpointDocID();

                var httpHandler = httpClientFactory.HttpHandler; 
                MockHttpRequestHandler.HttpResponseDelegate localResponder = (request) =>
                {
                    var id = String.Format("_local/{0}", checkpointId);
                    var json = "{" + String.Format("\"id\":\"{0}\",\"ok\":true,\"rev\":\"0-2\"", id) + "}";
                    return MockHttpRequestHandler.GenerateHttpResponseMessage(HttpStatusCode.Created, "OK", json);
                };
                httpHandler.SetResponder("_local", localResponder);

                var replicationIdleSignal = new CountdownEvent(1);
                var replicationIdleObserver = new ReplicationIdleObserver(replicationIdleSignal);
                firstPusher.Changed += replicationIdleObserver.Changed;
                firstPusher.Start();

                var success = replicationIdleSignal.Wait(TimeSpan.FromSeconds(30));
                Assert.IsTrue(success);
                StopReplication(firstPusher);

                // the last sequence should be "1" at this point.  we will use this later
                var lastSequence = database.LastSequenceWithCheckpointId(checkpointId);
                Assert.AreEqual("1", lastSequence);

                // start a second continuous replication
                var secondPusher = database.CreatePushReplication(remoteDb.RemoteUri);
                secondPusher.Continuous = true;
                var secondPusherCheckpointId = secondPusher.RemoteCheckpointDocID();
                Assert.AreEqual(checkpointId, secondPusherCheckpointId);

                // when this goes to fetch the checkpoint, return the last sequence from the previous replication
                localResponder = (request) =>
                {
                    var id = String.Format("_local/{0}", secondPusherCheckpointId);
                    var json = String.Format("{{\"id\":\"{0}\",\"ok\":true,\"rev\":\"0-2\",\"lastSequence\":\"{1}\"}}", id, lastSequence);
                    return MockHttpRequestHandler.GenerateHttpResponseMessage(HttpStatusCode.Created, "OK", json);
                };
                httpHandler.SetResponder("_local", localResponder);

                // start second replication
                replicationIdleSignal = new CountdownEvent(1);
                replicationIdleObserver = new ReplicationIdleObserver(replicationIdleSignal);
                secondPusher.Changed += replicationIdleObserver.Changed;
                secondPusher.Start();

                // wait until we get an IDLE event
                success = replicationIdleSignal.Wait(TimeSpan.FromSeconds(30));
                Assert.IsTrue(success);
                StopReplication(secondPusher);
            }
        }

        public void TestPushReplicationCanMissDocs()
        {
            Assert.Inconclusive("Not sure this is a valid test.");
            if (!Boolean.Parse((string)Runtime.Properties["replicationTestsEnabled"]))
            {
                Assert.Inconclusive("Replication tests disabled.");
                return;
            }
            Assert.AreEqual(0, database.GetLastSequenceNumber());

            var properties1 = new Dictionary<string, object>();
            properties1["doc1"] = "testPushReplicationCanMissDocs";
            CreateDocumentWithProperties(database, properties1);

            var properties2 = new Dictionary<string, object>();
            properties2["doc2"] = "testPushReplicationCanMissDocs";
            var doc2 = CreateDocumentWithProperties(database, properties2);

            var doc2UnsavedRev = doc2.CreateRevision();
            var attachmentStream = GetAsset("attachment.png");
            doc2UnsavedRev.SetAttachment("attachment.png", "image/png", attachmentStream);
            var doc2Rev = doc2UnsavedRev.Save();
            Assert.IsNotNull(doc2Rev);

            var httpClientFactory = new MockHttpClientFactory();
            manager.DefaultHttpClientFactory = httpClientFactory;

            var httpHandler = httpClientFactory.HttpHandler; 
            httpHandler.AddResponderFakeLocalDocumentUpdate404();

            var json = "{\"error\":\"not_found\",\"reason\":\"missing\"}";
            MockHttpRequestHandler.HttpResponseDelegate bulkDocsResponder = (request) =>
            {
                return MockHttpRequestHandler.GenerateHttpResponseMessage(HttpStatusCode.NotFound, null, json);
            };
            httpHandler.SetResponder("_bulk_docs", bulkDocsResponder);

            MockHttpRequestHandler.HttpResponseDelegate doc2Responder = (request) =>
            {
                var responseObject = new Dictionary<string, object>();
                responseObject["id"] = doc2.Id;
                responseObject["ok"] = true;
                responseObject[&q