Allow repeat push/pull of buffers

Author: bjester

morango/sync/utils.py

@@ -128,18 +128,26 @@ def validate_and_create_buffer_data(  # noqa: C901
         buffer_list += [Buffer(**record)]
 
     with transaction.atomic():
-        transfer_session.records_transferred += len(data)
+        deleted_buffers, _ = Buffer.objects.filter(
+            transfer_session=transfer_session,
+            model_uuid__in=[record["model_uuid"] for record in data]
+        ).delete()
+        RecordMaxCounterBuffer.objects.filter(
+            transfer_session=transfer_session,
+            model_uuid__in=[record["model_uuid"] for record in data]
+        ).delete()
 
         if connection is not None:
             transfer_session.bytes_sent = connection.bytes_sent
         if connection is not None:
             transfer_session.bytes_received = connection.bytes_received
 
-        transfer_session.save()
-
         Buffer.objects.bulk_create(buffer_list)
         RecordMaxCounterBuffer.objects.bulk_create(rmcb_list)
 
+        transfer_session.records_transferred += len(buffer_list) - deleted_buffers
+        transfer_session.save()
+
 
 class SyncSignal(object):
     """

tests/testapp/tests/test_api.py

@@ -897,7 +897,7 @@ def build_buffer_item(self, **kwargs):
 
         return buffermodel
 
-    def make_buffer_post_request(self, buffers, expected_status=201, gzip=False):
+    def make_buffer_post_request(self, buffers, expected_status=201, gzip=False, pre_count=0):
         serialized_recs = BufferSerializer(buffers, many=True)
 
         # extract that data that is to be posted
@@ -911,9 +911,13 @@ def make_buffer_post_request(self, buffers, expected_status=201, gzip=False):
             headers["content_type"] = "application/gzip"
             headers["format"] = None
 
-        # delete the records from the DB so we don't conflict when we POST
-        Buffer.objects.all().delete()
-        RecordMaxCounterBuffer.objects.all().delete()
+        # delete the records to match pre_count if zero
+        if pre_count == 0:
+            Buffer.objects.all().delete()
+            RecordMaxCounterBuffer.objects.all().delete()
+        else:
+            self.assertEqual(Buffer.objects.count(), pre_count)
+            self.assertEqual(RecordMaxCounterBuffer.objects.count(), pre_count * 3)
 
         response = self.client.post(reverse("buffers-list"), data, **headers)
         self.assertEqual(response.status_code, expected_status)
@@ -964,6 +968,22 @@ def test_push_fails_for_pull_transfersession(self):
         rec_3 = self.build_buffer_item(transfer_session=rec_1.transfer_session)
         self.make_buffer_post_request([rec_1, rec_2, rec_3], expected_status=403)
 
+    def test_push_repeat_chunk(self):
+        rec_1 = self.build_buffer_item(push=True, filter=self.default_push_filter)
+        transfer_session = rec_1.transfer_session
+        rec_2 = self.build_buffer_item(transfer_session=transfer_session)
+        # need to create a third buffer so it doesn't mark the transfer complete after pushing 2
+        self.build_buffer_item(transfer_session=transfer_session)
+        self.assertEqual(transfer_session.records_transferred, 0)
+
+        self.make_buffer_post_request([rec_1, rec_2], expected_status=201)
+        transfer_session.refresh_from_db()
+        self.assertEqual(transfer_session.records_transferred, 2)
+
+        self.make_buffer_post_request([rec_1, rec_2], expected_status=201, pre_count=2)
+        transfer_session.refresh_from_db()
+        self.assertEqual(transfer_session.records_transferred, 2)
+
     def create_records_for_pulling(self, count=3, **kwargs):
 
         assert count >= 1
@@ -1025,11 +1045,9 @@ def make_buffer_get_request(
                 transfer_session_id=t_id, model_uuid__in=model_uuids
             ).delete()
 
-            # run the validation logic to ensure no errors were returned
-            errors = validate_and_create_buffer_data(
+            validate_and_create_buffer_data(
                 data, TransferSession.objects.get(id=t_id)
             )
-            self.assertFalse(errors)
 
             # check that the correct number of buffer items were created
             self.assertEqual(
@@ -1061,6 +1079,13 @@ def test_pull_valid_buffer_list(self):
 
         self.make_buffer_get_request(transfer_session_id=transfer_session_id)
 
+    def test_pull_repeat(self):
+
+        transfer_session_id = self.create_records_for_pulling(count=3)
+
+        self.make_buffer_get_request(transfer_session_id=transfer_session_id, expected_count=3)
+        self.make_buffer_get_request(transfer_session_id=transfer_session_id, expected_count=3)
+
     def test_pull_fails_when_transfer_session_id_not_specified(self):
 
         self.create_records_for_pulling()