carry _self_ref_order through Sync API, Queue and Dequeue

Author: ozer550

morango/api/serializers.py

@@ -178,5 +178,6 @@ class Meta:
             "profile",
             "rmcb_list",
             "_self_ref_fk",
+            "_self_ref_order",
         )
         read_only_fields = fields

morango/sync/backends/postgres.py

@@ -204,10 +204,10 @@ def _dequeuing_merge_conflict_rmcb(self, cursor, transfersession_id):
     def _dequeuing_merge_conflict_buffer(self, cursor, current_id, transfersession_id):
         # transfer buffer serialized into conflicting store
         merge_conflict_store = """UPDATE {store} as store SET (serialized, deleted, last_saved_instance, last_saved_counter, hard_deleted, model_name,
-                                                        profile, partition, source_id, conflicting_serialized_data, dirty_bit, _self_ref_fk, deserialization_error, last_transfer_session_id)
+                                                        profile, partition, source_id, conflicting_serialized_data, dirty_bit, _self_ref_fk, _self_ref_order, deserialization_error, last_transfer_session_id)
                                             = (CASE buffer.hard_deleted WHEN TRUE THEN '' ELSE store.serialized END, store.deleted OR buffer.deleted, '{current_instance_id}',
                                                    {current_instance_counter}, store.hard_deleted, store.model_name, store.profile, store.partition, store.source_id,
-                                                   CASE buffer.hard_deleted WHEN TRUE THEN '' ELSE buffer.serialized || '\n' || store.conflicting_serialized_data END, TRUE, store._self_ref_fk,
+                                                   CASE buffer.hard_deleted WHEN TRUE THEN '' ELSE buffer.serialized || '\n' || store.conflicting_serialized_data END, TRUE, store._self_ref_fk, store._self_ref_order,
                                                    '', '{transfer_session_id}')
                                             /*Scope to a single record.*/
                                             FROM {buffer} AS buffer
@@ -278,26 +278,26 @@ def _dequeuing_insert_remaining_buffer(self, cursor, transfersession_id):
             WITH new_values as
             (
                 SELECT buffer.model_uuid, buffer.serialized, buffer.deleted, buffer.last_saved_instance, buffer.last_saved_counter, buffer.hard_deleted,
-                       buffer.model_name, buffer.profile, buffer.partition, buffer.source_id, buffer.conflicting_serialized_data, buffer._self_ref_fk
+                       buffer.model_name, buffer.profile, buffer.partition, buffer.source_id, buffer.conflicting_serialized_data, buffer._self_ref_fk, buffer._self_ref_order
                 FROM {buffer} as buffer
                 WHERE buffer.transfer_session_id = '{transfer_session_id}'
             ),
             updated as
             (
                 UPDATE {store} store SET (serialized, deleted, last_saved_instance, last_saved_counter, hard_deleted, model_name, profile,
-                                     partition, source_id, conflicting_serialized_data, dirty_bit, _self_ref_fk, deserialization_error, last_transfer_session_id)
+                                     partition, source_id, conflicting_serialized_data, dirty_bit, _self_ref_fk, _self_ref_order, deserialization_error, last_transfer_session_id)
                                     = (nv.serialized, nv.deleted, nv.last_saved_instance, nv.last_saved_counter, nv.hard_deleted,
                                        nv.model_name, nv.profile, nv.partition, nv.source_id, nv.conflicting_serialized_data, TRUE,
-                                       nv._self_ref_fk, '', '{transfer_session_id}')
+                                       nv._self_ref_fk, nv._self_ref_order, '', '{transfer_session_id}')
                 FROM new_values nv
                 WHERE nv.model_uuid = store.id
                 returning store.*
             )
             INSERT INTO {store}(id, serialized, deleted, last_saved_instance, last_saved_counter, hard_deleted, model_name, profile,
-                                partition, source_id, conflicting_serialized_data, dirty_bit, _self_ref_fk, deserialization_error, last_transfer_session_id)
+                                partition, source_id, conflicting_serialized_data, dirty_bit, _self_ref_fk, _self_ref_order, deserialization_error, last_transfer_session_id)
             SELECT ut.model_uuid, ut.serialized, ut.deleted, ut.last_saved_instance, ut.last_saved_counter, ut.hard_deleted,
                        ut.model_name, ut.profile, ut.partition, ut.source_id, ut.conflicting_serialized_data, TRUE,
-                       ut._self_ref_fk, '', '{transfer_session_id}'
+                       ut._self_ref_fk, ut._self_ref_order, '', '{transfer_session_id}'
             FROM new_values ut
             WHERE ut.model_uuid not in (SELECT id FROM updated)
         """.format(

morango/sync/backends/sqlite.py

@@ -137,10 +137,10 @@ def _dequeuing_merge_conflict_rmcb(self, cursor, transfersession_id):
     def _dequeuing_merge_conflict_buffer(self, cursor, current_id, transfersession_id):
         # transfer buffer serialized into conflicting store
         merge_conflict_store = """REPLACE INTO {store} (id, serialized, deleted, last_saved_instance, last_saved_counter, hard_deleted, model_name, profile, partition,
-                                                        source_id, conflicting_serialized_data, dirty_bit, _self_ref_fk, deserialization_error, last_transfer_session_id)
+                                                        source_id, conflicting_serialized_data, dirty_bit, _self_ref_fk, _self_ref_order, deserialization_error, last_transfer_session_id)
                                             SELECT store.id, CASE buffer.hard_deleted WHEN 1 THEN '' ELSE store.serialized END, store.deleted OR buffer.deleted, '{current_instance_id}',
                                                    {current_instance_counter}, store.hard_deleted OR buffer.hard_deleted, store.model_name, store.profile, store.partition, store.source_id,
-                                                   CASE buffer.hard_deleted WHEN 1 THEN '' ELSE buffer.serialized || '\n' || store.conflicting_serialized_data END, 1, store._self_ref_fk,
+                                                   CASE buffer.hard_deleted WHEN 1 THEN '' ELSE buffer.serialized || '\n' || store.conflicting_serialized_data END, 1, store._self_ref_fk, store._self_ref_order,
                                                    '', '{transfer_session_id}'
                                             FROM {buffer} AS buffer, {store} AS store
                                             /*Scope to a single record.*/
@@ -191,10 +191,10 @@ def _dequeuing_update_rmcs_last_saved_by(
     def _dequeuing_insert_remaining_buffer(self, cursor, transfersession_id):
         # insert remaining records into store
         insert_remaining_buffer = """REPLACE INTO {store} (id, serialized, deleted, last_saved_instance, last_saved_counter, hard_deleted, model_name, profile, partition,
-                                                           source_id, conflicting_serialized_data, dirty_bit, _self_ref_fk, deserialization_error, last_transfer_session_id)
+                                                           source_id, conflicting_serialized_data, dirty_bit, _self_ref_fk, _self_ref_order, deserialization_error, last_transfer_session_id)
                                     SELECT buffer.model_uuid, buffer.serialized, buffer.deleted, buffer.last_saved_instance, buffer.last_saved_counter, buffer.hard_deleted,
                                            buffer.model_name, buffer.profile, buffer.partition, buffer.source_id, buffer.conflicting_serialized_data, 1,
