Add new source for deserialization of store records

Author: bjester

morango/sync/stream/deserialize.py

@@ -0,0 +1,79 @@
+from typing import Dict
+from typing import Generator
+from typing import List
+from typing import Optional
+from typing import Type
+
+from morango.models.certificates import Filter
+from morango.models.core import Store
+from morango.models.core import SyncableModel
+from morango.registry import syncable_models
+from morango.sync.stream.source import MorangoSource
+from morango.sync.stream.source import SourceTask
+
+
+class DeserializeTask(SourceTask):
+    """Carrier class for providing context through the deserialization pipeline."""
+
+    __slots__ = ("store", "app_model", "fk_cache", "errors")
+
+    def __init__(self, store: Store):
+        self.store = store
+        self.app_model: Optional[SyncableModel] = None
+        self.fk_cache: Dict = {}
+        self.errors: List[Exception] = []
+
+    @property
+    def id(self) -> str:
+        return self.store.id
+
+    @property
+    def model(self) -> Type[SyncableModel]:
+        return syncable_models.get_model(self.store.profile, self.store.model_name)
+
+    @property
+    def has_errors(self) -> bool:
+        return len(self.errors) > 0
+
+    def set_app_model(self, app_model: Optional[SyncableModel]) -> None:
+        self.app_model = app_model
+
+    def add_error(self, error: Exception) -> None:
+        self.errors.append(error)
+
+
+class StoreModelSource(MorangoSource[DeserializeTask]):
+    """
+    Yields ``DeserializeTask`` objects for dirty store models that match the optional
+    *sync_filter*.
+    """
+
+    def __init__(
+        self,
+        profile: str,
+        sync_filter: Optional[Filter] = None,
+        dirty_only: bool = True,
+        partition_order: str = "asc",
+        skip_errored: bool = True,
+    ):
+        """
+        :param profile: The Morango model profile
+        :param sync_filter: The Filter object for this sync
+        :param dirty_only: Whether to filter on dirty records only
+        :param partition_order: Controls how the filter specificity is applied, "asc" or "desc"
+        """
+        super().__init__(profile, sync_filter, dirty_only, partition_order)
+        self.skip_errored = skip_errored
+
+    def stream_for_filter(
+        self, partition_condition: Optional[str]
+    ) -> Generator[DeserializeTask, None, None]:
+        qs = Store.objects.filter(profile=self.profile)
+        if partition_condition is not None:
+            qs = qs.filter(partition__startswith=partition_condition)
+        if self.dirty_only:
+            qs = qs.filter(dirty_bit=True)
+        if self.skip_errored:
+            qs = qs.filter(deserialization_error="")
+        for store_model in qs.iterator():
+            yield DeserializeTask(store_model)

tests/testapp/tests/sync/stream/test_deserialize.py

@@ -0,0 +1,132 @@
+import mock
+from django.test import SimpleTestCase
+
+from morango.models.certificates import Filter
+from morango.models.core import Store
+from morango.models.core import SyncableModel
+from morango.sync.stream.deserialize import DeserializeTask
+from morango.sync.stream.deserialize import StoreModelSource
+
+
+class DeserializeTaskTestCase(SimpleTestCase):
+    def setUp(self):
+        self.store = mock.Mock(spec_set=Store)
+        self.store.profile = "test"
+        self.store.model_name = "testmodel"
+        self.task = DeserializeTask(self.store)
+
+    @mock.patch("morango.sync.stream.deserialize.syncable_models.get_model")
+    def test_model(self, mock_get_model):
+        model = mock.Mock(spec_set=SyncableModel)
+        mock_get_model.return_value = model
+
+        self.assertEqual(self.task.model, model)
+        mock_get_model.assert_called_once_with("test", "testmodel")
+
+    def test_has_errors(self):
+        self.assertFalse(self.task.has_errors)
+        self.task.add_error(ValueError("bad data"))
+        self.assertTrue(self.task.has_errors)
+
+    def test_set_app_model(self):
+        app_model = mock.Mock(spec_set=SyncableModel)
+        self.task.set_app_model(app_model)
+        self.assertEqual(self.task.app_model, app_model)
+
+
+class StoreModelSourceTestCase(SimpleTestCase):
+    def test_prefix_conditions__none(self):
+        source = StoreModelSource(profile="test")
+        conditions = list(source.prefix_conditions())
+        self.assertEqual(conditions, [None])
+
+    def test_prefix_conditions__with_filter_asc(self):
+        sync_filter = Filter("b\na")
+        source = StoreModelSource(
+            profile="test", sync_filter=sync_filter, partition_order="asc"
+        )
+        conditions = list(source.prefix_conditions())
+        self.assertEqual(
+            conditions,
+            ["a", "b"],
+        )
+
+    def test_prefix_conditions__with_filter_desc(self):
+        sync_filter = Filter("a\nb")
+        source = StoreModelSource(
+            profile="test", sync_filter=sync_filter, partition_order="desc"
+        )
+        conditions = list(source.prefix_conditions())
+        self.assertEqual(
+            conditions,
+            ["b", "a"],
+        )
+
+    @mock.patch("morango.sync.stream.deserialize.Store.objects.filter")
+    def test_stream__no_partition(self, mock_store_filter):
+        qs = mock.Mock()
+        mock_store_filter.return_value = qs
+        qs.filter.return_value = qs
+        store = mock.Mock(spec_set=Store)
+        store.id = "123"
+        qs.iterator.return_value = [store]
+
+        source = StoreModelSource(profile="test")
+        tasks = list(source.stream())
+
+        self.assertEqual(len(tasks), 1)
+        self.assertEqual(tasks[0].store, store)
+        mock_store_filter.assert_called_once_with(profile="test")
+        self.assertEqual(
+            qs.filter.call_args_list,
+            [mock.call(dirty_bit=True), mock.call(deserialization_error="")],
+        )
+
+    @mock.patch("morango.sync.stream.deserialize.Store.objects.filter")
+    def test_stream__partition_order_and_seen_once(self, mock_store_filter):
+        qs_a = mock.Mock()
+        qs_ab = mock.Mock()
+        qs_a.filter.return_value = qs_a
+        qs_ab.filter.return_value = qs_ab
+
+        first = mock.Mock(spec_set=Store)
+        first.id = "1"
+        second = mock.Mock(spec_set=Store)
+        second.id = "2"
+        third = mock.Mock(spec_set=Store)
+        third.id = "3"
+        qs_a.iterator.return_value = [first, second]
+        qs_ab.iterator.return_value = [second, third]
+
+        mock_store_filter.side_effect = [qs_a, qs_ab]
+
+        source = StoreModelSource(profile="test", sync_filter=Filter("a\na/b"))
+        tasks = list(source.stream())
+
+        self.assertEqual([task.store.id for task in tasks], ["1", "2", "3"])
+        self.assertEqual(
+            mock_store_filter.call_args_list,
+            [mock.call(profile="test"), mock.call(profile="test")],
+        )
+        self.assertEqual(
+            qs_a.filter.call_args_list[0],
+            mock.call(partition__startswith="a"),
+        )
+        self.assertEqual(
+            qs_ab.filter.call_args_list[0],
+            mock.call(partition__startswith="a/b"),
+        )
+
+    @mock.patch("morango.sync.stream.deserialize.Store.objects.filter")
+    def test_stream__no_dirty_filter(self, mock_store_filter):
+        qs = mock.Mock()
+        mock_store_filter.return_value = qs
+        qs.filter.return_value = qs
+        qs.iterator.return_value = []
+
+        source = StoreModelSource(profile="test", dirty_only=False)
+        list(source.stream())
+
+        self.assertEqual(
+            qs.filter.call_args_list, [mock.call(deserialization_error="")]
+        )