-                                           buffer._self_ref_fk, '', '{transfer_session_id}'
+                                           buffer._self_ref_fk, buffer._self_ref_order, '', '{transfer_session_id}'
                                     FROM {buffer} AS buffer
                                     WHERE buffer.transfer_session_id = '{transfer_session_id}'
                            """.format(

morango/sync/operations.py

@@ -483,7 +483,7 @@ def _queue_into_buffer_v1(transfersession):
                 """SELECT
                        id, serialized, deleted, last_saved_instance, last_saved_counter, hard_deleted, model_name, profile,
                        partition, source_id, conflicting_serialized_data,
-                       CAST ('{transfer_session_id}' AS {transfer_session_id_type}), _self_ref_fk
+                       CAST ('{transfer_session_id}' AS {transfer_session_id_type}), _self_ref_fk, _self_ref_order
                    FROM {store} WHERE {condition}
                 """.format(
                     transfer_session_id=transfersession.id,
@@ -514,7 +514,7 @@ def _queue_into_buffer_v1(transfersession):
                 """INSERT INTO {outgoing_buffer}
                    (model_uuid, serialized, deleted, last_saved_instance, last_saved_counter,
                    hard_deleted, model_name, profile, partition, source_id, conflicting_serialized_data,
-                   transfer_session_id, _self_ref_fk)
+                   transfer_session_id, _self_ref_fk, _self_ref_order)
                    {select}
                 """.format(
                     outgoing_buffer=Buffer._meta.db_table,
@@ -632,7 +632,7 @@ def _queue_into_buffer_v2(transfersession, chunk_size=200):
                 """SELECT
                         id, serialized, deleted, last_saved_instance, last_saved_counter, hard_deleted, model_name, profile,
                         partition, source_id, conflicting_serialized_data,
-                        CAST ('{transfer_session_id}' AS {transfer_session_id_type}), _self_ref_fk
+                        CAST ('{transfer_session_id}' AS {transfer_session_id_type}), _self_ref_fk, _self_ref_order
                     FROM {store} WHERE {condition}
                 """.format(
                     transfer_session_id=transfersession.id,
@@ -661,7 +661,7 @@ def _queue_into_buffer_v2(transfersession, chunk_size=200):
                 """INSERT INTO {outgoing_buffer}
                    (model_uuid, serialized, deleted, last_saved_instance, last_saved_counter,
                    hard_deleted, model_name, profile, partition, source_id, conflicting_serialized_data,
-                   transfer_session_id, _self_ref_fk)
+                   transfer_session_id, _self_ref_fk, _self_ref_order)
                    {select}
                 """.format(
                     outgoing_buffer=Buffer._meta.db_table,

tests/testapp/tests/sync/test_operations.py

@@ -88,6 +88,14 @@ def test_all_fsics(self):
         assertRecordsBuffered(self.data["group1_c2"])
         assertRecordsBuffered(self.data["group2_c1"])
 
+    def test_self_ref_order_propagates_to_buffer(self):
+        Store.objects.update(_self_ref_order=42)
+        fsics = {self.data["group1_id"].id: 1, self.data["group2_id"].id: 1}
+        self.transfer_session.client_fsic = json.dumps(fsics)
+        _queue_into_buffer_v1(self.transfer_session)
+        self.assertTrue(Buffer.objects.exists())
+        self.assertFalse(Buffer.objects.exclude(_self_ref_order=42).exists())
+
     def test_very_many_fsics(self):
         """
         Regression test against 'Expression tree is too large (maximum depth 1000)' error with many fsics
@@ -278,6 +286,18 @@ def setUp(self):
             capabilities=[FSIC_V2_FORMAT],
         )
 
+    def test_self_ref_order_propagates_to_buffer(self):
+        Store.objects.update(_self_ref_order=42)
+        fsics = {
+            "super": {},
+            "sub": {"": {self.data["group1_id"].id: 1, self.data["group2_id"].id: 1}},
+        }
+        self.transfer_session.client_fsic = json.dumps(fsics)
+        self.transfer_session.server_fsic = json.dumps({"super": {}, "sub": {}})
+        _queue_into_buffer_v2(self.transfer_session)
+        self.assertTrue(Buffer.objects.exists())
+        self.assertFalse(Buffer.objects.exclude(_self_ref_order=42).exists())
+
     # @pytest.mark.skip("Takes 30+ seconds, manual run only")
     def test_very_many_instances_in_fsic(self):
         """
@@ -788,6 +808,14 @@ def test_dequeuing_merge_conflict_hard_delete(self):
         self.assertEqual(store.serialized, "")
         self.assertEqual(store.conflicting_serialized_data, "")
 
+    def test_dequeuing_merge_conflict_buffer__self_ref_order_preserved(self):
+        Store.objects.filter(id=self.data["model2"]).update(_self_ref_order=11)
+        Buffer.objects.filter(model_uuid=self.data["model2"]).update(_self_ref_order=99)
+        with connection.cursor() as cursor:
+            current_id = InstanceIDModel.get_current_instance_and_increment_counter()
+            DBBackend._dequeuing_merge_conflict_buffer(cursor, current_id, self.transfer_session.id)
+        self.assertEqual(Store.objects.get(id=self.data["model2"])._self_ref_order, 11)
+
     def test_dequeuing_update_rmcs_last_saved_by(self):
         self.assertFalse(RecordMaxCounter.objects.filter(instance_id=self.current_id.id).exists())
         with connection.cursor() as cursor:
@@ -842,6 +870,13 @@ def test_dequeuing_insert_remaining_buffer(self):
         self.assertEqual(Store.objects.get(id=self.data["model3"]).serialized, "buffer")
         self.assertTrue(Store.objects.filter(id=self.data["model4"]).exists())
 
+    def test_dequeuing_insert_remaining_buffer__self_ref_order_propagates(self):
+        Buffer.objects.filter(model_uuid=self.data["model4"]).update(_self_ref_order=7)
+        self.assertFalse(Store.objects.filter(id=self.data["model4"]).exists())
+        with connection.cursor() as cursor:
+            DBBackend._dequeuing_insert_remaining_buffer(cursor, self.transfer_session.id)
+        self.assertEqual(Store.objects.get(id=self.data["model4"])._self_ref_order, 7)
+
     def test_dequeuing_insert_remaining_rmcb(self):
         for i in self.data["model4_rmcb_ids"]:
             self.assertFalse(

tests/testapp/tests/test_api.py

@@ -986,6 +986,12 @@ def test_buffer_serializer_makes_no_transfer_session_query(self):
             for q in ctx.captured_queries:
                 self.assertFalse("morango_transfersession" in q["sql"])
 
+    def test_buffer_serializer_includes_self_ref_order(self):
+        transfer_session_id = self.create_records_for_pulling()
+        Buffer.objects.filter(transfer_session_id=transfer_session_id).update(_self_ref_order=4)
+        buffer = Buffer.objects.filter(transfer_session_id=transfer_session_id).first()
+        self.assertEqual(BufferSerializer(instance=buffer).data["_self_ref_order"], 4)
+
     def test_pull_valid_buffer_list(self):
 
         transfer_session_id = self.create_records_for_pulling()