From 93636aaceb45dd88bfc9d014d65429f385d259f6 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Thu, 8 Oct 2026 13:32:15 -0700 Subject: [PATCH 01/13] Add outcome resolution to deserialization task Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 42 +++++++++++++- .../tests/sync/stream/test_deserialize.py | 56 +++++++++++++++++++ 2 files changed, 97 insertions(+), 1 deletion(-) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index 38121987..2f423397 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -12,16 +12,30 @@ from morango.sync.stream.source import SourceTask +class DeserializeOutcome: + """The resolved action for a `DeserializeTask`, which the sink applies""" + + SAVE = "save" + DELETE = "delete" + PROPAGATE_DELETE = "propagate_delete" + PROPAGATE_HARD_DELETE = "propagate_hard_delete" + ERROR = "error" + + class DeserializeTask(SourceTask): """Carrier class for providing context through the deserialization pipeline.""" - __slots__ = ("store", "app_model", "fk_cache", "errors") + __slots__ = ("store", "app_model", "fk_cache", "errors", "_propagated_deletion") + + HARD_DELETION = "hard" + SOFT_DELETION = "soft" def __init__(self, store: Store, fk_cache: Dict): self.store = store self.fk_cache: Dict = fk_cache self.app_model: Optional[SyncableModel] = None self.errors: List[Exception] = [] + self._propagated_deletion: Optional[str] = None @property def id(self) -> str: @@ -35,9 +49,35 @@ def model(self) -> Type[SyncableModel]: def has_errors(self) -> bool: return len(self.errors) > 0 + @property + def outcome(self) -> str: + """ + The resolved `DeserializeOutcome` for this task, in order of precedence + :raises AssertionError: if the pipeline has not resolved an outcome + """ + if self.has_errors: + return DeserializeOutcome.ERROR + if self.store.deleted: + return DeserializeOutcome.DELETE + if self._propagated_deletion == self.HARD_DELETION: + return DeserializeOutcome.PROPAGATE_HARD_DELETE + if self._propagated_deletion == self.SOFT_DELETION: + return DeserializeOutcome.PROPAGATE_DELETE + if self.app_model is not None: + return DeserializeOutcome.SAVE + raise AssertionError("DeserializeTask has no resolved outcome") + + def self_referential_fk(self) -> Optional[str]: + """Return the attname of the self-referential FK on the task's model, or ``None``.""" + return syncable_models.get_self_referential_fk(self.model) + def set_app_model(self, app_model: Optional[SyncableModel]) -> None: self.app_model = app_model + def set_propagated_deletion(self, hard: bool) -> None: + """Mark this task's record as deleted because a record it references was deleted""" + self._propagated_deletion = self.HARD_DELETION if hard else self.SOFT_DELETION + def add_error(self, error: Exception) -> None: self.errors.append(error) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 41f7c870..5c75fe23 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -7,6 +7,7 @@ from morango.models.certificates import Filter from morango.models.core import Store from morango.models.core import SyncableModel +from morango.sync.stream.deserialize import DeserializeOutcome from morango.sync.stream.deserialize import DeserializeTask from morango.sync.stream.deserialize import StoreModelSource @@ -18,6 +19,7 @@ def setUp(self): self.store = mock.Mock(spec_set=Store) self.store.profile = "test" self.store.model_name = "testmodel" + self.store.deleted = False self.task = DeserializeTask(self.store, {}) @mock.patch("morango.sync.stream.deserialize.syncable_models.get_model") @@ -38,6 +40,60 @@ def test_set_app_model(self): self.task.set_app_model(app_model) self.assertEqual(self.task.app_model, app_model) + @mock.patch("morango.sync.stream.deserialize.syncable_models.get_self_referential_fk") + @mock.patch("morango.sync.stream.deserialize.syncable_models.get_model") + def test_self_referential_fk(self, mock_get_model, mock_get_self_referential_fk): + model = mock.Mock(spec_set=SyncableModel) + mock_get_model.return_value = model + mock_get_self_referential_fk.return_value = "parent_id" + + self.assertEqual(self.task.self_referential_fk(), "parent_id") + mock_get_self_referential_fk.assert_called_once_with(model) + + def test_outcome__save(self): + self.task.set_app_model(mock.Mock(spec_set=SyncableModel)) + self.assertEqual(self.task.outcome, DeserializeOutcome.SAVE) + + def test_outcome__delete(self): + self.store.deleted = True + self.assertEqual(self.task.outcome, DeserializeOutcome.DELETE) + + def test_outcome__propagate_delete(self): + self.task.set_propagated_deletion(hard=False) + self.assertEqual(self.task.outcome, DeserializeOutcome.PROPAGATE_DELETE) + + def test_outcome__propagate_hard_delete(self): + self.task.set_propagated_deletion(hard=True) + self.assertEqual(self.task.outcome, DeserializeOutcome.PROPAGATE_HARD_DELETE) + + def test_outcome__error(self): + self.task.add_error(ValueError("bad data")) + self.assertEqual(self.task.outcome, DeserializeOutcome.ERROR) + + def test_outcome__error_precedes_delete(self): + self.store.deleted = True + self.task.add_error(ValueError("bad data")) + self.assertEqual(self.task.outcome, DeserializeOutcome.ERROR) + + def test_outcome__error_precedes_propagated_deletion(self): + self.task.set_propagated_deletion(hard=True) + self.task.add_error(ValueError("bad data")) + self.assertEqual(self.task.outcome, DeserializeOutcome.ERROR) + + def test_outcome__delete_precedes_propagated_deletion(self): + self.store.deleted = True + self.task.set_propagated_deletion(hard=True) + self.assertEqual(self.task.outcome, DeserializeOutcome.DELETE) + + def test_outcome__propagated_deletion_precedes_save(self): + self.task.set_app_model(mock.Mock(spec_set=SyncableModel)) + self.task.set_propagated_deletion(hard=False) + self.assertEqual(self.task.outcome, DeserializeOutcome.PROPAGATE_DELETE) + + def test_outcome__unresolved(self): + with self.assertRaises(AssertionError): + _ = self.task.outcome + class StoreModelSourcePrefixConditionsTestCase(SimpleTestCase): """`prefix_conditions` is pure ordering logic over the sync filter and touches no ORM""" From 12b3880007f09f936a0f1ca3e64893e0db52faab Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Thu, 8 Oct 2026 13:32:16 -0700 Subject: [PATCH 02/13] Add flag to toggle parent dereference in test Facility validation Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/testapp/facility_profile/models.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/tests/testapp/facility_profile/models.py b/tests/testapp/facility_profile/models.py index 3bf3cbef..37457471 100644 --- a/tests/testapp/facility_profile/models.py +++ b/tests/testapp/facility_profile/models.py @@ -24,6 +24,10 @@ class SyncableUserModelManager(SyncableModelManager, UserManager): class Facility(FacilityDataSyncableModel): morango_model_name = "facility" + # when True, clean_fields loads the parent object, which raises ObjectDoesNotExist if the + # parent row is not in the app table; used to test non-validation error handling + clean_dereferences_parent = True + name = models.CharField(max_length=100) now_date = models.DateTimeField(default=timezone.now) parent = models.ForeignKey( @@ -45,8 +49,9 @@ def calculate_partition(self, *args, **kwargs): return "{id}".format(id=self.ID_PLACEHOLDER) def clean_fields(self, *args, **kwargs): - # reference parent here just to trigger a non-validation error to make sure we handle it - _ = self.parent + if self.clean_dereferences_parent: + # reference parent here just to trigger a non-validation error to make sure we handle it + _ = self.parent super(Facility, self).clean_fields(*args, **kwargs) From b608f2059b18ad9925fd4904cc761e1dde27d667 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Thu, 8 Oct 2026 13:41:34 -0700 Subject: [PATCH 03/13] Add transforms for deserializing and validating store records Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 228 ++++++++ .../tests/sync/stream/test_deserialize.py | 485 ++++++++++++++++++ 2 files changed, 713 insertions(+) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index 2f423397..c8414408 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -1,16 +1,38 @@ +import json +from collections import namedtuple from typing import Dict from typing import Generator from typing import List from typing import Optional +from typing import Set +from typing import Tuple from typing import Type +from django.core import exceptions +from django.db.models import ForeignKey +from django.db.utils import IntegrityError + +from morango.errors import MorangoDirtyParent +from morango.errors import MorangoMissingParent 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.core import Transform from morango.sync.stream.source import MorangoSource from morango.sync.stream.source import SourceTask +DIRTY_PARENT_MESSAGE = "Parent is dirty; could not deserialize." +MISSING_PARENT_MESSAGE = "Parent does not exist in Store; could not deserialize." + +# the store state of an FK target, used to resolve validation failures +_TargetState = namedtuple("_TargetState", ["deleted", "hard_deleted", "dirty_bit"]) + + +def _fk_cache_key(model: Type[SyncableModel], record_id: str) -> str: + """The key `SyncableModel.cached_clean_fields` uses for a record in the `fk_cache`""" + return "{id}_{db_table}".format(id=record_id, db_table=model._meta.db_table) + class DeserializeOutcome: """The resolved action for a `DeserializeTask`, which the sink applies""" @@ -132,3 +154,209 @@ def stream_for_filter( for store_model in qs.iterator(): yield DeserializeTask(store_model, self.fk_cache) + + +class AppModelDeserialize(Transform[DeserializeTask]): + """ + Builds the unsaved app model from each task's serialized store data. Deleted store records + pass through untouched, since deleting the app model is left to the sink. + """ + + def __init__(self, sync_filter: Optional[Filter] = None): + self.sync_filter = sync_filter + + def transform(self, task: DeserializeTask) -> DeserializeTask: + if task.store.deleted: + return task + + try: + app_model = task.model.deserialize( + json.loads(task.store.serialized), sync_filter=self.sync_filter + ) + except (ValueError, exceptions.ValidationError) as e: + # `json.JSONDecodeError` is a `ValueError` + task.add_error(e) + return task + + app_model._morango_source_id = task.store.source_id + app_model._morango_partition = task.store.partition + app_model._morango_dirty_bit = False + task.set_app_model(app_model) + return task + + +class AppModelValidate(Transform[DeserializeTask]): + """ + Validates each task's app model and resolves its outcome. The source streams FK targets ahead + of the records that reference them, so a target validated earlier in the run is already in + the shared `fk_cache` even though the sink may not have written it yet. A cache miss falls back + to looking up the target in the app tables. + + Must be instantiated per pipeline run, since it tracks which records failed or were deleted + during the run, so that the records referencing them can be resolved accordingly. + + Deletions spread to referencing records as in legacy deserialization. For models without a + self-referential FK, a reference to any deleted record spreads its deletion. For models with + one, only a validation failure of `ObjectDoesNotExist` spreads it. + """ + + def __init__(self, sync_filter: Optional[Filter] = None): + self.sync_filter = sync_filter + self.failed_ids: Set[str] = set() + # maps the ID of a deleted record to whether the deletion is a hard deletion + self.deleted_ids: Dict[str, bool] = {} + + def transform(self, task: DeserializeTask) -> DeserializeTask: + if task.has_errors: + self.failed_ids.add(task.id) + return task + + if task.store.deleted: + self.deleted_ids[task.id] = task.store.hard_deleted + return task + + fk_references = self._fk_references(task.app_model) + + for field, target_id in fk_references: + if target_id in self.deleted_ids: + if task.self_referential_fk() is not None: + # the sink has not deleted the target's app row yet, so validation can't + # report the missing target; fail as legacy does once the row is gone + self._fail(task, self._deleted_target_error(field, target_id)) + else: + self._propagate_deletion(task, hard=self.deleted_ids[target_id]) + return task + + for field, target_id in fk_references: + if target_id in self.failed_ids: + self._fail(task, self._failed_target_error(task, field, target_id)) + return task + + try: + task.app_model.cached_clean_fields(task.fk_cache, sync_filter=self.sync_filter) + except (exceptions.ValidationError, exceptions.ObjectDoesNotExist, ValueError) as e: + self._classify_failure(task, fk_references, e) + return task + + @staticmethod + def _fk_references(app_model: SyncableModel) -> List[Tuple[ForeignKey, str]]: + """The FK fields on the app model that hold a value, paired with that value""" + references = [] + for field in app_model._meta.fields: + if not isinstance(field, ForeignKey): + continue + target_id = getattr(app_model, field.attname) + if target_id is not None: + references.append((field, target_id)) + return references + + @staticmethod + def _failed_target_error(task: DeserializeTask, field: ForeignKey, target_id: str) -> Exception: + if field.attname == task.self_referential_fk(): + return MorangoDirtyParent(DIRTY_PARENT_MESSAGE) + return exceptions.ValidationError( + "{} with id {} failed to deserialize".format( + field.related_model._meta.verbose_name, target_id + ) + ) + + @staticmethod + def _deleted_target_error(field: ForeignKey, target_id: str) -> Exception: + return exceptions.ValidationError( + "{} with id {} was deleted".format(field.related_model._meta.verbose_name, target_id) + ) + + def _classify_failure( + self, + task: DeserializeTask, + fk_references: List[Tuple[ForeignKey, str]], + error: Exception, + ) -> None: + """ + Resolves a validation failure using the store records of the FK targets. Only runs on + failure, so it costs at most one query per failed record and none per valid record. + """ + targets = self._lookup_targets(fk_references) + + # a reference to a deleted record propagates the deletion, instead of failing + if task.self_referential_fk() is None or isinstance(error, exceptions.ObjectDoesNotExist): + for _, target_id in fk_references: + target = targets.get(target_id) + if target is not None and (target.deleted or target.hard_deleted): + self._propagate_deletion(task, hard=target.hard_deleted) + return + + self._fail(task, self._failure_error(task, fk_references, targets, error)) + + @staticmethod + def _lookup_targets( + fk_references: List[Tuple[ForeignKey, str]], + ) -> Dict[str, _TargetState]: + """The store state of each FK target that has a store record, keyed by the target ID""" + if not fk_references: + return {} + target_ids = [target_id for _, target_id in fk_references] + return { + target_id: _TargetState(deleted, hard_deleted, dirty_bit) + for target_id, deleted, hard_deleted, dirty_bit in Store.objects.filter( + id__in=target_ids + ).values_list("id", "deleted", "hard_deleted", "dirty_bit") + } + + def _failure_error( + self, + task: DeserializeTask, + fk_references: List[Tuple[ForeignKey, str]], + targets: Dict[str, _TargetState], + error: Exception, + ) -> Exception: + """ + The error to record for a failure that did not propagate a deletion. Problems with the + parent take precedence, then missing targets, otherwise the original validation error. + """ + self_ref_fk = task.self_referential_fk() + for field, target_id in fk_references: + if field.attname != self_ref_fk: + continue + if target_id not in targets: + return MorangoMissingParent(MISSING_PARENT_MESSAGE) + if targets[target_id].dirty_bit and self._parent_not_written( + task, field, target_id, error + ): + return MorangoDirtyParent(DIRTY_PARENT_MESSAGE) + + for field, target_id in fk_references: + if target_id not in targets: + return IntegrityError( + "{from_model}.{from_field} references non-existent {to_model} instance with " + "id '{to_pk}'".format( + from_model=task.model.__name__, + from_field=field.attname, + to_model=field.related_model._meta.verbose_name, + to_pk=target_id, + ) + ) + + return error + + @staticmethod + def _parent_not_written( + task: DeserializeTask, field: ForeignKey, parent_id: str, error: Exception + ) -> bool: + """ + Whether a failure is explained by a dirty parent's app row not being written. The parent + is in `fk_cache` when its FK resolved, either because it was accepted earlier in the run or + because an older app row exists, and then only loading the parent object can fail on it. + Any other error is the record's own. + """ + return _fk_cache_key(field.related_model, parent_id) not in task.fk_cache or isinstance( + error, exceptions.ObjectDoesNotExist + ) + + def _propagate_deletion(self, task: DeserializeTask, hard: bool) -> None: + task.set_propagated_deletion(hard) + self.deleted_ids[task.id] = hard + + def _fail(self, task: DeserializeTask, error: Exception) -> None: + task.add_error(error) + self.failed_ids.add(task.id) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 5c75fe23..491929f8 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -1,12 +1,22 @@ import uuid import mock +from django.core.exceptions import ValidationError +from django.core.serializers.json import DjangoJSONEncoder +from django.db.utils import IntegrityError from django.test import SimpleTestCase from django.test import TestCase +from facility_profile.models import ConditionalLog +from facility_profile.models import Facility +from facility_profile.models import SummaryLog +from morango.errors import MorangoDirtyParent +from morango.errors import MorangoMissingParent from morango.models.certificates import Filter from morango.models.core import Store from morango.models.core import SyncableModel +from morango.sync.stream.deserialize import AppModelDeserialize +from morango.sync.stream.deserialize import AppModelValidate from morango.sync.stream.deserialize import DeserializeOutcome from morango.sync.stream.deserialize import DeserializeTask from morango.sync.stream.deserialize import StoreModelSource @@ -95,6 +105,481 @@ def test_outcome__unresolved(self): _ = self.task.outcome +def _facility_store(facility_id=None, parent_id=None, **kwargs): + """An unsaved `Store` record holding a serialized `Facility`""" + facility_id = facility_id or uuid.uuid4().hex + facility = Facility(id=facility_id, name="Facility {}".format(facility_id), parent_id=parent_id) + store_kwargs = dict( + id=facility_id, + profile="facilitydata", + model_name="facility", + partition=facility_id, + source_id=facility.name, + serialized=DjangoJSONEncoder().encode(facility.serialize()), + last_saved_instance=uuid.uuid4().hex, + last_saved_counter=1, + ) + store_kwargs.update(kwargs) + return Store(**store_kwargs) + + +class AppModelDeserializeTestCase(SimpleTestCase): + def setUp(self): + self.store = _facility_store() + self.task = DeserializeTask(self.store, {}) + + def test_transform__deserializes_app_model(self): + task = AppModelDeserialize().transform(self.task) + + self.assertIs(task, self.task) + self.assertIsInstance(task.app_model, Facility) + self.assertEqual(task.app_model.id, self.store.id) + self.assertEqual(task.outcome, DeserializeOutcome.SAVE) + + def test_transform__sets_morango_fields_from_store(self): + task = AppModelDeserialize().transform(self.task) + + self.assertEqual(task.app_model._morango_source_id, self.store.source_id) + self.assertEqual(task.app_model._morango_partition, self.store.partition) + self.assertFalse(task.app_model._morango_dirty_bit) + + def test_transform__passes_sync_filter(self): + sync_filter = Filter(self.store.partition) + with mock.patch.object(Facility, "deserialize", wraps=Facility.deserialize) as deserialize: + AppModelDeserialize(sync_filter=sync_filter).transform(self.task) + + deserialize.assert_called_once_with(mock.ANY, sync_filter=sync_filter) + + def test_transform__deleted_store(self): + """Deleting the app model is the sink's job, so the deleted record passes through as-is""" + self.store.deleted = True + with mock.patch.object(Facility, "deserialize") as deserialize: + task = AppModelDeserialize().transform(self.task) + + deserialize.assert_not_called() + self.assertIsNone(task.app_model) + self.assertEqual(task.outcome, DeserializeOutcome.DELETE) + + def test_transform__invalid_json(self): + self.store.serialized = "{bad" + task = AppModelDeserialize().transform(self.task) + + self.assertIsNone(task.app_model) + self.assertIsInstance(task.errors[0], ValueError) + self.assertEqual(task.outcome, DeserializeOutcome.ERROR) + + def test_transform__validation_error(self): + error = ValidationError("bad value") + with mock.patch.object(Facility, "deserialize", side_effect=error): + task = AppModelDeserialize().transform(self.task) + + self.assertIsNone(task.app_model) + self.assertEqual(task.errors, [error]) + + def test_transform__unexpected_error_propagates(self): + """Only per-record data errors are recorded; anything else is a defect and must surface""" + with mock.patch.object(Facility, "deserialize", side_effect=RuntimeError("defect")): + with self.assertRaises(RuntimeError): + AppModelDeserialize().transform(self.task) + + +def _validate_task(app_model, fk_cache=None, **store_kwargs): + """A task carrying an already deserialized `app_model`, as `AppModelDeserialize` leaves it""" + store = Store( + id=app_model.id, + profile=app_model.morango_profile, + model_name=app_model.morango_model_name, + **store_kwargs, + ) + task = DeserializeTask(store, fk_cache if fk_cache is not None else {}) + task.set_app_model(app_model) + return task + + +class AppModelValidateTestCase(SimpleTestCase): + """ + Covers the in-run checks that run ahead of model validation. Model validation itself is + patched out, so these never reach the database. + """ + + def setUp(self): + self.validate = AppModelValidate() + self.parent_id = uuid.uuid4().hex + self.user_id = uuid.uuid4().hex + + patcher = mock.patch.object(SyncableModel, "cached_clean_fields") + self.cached_clean_fields = patcher.start() + self.addCleanup(patcher.stop) + + def _facility_task(self, parent_id=None): + return _validate_task(Facility(id=uuid.uuid4().hex, parent_id=parent_id)) + + def _summary_log_task(self): + return _validate_task(SummaryLog(id=uuid.uuid4().hex, user_id=self.user_id)) + + def test_transform__valid(self): + task = self._facility_task(parent_id=self.parent_id) + + self.assertIs(self.validate.transform(task), task) + self.assertEqual(task.outcome, DeserializeOutcome.SAVE) + + def test_transform__validates_with_fk_cache_and_sync_filter(self): + sync_filter = Filter("abc") + task = self._facility_task(parent_id=self.parent_id) + + AppModelValidate(sync_filter=sync_filter).transform(task) + + self.cached_clean_fields.assert_called_once_with(task.fk_cache, sync_filter=sync_filter) + + def test_transform__errored_task_recorded_as_failed(self): + task = self._facility_task() + task.add_error(ValueError("bad data")) + + self.validate.transform(task) + + self.cached_clean_fields.assert_not_called() + self.assertIn(task.id, self.validate.failed_ids) + self.assertEqual(task.outcome, DeserializeOutcome.ERROR) + + def test_transform__deleted_task_recorded_as_deleted(self): + task = _validate_task(Facility(id=uuid.uuid4().hex), deleted=True) + + self.validate.transform(task) + + self.cached_clean_fields.assert_not_called() + self.assertEqual(self.validate.deleted_ids, {task.id: False}) + self.assertEqual(task.outcome, DeserializeOutcome.DELETE) + + def test_transform__hard_deleted_task_recorded_as_hard_deleted(self): + task = _validate_task(Facility(id=uuid.uuid4().hex), deleted=True, hard_deleted=True) + + self.validate.transform(task) + + self.assertEqual(self.validate.deleted_ids, {task.id: True}) + + def test_transform__target_deleted_in_run(self): + """ + The sink may not have deleted the target's app row yet, so validation could wrongly pass + """ + self.validate.deleted_ids[self.user_id] = False + task = self._summary_log_task() + + self.validate.transform(task) + + self.cached_clean_fields.assert_not_called() + self.assertEqual(task.outcome, DeserializeOutcome.PROPAGATE_DELETE) + self.assertEqual(self.validate.deleted_ids[task.id], False) + + def test_transform__target_hard_deleted_in_run(self): + self.validate.deleted_ids[self.user_id] = True + task = self._summary_log_task() + + self.validate.transform(task) + + self.assertEqual(task.outcome, DeserializeOutcome.PROPAGATE_HARD_DELETE) + self.assertEqual(self.validate.deleted_ids[task.id], True) + + def test_transform__parent_deleted_in_run(self): + """Models with a self-referential FK fail instead, as legacy does once the row is gone""" + self.validate.deleted_ids[self.parent_id] = True + task = self._facility_task(parent_id=self.parent_id) + + self.validate.transform(task) + + self.cached_clean_fields.assert_not_called() + self.assertIsInstance(task.errors[0], ValidationError) + self.assertIn(self.parent_id, str(task.errors[0])) + self.assertIn(task.id, self.validate.failed_ids) + self.assertNotIn(task.id, self.validate.deleted_ids) + + def test_transform__parent_deleted_in_run__grandchildren_fail(self): + self.validate.deleted_ids[self.parent_id] = False + child = self._facility_task(parent_id=self.parent_id) + grandchild = self._facility_task(parent_id=child.id) + + self.validate.transform(child) + self.validate.transform(grandchild) + + self.assertIsInstance(grandchild.errors[0], MorangoDirtyParent) + + def test_transform__parent_failed_in_run(self): + self.validate.failed_ids.add(self.parent_id) + task = self._facility_task(parent_id=self.parent_id) + + self.validate.transform(task) + + self.cached_clean_fields.assert_not_called() + self.assertIsInstance(task.errors[0], MorangoDirtyParent) + self.assertEqual(str(task.errors[0]), "Parent is dirty; could not deserialize.") + self.assertIn(task.id, self.validate.failed_ids) + + def test_transform__failure_propagates_to_grandchildren(self): + self.validate.failed_ids.add(self.parent_id) + child = self._facility_task(parent_id=self.parent_id) + grandchild = self._facility_task(parent_id=child.id) + + self.validate.transform(child) + self.validate.transform(grandchild) + + self.assertIsInstance(grandchild.errors[0], MorangoDirtyParent) + + def test_transform__target_failed_in_run(self): + self.validate.failed_ids.add(self.user_id) + task = self._summary_log_task() + + self.validate.transform(task) + + self.assertIsInstance(task.errors[0], ValidationError) + self.assertIn(self.user_id, str(task.errors[0])) + self.assertIn(task.id, self.validate.failed_ids) + + def test_transform__deleted_target_precedes_failed_target(self): + task = _validate_task( + ConditionalLog(id=uuid.uuid4().hex, facility_id=self.parent_id, user_id=self.user_id) + ) + self.validate.failed_ids.add(self.parent_id) + self.validate.deleted_ids[self.user_id] = False + + self.validate.transform(task) + + self.assertEqual(task.outcome, DeserializeOutcome.PROPAGATE_DELETE) + + def test_transform__null_fk_ignored(self): + """A null FK has no target, so the in-run checks must not match it""" + self.validate.failed_ids.add(None) + task = self._facility_task(parent_id=None) + + self.validate.transform(task) + + self.assertEqual(task.outcome, DeserializeOutcome.SAVE) + + +class AppModelValidateClassifyFailureTestCase(TestCase): + """ + When model validation fails, the FK targets' store records determine how the failure is + resolved and reported + """ + + def setUp(self): + self.validate = AppModelValidate() + self.parent_id = uuid.uuid4().hex + self.user_id = uuid.uuid4().hex + self.error = ValidationError("invalid") + + patcher = mock.patch.object(SyncableModel, "cached_clean_fields", side_effect=self.error) + self.cached_clean_fields = patcher.start() + self.addCleanup(patcher.stop) + + def _target_store(self, target_id, model_name, **kwargs): + return StoreFactory( + id=target_id, + profile="facilitydata", + model_name=model_name, + partition=target_id, + serialized="{}", + last_saved_instance=uuid.uuid4().hex, + last_saved_counter=1, + **kwargs, + ) + + def _facility_task(self): + return _validate_task(Facility(id=uuid.uuid4().hex, parent_id=self.parent_id)) + + def _summary_log_task(self): + return _validate_task(SummaryLog(id=uuid.uuid4().hex, user_id=self.user_id)) + + def test_target_deleted(self): + """The target was deleted in an earlier run, so its app row is already gone""" + self._target_store(self.user_id, "user", deleted=True) + task = self._summary_log_task() + + self.validate.transform(task) + + self.assertEqual(task.outcome, DeserializeOutcome.PROPAGATE_DELETE) + self.assertEqual(self.validate.deleted_ids, {task.id: False}) + self.assertNotIn(task.id, self.validate.failed_ids) + + def test_parent_deleted__object_does_not_exist(self): + """Models with a self-referential FK only propagate the deletion on this error""" + self.cached_clean_fields.side_effect = Facility.DoesNotExist() + self._target_store(self.parent_id, "facility", deleted=True) + task = self._facility_task() + + self.validate.transform(task) + + self.assertEqual(task.outcome, DeserializeOutcome.PROPAGATE_DELETE) + self.assertEqual(self.validate.deleted_ids, {task.id: False}) + + def test_parent_deleted__validation_error(self): + self._target_store(self.parent_id, "facility", deleted=True, hard_deleted=True) + task = self._facility_task() + + self.validate.transform(task) + + self.assertEqual(task.errors, [self.error]) + self.assertIn(task.id, self.validate.failed_ids) + self.assertEqual(self.validate.deleted_ids, {}) + + def test_target_hard_deleted(self): + self._target_store(self.user_id, "user", deleted=True, hard_deleted=True) + task = self._summary_log_task() + + self.validate.transform(task) + + self.assertEqual(task.outcome, DeserializeOutcome.PROPAGATE_HARD_DELETE) + self.assertEqual(self.validate.deleted_ids, {task.id: True}) + + def test_target_hard_deleted_only(self): + """Serialization's hard deletion only sets `hard_deleted`, which still counts as deleted""" + self._target_store(self.user_id, "user", hard_deleted=True) + task = self._summary_log_task() + + self.validate.transform(task) + + self.assertEqual(task.outcome, DeserializeOutcome.PROPAGATE_HARD_DELETE) + + def test_target_deleted_precedes_missing_target(self): + """The hard flag comes from the deleted target, regardless of the other targets""" + self._target_store(self.user_id, "user", deleted=True, hard_deleted=True) + task = _validate_task( + ConditionalLog(id=uuid.uuid4().hex, facility_id=self.parent_id, user_id=self.user_id) + ) + + self.validate.transform(task) + + self.assertEqual(task.outcome, DeserializeOutcome.PROPAGATE_HARD_DELETE) + + def test_parent_missing(self): + task = self._facility_task() + + self.validate.transform(task) + + self.assertIsInstance(task.errors[0], MorangoMissingParent) + self.assertEqual( + str(task.errors[0]), "Parent does not exist in Store; could not deserialize." + ) + self.assertIn(task.id, self.validate.failed_ids) + + def test_parent_dirty(self): + """The parent is dirty but was not deserialized in this run, e.g. it was skipped""" + self._target_store(self.parent_id, "facility", dirty_bit=True) + task = self._facility_task() + + self.validate.transform(task) + + self.assertIsInstance(task.errors[0], MorangoDirtyParent) + self.assertEqual(str(task.errors[0]), "Parent is dirty; could not deserialize.") + self.assertIn(task.id, self.validate.failed_ids) + + def _resolved_parent_task(self): + """A task whose parent FK resolved during validation, so the parent is in `fk_cache`""" + task = self._facility_task() + task.fk_cache["{}_{}".format(self.parent_id, Facility._meta.db_table)] = 1 + return task + + def test_parent_resolved(self): + """ + The parent's store record is dirty, but its FK resolved, either because it was accepted + earlier in the run and the sink has not written it yet, or because an older app row exists. + So the error is the record's own. + """ + self._target_store(self.parent_id, "facility", dirty_bit=True) + task = self._resolved_parent_task() + + self.validate.transform(task) + + self.assertEqual(task.errors, [self.error]) + + def test_parent_resolved__object_does_not_exist(self): + """Loading the parent fails because the sink has not written it yet""" + self.cached_clean_fields.side_effect = Facility.DoesNotExist() + self._target_store(self.parent_id, "facility", dirty_bit=True) + task = self._resolved_parent_task() + + self.validate.transform(task) + + self.assertIsInstance(task.errors[0], MorangoDirtyParent) + + def test_target_missing(self): + task = self._summary_log_task() + + self.validate.transform(task) + + self.assertIsInstance(task.errors[0], IntegrityError) + self.assertEqual( + str(task.errors[0]), + "SummaryLog.user_id references non-existent my user instance with id '{}'".format( + self.user_id + ), + ) + self.assertIn(task.id, self.validate.failed_ids) + + def test_parent_clean(self): + """With nothing wrong with the target, the failure is the record's own""" + self._target_store(self.parent_id, "facility", dirty_bit=False) + task = self._facility_task() + + self.validate.transform(task) + + self.assertEqual(task.errors, [self.error]) + self.assertIn(task.id, self.validate.failed_ids) + + def test_target_dirty(self): + """Only the self-referential parent is checked for dirtiness, matching legacy behavior""" + self._target_store(self.user_id, "user", dirty_bit=True) + task = self._summary_log_task() + + self.validate.transform(task) + + self.assertEqual(task.errors, [self.error]) + + def test_no_fk_references(self): + task = _validate_task(Facility(id=uuid.uuid4().hex)) + + with self.assertNumQueries(0): + self.validate.transform(task) + + self.assertEqual(task.errors, [self.error]) + + def test_object_does_not_exist(self): + """e.g. `clean_fields` loading a related object whose row does not exist""" + self.cached_clean_fields.side_effect = Facility.DoesNotExist() + task = self._facility_task() + + self.validate.transform(task) + + self.assertIsInstance(task.errors[0], MorangoMissingParent) + + def test_value_error(self): + """e.g. `clean_fields` failing to coerce bad synced data""" + error = ValueError("bad data") + self.cached_clean_fields.side_effect = error + self._target_store(self.parent_id, "facility", dirty_bit=False) + task = self._facility_task() + + self.validate.transform(task) + + self.assertEqual(task.errors, [error]) + self.assertIn(task.id, self.validate.failed_ids) + + def test_failure_queries_store_once(self): + task = _validate_task( + ConditionalLog(id=uuid.uuid4().hex, facility_id=self.parent_id, user_id=self.user_id) + ) + + with self.assertNumQueries(1): + self.validate.transform(task) + + def test_success_queries_store_never(self): + self.cached_clean_fields.side_effect = None + task = self._facility_task() + + with self.assertNumQueries(0): + self.validate.transform(task) + + self.assertEqual(task.outcome, DeserializeOutcome.SAVE) + + class StoreModelSourcePrefixConditionsTestCase(SimpleTestCase): """`prefix_conditions` is pure ordering logic over the sync filter and touches no ORM""" From dadac46d5caf555fba1d85db3e27b198145a08b4 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Thu, 8 Oct 2026 14:57:53 -0700 Subject: [PATCH 04/13] Add integration and legacy parity tests for deserialize transforms Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tests/sync/stream/test_deserialize.py | 397 ++++++++++++++++++ 1 file changed, 397 insertions(+) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 491929f8..37b36e8c 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -1,25 +1,34 @@ +import json import uuid import mock from django.core.exceptions import ValidationError from django.core.serializers.json import DjangoJSONEncoder +from django.db import connection from django.db.utils import IntegrityError from django.test import SimpleTestCase from django.test import TestCase +from django.test.utils import CaptureQueriesContext from facility_profile.models import ConditionalLog from facility_profile.models import Facility +from facility_profile.models import MyUser from facility_profile.models import SummaryLog from morango.errors import MorangoDirtyParent from morango.errors import MorangoMissingParent from morango.models.certificates import Filter +from morango.models.core import DeletedModels +from morango.models.core import HardDeletedModels from morango.models.core import Store from morango.models.core import SyncableModel +from morango.sync.operations import _deserialize_from_store +from morango.sync.stream.core import Sink from morango.sync.stream.deserialize import AppModelDeserialize from morango.sync.stream.deserialize import AppModelValidate from morango.sync.stream.deserialize import DeserializeOutcome from morango.sync.stream.deserialize import DeserializeTask from morango.sync.stream.deserialize import StoreModelSource +from morango.utils import exception_path from ...helpers import StoreFactory @@ -756,3 +765,391 @@ def test_stream__ignores_other_profiles(self): ) self.assertEqual(self._stream_ids(), [included.id]) + + +class CollectingSink(Sink[DeserializeTask]): + """Collects each task's resolved outcome and first error type, keyed by the task's ID""" + + def __init__(self): + self.results = {} + + def consume(self, task: DeserializeTask) -> None: + error_type = type(task.errors[0]) if task.errors else None + self.results[task.id] = (task.outcome, error_type) + + +class DeserializeFixturesMixin: + """Builds store records for the `facilitydata` profile and runs them through the transforms""" + + profile = "facilitydata" + + def _store(self, app_model, dirty_bit=True, **kwargs): + store_kwargs = dict( + id=app_model.id, + profile=self.profile, + model_name=app_model.morango_model_name, + partition=app_model.id, + source_id=uuid.uuid4().hex, + serialized=DjangoJSONEncoder().encode(app_model.serialize()), + last_saved_instance=uuid.uuid4().hex, + last_saved_counter=1, + dirty_bit=dirty_bit, + ) + store_kwargs.update(kwargs) + return StoreFactory(**store_kwargs) + + def _facility(self, parent=None, order=None, name="facility", **kwargs): + facility = Facility( + id=uuid.uuid4().hex, + name=name, + parent_id=parent.id if parent else None, + ) + if order is None and parent is None: + order = 0 + return self._store( + facility, _self_ref_fk=facility.parent_id or "", _self_ref_order=order, **kwargs + ) + + def _user(self, **kwargs): + user = MyUser(id=uuid.uuid4().hex, username=uuid.uuid4().hex[:20], password="password") + return self._store(user, **kwargs) + + def _summary_log(self, user_id, **kwargs): + return self._store(SummaryLog(id=uuid.uuid4().hex, user_id=user_id), **kwargs) + + def _run(self, sync_filter=None, skip_errored=False): + sink = CollectingSink() + ( + StoreModelSource(self.profile, sync_filter=sync_filter, skip_errored=skip_errored) + .pipe(AppModelDeserialize(sync_filter=sync_filter)) + .pipe(AppModelValidate(sync_filter=sync_filter)) + .end(sink) + ) + return sink.results + + def _run_both_modes(self, **run_kwargs): + """Runs once for each `Facility.clean_dereferences_parent` mode, returning both results""" + results = {} + for dereferences_parent in (False, True): + with mock.patch.object(Facility, "clean_dereferences_parent", dereferences_parent): + results[dereferences_parent] = self._run(**run_kwargs) + return results + + def _dirty_parent_with_app_row(self, **kwargs): + """A parent whose store record is dirty, but which has an older app row""" + parent = Facility.objects.create(name="parent") + self._store(parent, _self_ref_order=0, **kwargs) + return parent + + +class DeserializeTransformsIntegrationTestCase(DeserializeFixturesMixin, TestCase): + """ + Streams real store records through the source and both transforms, without a writing sink, + so app rows only exist where a test creates them. This mirrors the sink lag, where records + accepted earlier in the run have not been written yet. + """ + + def test_self_ref_chain_in_run(self): + root = self._facility() + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + + with mock.patch.object(Facility, "clean_dereferences_parent", False): + results = self._run() + + self.assertEqual(results[root.id], (DeserializeOutcome.SAVE, None)) + self.assertEqual(results[child.id], (DeserializeOutcome.SAVE, None)) + self.assertEqual(results[grandchild.id], (DeserializeOutcome.SAVE, None)) + self.assertEqual(Facility.objects.count(), 0) + + def test_self_ref_chain_in_run__dereferences_parent(self): + """Loading the parent fails, because the sink has not written it yet""" + root = self._facility() + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + + with mock.patch.object(Facility, "clean_dereferences_parent", True): + results = self._run() + + self.assertEqual(results[root.id], (DeserializeOutcome.SAVE, None)) + self.assertEqual(results[child.id], (DeserializeOutcome.ERROR, MorangoDirtyParent)) + self.assertEqual(results[grandchild.id], (DeserializeOutcome.ERROR, MorangoDirtyParent)) + + def test_own_error_with_parent_accepted_in_run(self): + """The parent's store record is still dirty, but the child fails on its own field""" + root = self._facility() + child = self._facility(parent=root, order=1, name="x" * 200) + + with mock.patch.object(Facility, "clean_dereferences_parent", False): + results = self._run() + + self.assertEqual(results[root.id], (DeserializeOutcome.SAVE, None)) + self.assertEqual(results[child.id], (DeserializeOutcome.ERROR, ValidationError)) + + def test_cross_model_chain_in_run(self): + user = self._user() + log = self._summary_log(user.id) + + results = self._run() + + self.assertEqual(results[user.id], (DeserializeOutcome.SAVE, None)) + self.assertEqual(results[log.id], (DeserializeOutcome.SAVE, None)) + + def test_parent_not_in_run(self): + """The common case, where the parent is unchanged and found through the app tables""" + parent = Facility.objects.create(name="parent") + self._store(parent, dirty_bit=False, _self_ref_order=0) + child = self._facility(parent=parent, order=1) + + for mode, results in self._run_both_modes().items(): + with self.subTest(dereferences_parent=mode): + self.assertEqual(results[child.id], (DeserializeOutcome.SAVE, None)) + + def test_dirty_parent_not_in_run__skip_errored(self): + """ + Unlike legacy, the child validates against the parent's older app row, so it is saved. A + sibling that fails its own validation keeps its own error. Legacy never deserializes + children of a parent whose store record is dirty, and records `MorangoMissingParent` when + the parent is left out of the run. + """ + parent = self._dirty_parent_with_app_row(deserialization_error="failed earlier") + child = self._facility(parent=parent, order=1) + invalid_child = self._facility(parent=parent, order=1, name="x" * 200) + + for mode, results in self._run_both_modes(skip_errored=True).items(): + with self.subTest(dereferences_parent=mode): + self.assertNotIn(parent.id, results) + self.assertEqual(results[child.id], (DeserializeOutcome.SAVE, None)) + self.assertEqual( + results[invalid_child.id], (DeserializeOutcome.ERROR, ValidationError) + ) + + def test_dirty_parent_not_in_run__filtered(self): + """Same as with `skip_errored`, when the sync filter leaves out the parent's partition""" + parent = self._dirty_parent_with_app_row() + child = self._facility(parent=parent, order=1) + + for mode, results in self._run_both_modes(sync_filter=Filter(child.partition)).items(): + with self.subTest(dereferences_parent=mode): + self.assertNotIn(parent.id, results) + self.assertEqual(results[child.id], (DeserializeOutcome.SAVE, None)) + + def test_parent_failed_in_run(self): + root = self._facility(serialized="{bad") + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + + for mode, results in self._run_both_modes().items(): + with self.subTest(dereferences_parent=mode): + self.assertEqual(results[root.id], (DeserializeOutcome.ERROR, json.JSONDecodeError)) + self.assertEqual(results[child.id], (DeserializeOutcome.ERROR, MorangoDirtyParent)) + self.assertEqual( + results[grandchild.id], (DeserializeOutcome.ERROR, MorangoDirtyParent) + ) + + def test_parent_missing(self): + missing_parent = Facility(id=uuid.uuid4().hex, name="missing") + child = self._facility(parent=missing_parent) + + for mode, results in self._run_both_modes().items(): + with self.subTest(dereferences_parent=mode): + self.assertEqual( + results[child.id], (DeserializeOutcome.ERROR, MorangoMissingParent) + ) + + def test_parent_deleted_in_run(self): + """ + Models with a self-referential FK don't propagate the deletion. When the parent object is + loaded, this differs from legacy, which propagates it once the sink has deleted the parent. + """ + root = self._facility(deleted=True) + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + + for mode, results in self._run_both_modes().items(): + with self.subTest(dereferences_parent=mode): + self.assertEqual(results[root.id], (DeserializeOutcome.DELETE, None)) + self.assertEqual(results[child.id], (DeserializeOutcome.ERROR, ValidationError)) + self.assertEqual( + results[grandchild.id], (DeserializeOutcome.ERROR, MorangoDirtyParent) + ) + + def test_parent_deleted_earlier(self): + root = self._facility(dirty_bit=False, deleted=True) + child = self._facility(parent=root, order=1) + + results = self._run_both_modes() + + self.assertEqual(results[False][child.id], (DeserializeOutcome.ERROR, ValidationError)) + self.assertEqual(results[True][child.id], (DeserializeOutcome.PROPAGATE_DELETE, None)) + + def test_parent_hard_deleted_earlier(self): + root = self._facility(dirty_bit=False, deleted=True, hard_deleted=True) + child = self._facility(parent=root, order=1) + + results = self._run_both_modes() + + self.assertEqual(results[False][child.id], (DeserializeOutcome.ERROR, ValidationError)) + self.assertEqual(results[True][child.id], (DeserializeOutcome.PROPAGATE_HARD_DELETE, None)) + + def test_target_deleted_in_run(self): + user = self._user(deleted=True) + log = self._summary_log(user.id) + + results = self._run() + + self.assertEqual(results[user.id], (DeserializeOutcome.DELETE, None)) + self.assertEqual(results[log.id], (DeserializeOutcome.PROPAGATE_DELETE, None)) + + def test_target_deleted_earlier(self): + user = self._user(dirty_bit=False, deleted=True, hard_deleted=True) + log = self._summary_log(user.id) + + results = self._run() + + self.assertEqual(results[log.id], (DeserializeOutcome.PROPAGATE_HARD_DELETE, None)) + + def test_target_missing(self): + log = self._summary_log(uuid.uuid4().hex) + + results = self._run() + + self.assertEqual(results[log.id], (DeserializeOutcome.ERROR, IntegrityError)) + + def test_transforms_only_read(self): + root = self._facility() + self._facility(parent=root, order=1) + self._facility(serialized="{bad") + self._facility(deleted=True) + self._summary_log(uuid.uuid4().hex) + + with CaptureQueriesContext(connection) as context: + self._run() + + statements = [query["sql"].strip().upper() for query in context.captured_queries] + self.assertTrue(statements) + for statement in statements: + # postgres wraps iterated reads in a server-side cursor (`DECLARE ... FOR SELECT`), so + # check for writes rather than requiring every statement to start with `SELECT` + self.assertFalse( + statement.startswith(("INSERT", "UPDATE", "DELETE")), + statement, + ) + + +class LegacyParityTestCase(DeserializeFixturesMixin, TestCase): + """ + Characterizes legacy `_deserialize_from_store` errors and propagated deletions against the + transforms, on the same store records. The transforms don't write, so both can run against the + same database state. Runs with the default `Facility.clean_dereferences_parent`, which is the + legacy test behavior, unless a test patches it. + """ + + def _legacy_result(self, store_id): + """The legacy result of a store record that did not save, in the transforms' terms""" + if HardDeletedModels.objects.filter(id=store_id).exists(): + return DeserializeOutcome.PROPAGATE_HARD_DELETE, None + if DeletedModels.objects.filter(id=store_id).exists(): + return DeserializeOutcome.PROPAGATE_DELETE, None + exception = Store.objects.get(id=store_id).deserialization_exception + self.assertIsNotNone(exception) + return DeserializeOutcome.ERROR, exception + + def _assert_parity(self, *store_ids): + results = self._run() + _deserialize_from_store(self.profile) + + for store_id in store_ids: + with self.subTest(store_id=store_id): + outcome, error_type = results[store_id] + error_path = exception_path(error_type) if error_type else None + self.assertEqual((outcome, error_path), self._legacy_result(store_id)) + + def test_parent_failed(self): + root = self._facility(serialized="{bad") + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + + self._assert_parity(root.id, child.id, grandchild.id) + + def test_parent_missing(self): + missing_parent = Facility(id=uuid.uuid4().hex, name="missing") + child = self._facility(parent=missing_parent) + + self._assert_parity(child.id) + + def test_target_missing(self): + log = self._summary_log(uuid.uuid4().hex) + + self._assert_parity(log.id) + + @mock.patch.object(Facility, "clean_dereferences_parent", False) + def test_own_error_with_parent_accepted(self): + root = self._facility() + child = self._facility(parent=root, order=1, name="x" * 200) + + self._assert_parity(child.id) + + def test_target_deleted_in_run(self): + user = self._user(deleted=True) + log = self._summary_log(user.id) + + self._assert_parity(log.id) + + def test_target_deleted_earlier(self): + user = self._user(dirty_bit=False, deleted=True) + log = self._summary_log(user.id) + + self._assert_parity(log.id) + + def test_target_hard_deleted_earlier(self): + user = self._user(dirty_bit=False, deleted=True, hard_deleted=True) + log = self._summary_log(user.id) + + self._assert_parity(log.id) + + def test_parent_deleted_earlier(self): + root = self._facility(dirty_bit=False, deleted=True) + child = self._facility(parent=root, order=1) + + self._assert_parity(child.id) + + def test_parent_hard_deleted_earlier(self): + root = self._facility(dirty_bit=False, deleted=True, hard_deleted=True) + child = self._facility(parent=root, order=1) + + self._assert_parity(child.id) + + @mock.patch.object(Facility, "clean_dereferences_parent", False) + def test_parent_deleted_earlier__parent_not_loaded(self): + root = self._facility(dirty_bit=False, deleted=True) + child = self._facility(parent=root, order=1) + + self._assert_parity(child.id) + + def test_dirty_parent_not_in_run__differs(self): + """ + Characterizes an intentional difference: legacy never deserializes children of a parent + whose store record is dirty. The transforms save the child, which validates against the + parent's older app row. + """ + parent = Facility.objects.create(name="parent") + self._store(parent, _self_ref_order=0, deserialization_error="failed earlier") + child = self._facility(parent=parent, order=1) + + results = self._run(skip_errored=True) + _deserialize_from_store(self.profile, skip_erroring=True) + + self.assertEqual(results[child.id], (DeserializeOutcome.SAVE, None)) + self.assertEqual( + self._legacy_result(child.id), + (DeserializeOutcome.ERROR, exception_path(MorangoMissingParent)), + ) + + @mock.patch.object(Facility, "clean_dereferences_parent", False) + def test_parent_deleted_in_run__parent_not_loaded(self): + root = self._facility(deleted=True) + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + + self._assert_parity(child.id, grandchild.id) From 8d5a2ec1703553d4fe27acd9d018da3e446f4240 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:20:13 -0700 Subject: [PATCH 05/13] Move deserialize FK helpers to module level for reuse by the sink Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 90 +++++++++++++++--------------- 1 file changed, 46 insertions(+), 44 deletions(-) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index c8414408..b1889ad5 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -156,6 +156,36 @@ def stream_for_filter( yield DeserializeTask(store_model, self.fk_cache) +def fk_references(app_model: SyncableModel) -> List[Tuple[ForeignKey, str]]: + """The FK fields on the app model that hold a value, paired with that value""" + references = [] + for field in app_model._meta.fields: + if not isinstance(field, ForeignKey): + continue + target_id = getattr(app_model, field.attname) + if target_id is not None: + references.append((field, target_id)) + return references + + +def failed_target_error(task: DeserializeTask, field: ForeignKey, target_id: str) -> Exception: + """The error for a task whose FK target failed to deserialize in the run""" + if field.attname == task.self_referential_fk(): + return MorangoDirtyParent(DIRTY_PARENT_MESSAGE) + return exceptions.ValidationError( + "{} with id {} failed to deserialize".format( + field.related_model._meta.verbose_name, target_id + ) + ) + + +def deleted_target_error(field: ForeignKey, target_id: str) -> Exception: + """The error for a task whose FK target was deleted in the run""" + return exceptions.ValidationError( + "{} with id {} was deleted".format(field.related_model._meta.verbose_name, target_id) + ) + + class AppModelDeserialize(Transform[DeserializeTask]): """ Builds the unsaved app model from each task's serialized store data. Deleted store records @@ -215,87 +245,59 @@ def transform(self, task: DeserializeTask) -> DeserializeTask: self.deleted_ids[task.id] = task.store.hard_deleted return task - fk_references = self._fk_references(task.app_model) + references = fk_references(task.app_model) - for field, target_id in fk_references: + for field, target_id in references: if target_id in self.deleted_ids: if task.self_referential_fk() is not None: # the sink has not deleted the target's app row yet, so validation can't # report the missing target; fail as legacy does once the row is gone - self._fail(task, self._deleted_target_error(field, target_id)) + self._fail(task, deleted_target_error(field, target_id)) else: self._propagate_deletion(task, hard=self.deleted_ids[target_id]) return task - for field, target_id in fk_references: + for field, target_id in references: if target_id in self.failed_ids: - self._fail(task, self._failed_target_error(task, field, target_id)) + self._fail(task, failed_target_error(task, field, target_id)) return task try: task.app_model.cached_clean_fields(task.fk_cache, sync_filter=self.sync_filter) except (exceptions.ValidationError, exceptions.ObjectDoesNotExist, ValueError) as e: - self._classify_failure(task, fk_references, e) + self._classify_failure(task, references, e) return task - @staticmethod - def _fk_references(app_model: SyncableModel) -> List[Tuple[ForeignKey, str]]: - """The FK fields on the app model that hold a value, paired with that value""" - references = [] - for field in app_model._meta.fields: - if not isinstance(field, ForeignKey): - continue - target_id = getattr(app_model, field.attname) - if target_id is not None: - references.append((field, target_id)) - return references - - @staticmethod - def _failed_target_error(task: DeserializeTask, field: ForeignKey, target_id: str) -> Exception: - if field.attname == task.self_referential_fk(): - return MorangoDirtyParent(DIRTY_PARENT_MESSAGE) - return exceptions.ValidationError( - "{} with id {} failed to deserialize".format( - field.related_model._meta.verbose_name, target_id - ) - ) - - @staticmethod - def _deleted_target_error(field: ForeignKey, target_id: str) -> Exception: - return exceptions.ValidationError( - "{} with id {} was deleted".format(field.related_model._meta.verbose_name, target_id) - ) - def _classify_failure( self, task: DeserializeTask, - fk_references: List[Tuple[ForeignKey, str]], + references: List[Tuple[ForeignKey, str]], error: Exception, ) -> None: """ Resolves a validation failure using the store records of the FK targets. Only runs on failure, so it costs at most one query per failed record and none per valid record. """ - targets = self._lookup_targets(fk_references) + targets = self._lookup_targets(references) # a reference to a deleted record propagates the deletion, instead of failing if task.self_referential_fk() is None or isinstance(error, exceptions.ObjectDoesNotExist): - for _, target_id in fk_references: + for _, target_id in references: target = targets.get(target_id) if target is not None and (target.deleted or target.hard_deleted): self._propagate_deletion(task, hard=target.hard_deleted) return - self._fail(task, self._failure_error(task, fk_references, targets, error)) + self._fail(task, self._failure_error(task, references, targets, error)) @staticmethod def _lookup_targets( - fk_references: List[Tuple[ForeignKey, str]], + references: List[Tuple[ForeignKey, str]], ) -> Dict[str, _TargetState]: """The store state of each FK target that has a store record, keyed by the target ID""" - if not fk_references: + if not references: return {} - target_ids = [target_id for _, target_id in fk_references] + target_ids = [target_id for _, target_id in references] return { target_id: _TargetState(deleted, hard_deleted, dirty_bit) for target_id, deleted, hard_deleted, dirty_bit in Store.objects.filter( @@ -306,7 +308,7 @@ def _lookup_targets( def _failure_error( self, task: DeserializeTask, - fk_references: List[Tuple[ForeignKey, str]], + references: List[Tuple[ForeignKey, str]], targets: Dict[str, _TargetState], error: Exception, ) -> Exception: @@ -315,7 +317,7 @@ def _failure_error( parent take precedence, then missing targets, otherwise the original validation error. """ self_ref_fk = task.self_referential_fk() - for field, target_id in fk_references: + for field, target_id in references: if field.attname != self_ref_fk: continue if target_id not in targets: @@ -325,7 +327,7 @@ def _failure_error( ): return MorangoDirtyParent(DIRTY_PARENT_MESSAGE) - for field, target_id in fk_references: + for field, target_id in references: if target_id not in targets: return IntegrityError( "{from_model}.{from_field} references non-existent {to_model} instance with " From 908fca092040fb1b666c3ed4582dbfe3e3e60910 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:22:41 -0700 Subject: [PATCH 06/13] Add write helpers for deleting app models and tracking propagated deletions Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 56 ++++++++ .../tests/sync/stream/test_deserialize.py | 121 ++++++++++++++++++ 2 files changed, 177 insertions(+) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index b1889ad5..4d49f228 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -9,18 +9,25 @@ from typing import Type from django.core import exceptions +from django.db import router from django.db.models import ForeignKey +from django.db.models import Model +from django.db.models import signals +from django.db.models.deletion import Collector from django.db.utils import IntegrityError from morango.errors import MorangoDirtyParent from morango.errors import MorangoMissingParent from morango.models.certificates import Filter +from morango.models.core import DeletedModels +from morango.models.core import HardDeletedModels from morango.models.core import Store from morango.models.core import SyncableModel from morango.registry import syncable_models from morango.sync.stream.core import Transform from morango.sync.stream.source import MorangoSource from morango.sync.stream.source import SourceTask +from morango.sync.utils import mute_signals DIRTY_PARENT_MESSAGE = "Parent is dirty; could not deserialize." MISSING_PARENT_MESSAGE = "Parent does not exist in Store; could not deserialize." @@ -186,6 +193,55 @@ def deleted_target_error(field: ForeignKey, target_id: str) -> Exception: ) +def row_key(model: Type[Model], pk) -> Tuple[str, str]: + """Identifies an app row across models, as the concrete model's label and the primary key""" + return model._meta.concrete_model._meta.label_lower, str(pk) + + +def delete_app_models(model: Type[SyncableModel], ids: List[str]) -> Set[Tuple[str, str]]: + """ + Deletes app rows without tracking them, and Django cascades to the rows that reference them. + + :return: The row keys of the cascaded rows, not including the rows in `ids` + """ + if not ids: + return set() + + with mute_signals(signals.post_delete): + # the collector checks for signal receivers when deciding to fast delete, so it must + # collect while `post_delete` is muted + collector = Collector(using=router.db_for_write(model)) + collector.collect(model.syncing_objects.filter(id__in=ids)) + + keys = set() + for related_model, instances in collector.data.items(): + keys.update(row_key(related_model, instance.pk) for instance in instances) + for qs in collector.fast_deletes: + keys.update(row_key(qs.model, pk) for pk in qs.values_list("pk", flat=True)) + + collector.delete() + + return keys - {row_key(model, pk) for pk in ids} + + +def record_propagated_deletions(tasks: List[DeserializeTask]) -> None: + """ + Adds the tasks to `DeletedModels`, and those with a propagated hard deletion to + `HardDeletedModels`, so the next serialization marks their store records deleted + """ + deleted = [DeletedModels(id=task.id, profile=task.store.profile) for task in tasks] + if deleted: + DeletedModels.objects.bulk_create(deleted, ignore_conflicts=True) + + hard_deleted = [ + HardDeletedModels(id=task.id, profile=task.store.profile) + for task in tasks + if task.outcome == DeserializeOutcome.PROPAGATE_HARD_DELETE + ] + if hard_deleted: + HardDeletedModels.objects.bulk_create(hard_deleted, ignore_conflicts=True) + + class AppModelDeserialize(Transform[DeserializeTask]): """ Builds the unsaved app model from each task's serialized store data. Deleted store records diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 37b36e8c..80393cfe 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -5,12 +5,15 @@ from django.core.exceptions import ValidationError from django.core.serializers.json import DjangoJSONEncoder from django.db import connection +from django.db.models import SET_NULL +from django.db.models import signals from django.db.utils import IntegrityError from django.test import SimpleTestCase from django.test import TestCase from django.test.utils import CaptureQueriesContext from facility_profile.models import ConditionalLog from facility_profile.models import Facility +from facility_profile.models import InteractionLog from facility_profile.models import MyUser from facility_profile.models import SummaryLog @@ -25,8 +28,11 @@ from morango.sync.stream.core import Sink from morango.sync.stream.deserialize import AppModelDeserialize from morango.sync.stream.deserialize import AppModelValidate +from morango.sync.stream.deserialize import delete_app_models from morango.sync.stream.deserialize import DeserializeOutcome from morango.sync.stream.deserialize import DeserializeTask +from morango.sync.stream.deserialize import record_propagated_deletions +from morango.sync.stream.deserialize import row_key from morango.sync.stream.deserialize import StoreModelSource from morango.utils import exception_path @@ -1153,3 +1159,118 @@ def test_parent_deleted_in_run__parent_not_loaded(self): grandchild = self._facility(parent=child, order=2) self._assert_parity(child.id, grandchild.id) + + +class DeleteAppModelsTestCase(TestCase): + def _facility(self, name, parent=None): + facility = Facility(name=name, parent=parent) + facility.save(update_dirty_bit_to=False) + return facility + + def _user(self, username="user"): + user = MyUser(username=username, password="password") + user.save(update_dirty_bit_to=False) + return user + + def test_self_ref_cascade(self): + root = self._facility("root") + child = self._facility("child", parent=root) + grandchild = self._facility("grandchild", parent=child) + + cascaded = delete_app_models(Facility, [root.id]) + + self.assertEqual(cascaded, {row_key(Facility, child.id), row_key(Facility, grandchild.id)}) + self.assertFalse( + Facility.objects.filter(id__in=[root.id, child.id, grandchild.id]).exists() + ) + self.assertFalse(DeletedModels.objects.exists()) + self.assertFalse(HardDeletedModels.objects.exists()) + + def test_cross_model_cascade__fast_delete(self): + """With `post_delete` muted, Django deletes the logs without loading them""" + user = self._user() + log = SummaryLog(user=user) + log.save(update_dirty_bit_to=False) + + cascaded = delete_app_models(MyUser, [user.id]) + + self.assertEqual(cascaded, {row_key(SummaryLog, log.id)}) + self.assertFalse(MyUser.objects.filter(id=user.id).exists()) + self.assertFalse(SummaryLog.objects.filter(id=log.id).exists()) + self.assertFalse(DeletedModels.objects.exists()) + + def test_set_null_not_cascaded(self): + user = self._user() + log = InteractionLog(user=user) + log.save(update_dirty_bit_to=False) + + remote_field = InteractionLog._meta.get_field("user").remote_field + with mock.patch.object(remote_field, "on_delete", SET_NULL): + cascaded = delete_app_models(MyUser, [user.id]) + + self.assertEqual(cascaded, set()) + self.assertIsNone(InteractionLog.objects.get(id=log.id).user_id) + + def test_post_delete_muted(self): + root = self._facility("root") + self._facility("child", parent=root) + receiver = mock.Mock() + signals.post_delete.connect(receiver) + self.addCleanup(signals.post_delete.disconnect, receiver) + + delete_app_models(Facility, [root.id]) + + receiver.assert_not_called() + + def test_missing_rows(self): + self.assertEqual(delete_app_models(Facility, [uuid.uuid4().hex]), set()) + + def test_empty_ids(self): + with self.assertNumQueries(0): + self.assertEqual(delete_app_models(Facility, []), set()) + + +class RecordPropagatedDeletionsTestCase(DeserializeFixturesMixin, TestCase): + def _task(self, hard=False): + task = DeserializeTask(self._user(), {}) + task.set_propagated_deletion(hard=hard) + return task + + def test_soft(self): + task = self._task() + + record_propagated_deletions([task]) + + self.assertEqual( + list(DeletedModels.objects.values_list("id", "profile")), [(task.id, self.profile)] + ) + self.assertFalse(HardDeletedModels.objects.exists()) + + def test_hard(self): + task = self._task(hard=True) + + record_propagated_deletions([task]) + + self.assertTrue(DeletedModels.objects.filter(id=task.id, profile=self.profile).exists()) + self.assertTrue(HardDeletedModels.objects.filter(id=task.id, profile=self.profile).exists()) + + def test_existing_row(self): + task = self._task() + DeletedModels.objects.create(id=task.id, profile=self.profile) + + record_propagated_deletions([task]) + + self.assertEqual(DeletedModels.objects.filter(id=task.id).count(), 1) + + def test_empty(self): + with self.assertNumQueries(0): + record_propagated_deletions([]) + + def test_full_chunk(self): + """Crosses SQLite's 499-row batch limit for `bulk_create`""" + tasks = [self._task(hard=True) for _ in range(500)] + + record_propagated_deletions(tasks) + + self.assertEqual(DeletedModels.objects.count(), 500) + self.assertEqual(HardDeletedModels.objects.count(), 500) From 8d857b73dd72615b428e9366b0b97573e8f360bd Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:25:36 -0700 Subject: [PATCH 07/13] Add sink that writes deserialization outcomes per chunk Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 119 +++++++++++ .../tests/sync/stream/test_deserialize.py | 191 ++++++++++++++++++ 2 files changed, 310 insertions(+) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index 4d49f228..f65799f5 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -1,4 +1,6 @@ import json +import logging +from collections import defaultdict from collections import namedtuple from typing import Dict from typing import Generator @@ -10,6 +12,7 @@ from django.core import exceptions from django.db import router +from django.db import transaction from django.db.models import ForeignKey from django.db.models import Model from django.db.models import signals @@ -24,11 +27,14 @@ from morango.models.core import Store from morango.models.core import SyncableModel from morango.registry import syncable_models +from morango.sync.stream.core import Sink from morango.sync.stream.core import Transform from morango.sync.stream.source import MorangoSource from morango.sync.stream.source import SourceTask from morango.sync.utils import mute_signals +logger = logging.getLogger(__name__) + DIRTY_PARENT_MESSAGE = "Parent is dirty; could not deserialize." MISSING_PARENT_MESSAGE = "Parent does not exist in Store; could not deserialize." @@ -418,3 +424,116 @@ def _propagate_deletion(self, task: DeserializeTask, hard: bool) -> None: def _fail(self, task: DeserializeTask, error: Exception) -> None: task.add_error(error) self.failed_ids.add(task.id) + + +class DeserializeSink(Sink[List[DeserializeTask]]): + """ + Writes each task's resolved outcome to the app tables, the store and the deletion tracking + tables. Consumes chunks that each hold tasks of one model. + + Must be instantiated per pipeline run, since it tracks the records that failed and the app rows + removed by cascading deletes. The buffer in front of the sink means a task is validated before + the sink writes the chunks ahead of it, so a save is checked against both before saving. + """ + + def __init__(self): + self.failed_ids: Set[str] = set() + self.cascaded_keys: Set[Tuple[str, str]] = set() + self.counts: Dict[str, int] = defaultdict(int) + + def consume(self, tasks: List[DeserializeTask]) -> None: + by_outcome = defaultdict(list) + for task in tasks: + by_outcome[task.outcome].append(task) + + clean = [] + errored = [] + + for task in by_outcome[DeserializeOutcome.ERROR]: + self._record_error(task, task.errors[0]) + errored.append(task) + + # deletes run before saves, so the saves are checked against the rows they cascade to + deletes = by_outcome[DeserializeOutcome.DELETE] + if deletes: + self.cascaded_keys |= delete_app_models(deletes[0].model, [task.id for task in deletes]) + clean.extend(deletes) + + propagated = ( + by_outcome[DeserializeOutcome.PROPAGATE_DELETE] + + by_outcome[DeserializeOutcome.PROPAGATE_HARD_DELETE] + ) + record_propagated_deletions(propagated) + clean.extend(propagated) + + self._save_all(by_outcome[DeserializeOutcome.SAVE], clean, errored) + self._write_stores(clean, errored) + + def _save_all( + self, + tasks: List[DeserializeTask], + clean: List[DeserializeTask], + errored: List[DeserializeTask], + ) -> None: + """Saves the tasks' app models in arrival order, adding each task to `clean` or `errored`""" + if not tasks: + return + with mute_signals(signals.pre_save, signals.post_save): + for task in tasks: + if self._save(task): + clean.append(task) + else: + errored.append(task) + + def _write_stores(self, clean: List[DeserializeTask], errored: List[DeserializeTask]) -> None: + """Clears the store records that were written, and saves the errors of those that weren't""" + if clean: + Store.objects.filter(id__in=[task.id for task in clean]).update( + dirty_bit=False, deserialization_error=None, deserialization_exception=None + ) + # errors are rare, and `bulk_update` is slower than saving each record + for task in errored: + task.store.save(update_fields=["deserialization_error", "deserialization_exception"]) + + for task in clean: + self.counts[task.outcome] += 1 + if errored: + self.counts[DeserializeOutcome.ERROR] += len(errored) + + def summary(self) -> Dict[str, int]: + """The number of tasks consumed per outcome, counting failed saves as errors""" + return dict(self.counts) + + def _save(self, task: DeserializeTask) -> bool: + """Saves the task's app model in its own savepoint, recording the error if it fails""" + try: + with transaction.atomic(): + task.app_model.save(update_dirty_bit_to=False) + except ( + exceptions.ValidationError, + exceptions.ObjectDoesNotExist, + ValueError, + IntegrityError, + ) as e: + self._record_error(task, e) + return False + return True + + def _record_error(self, task: DeserializeTask, error: Exception) -> None: + task.store.set_deserialization_error(error) + self.failed_ids.add(task.id) + log_deserialization_error(task, error) + + +def log_deserialization_error(task: DeserializeTask, error: Exception) -> None: + """Logs a record that failed to deserialize, with the message used by legacy deserialization""" + logger.warning( + "Failed to deserialize Store record %s for %s: %s", + task.id, + task.model.__name__, + error, + ) + + +def store_model_partition_fn(task: DeserializeTask) -> str: + return task.store.model_name diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 80393cfe..42b38b33 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -25,14 +25,17 @@ from morango.models.core import Store from morango.models.core import SyncableModel from morango.sync.operations import _deserialize_from_store +from morango.sync.stream.core import Buffer from morango.sync.stream.core import Sink from morango.sync.stream.deserialize import AppModelDeserialize from morango.sync.stream.deserialize import AppModelValidate from morango.sync.stream.deserialize import delete_app_models from morango.sync.stream.deserialize import DeserializeOutcome +from morango.sync.stream.deserialize import DeserializeSink from morango.sync.stream.deserialize import DeserializeTask from morango.sync.stream.deserialize import record_propagated_deletions from morango.sync.stream.deserialize import row_key +from morango.sync.stream.deserialize import store_model_partition_fn from morango.sync.stream.deserialize import StoreModelSource from morango.utils import exception_path @@ -1274,3 +1277,191 @@ def test_full_chunk(self): self.assertEqual(DeletedModels.objects.count(), 500) self.assertEqual(HardDeletedModels.objects.count(), 500) + + +class ListSink(Sink[DeserializeTask]): + """Collects the tasks in stream order""" + + def __init__(self): + self.tasks = [] + + def consume(self, task: DeserializeTask) -> None: + self.tasks.append(task) + + +class DeserializeSinkTestCase(DeserializeFixturesMixin, TestCase): + logger_name = "morango.sync.stream.deserialize" + + def _transform(self): + """The dirty store records as tasks with resolved outcomes, in stream order""" + sink = ListSink() + ( + StoreModelSource(self.profile) + .pipe(AppModelDeserialize()) + .pipe(AppModelValidate()) + .end(sink) + ) + return sink.tasks + + def _run(self, sink=None): + sink = sink or DeserializeSink() + ( + StoreModelSource(self.profile) + .pipe(AppModelDeserialize()) + .pipe(AppModelValidate()) + .pipe(Buffer(size=500, partition_fn=store_model_partition_fn)) + .end(sink) + ) + return sink + + def _assert_store(self, store_id, dirty_bit, exception=None): + store = Store.objects.get(id=store_id) + self.assertEqual(store.dirty_bit, dirty_bit) + self.assertEqual(store.deserialization_exception, exception) + if exception is None: + self.assertIsNone(store.deserialization_error) + else: + self.assertTrue(store.deserialization_error) + + def _app_user(self, username="user"): + user = MyUser(username=username, password="password") + user.save(update_dirty_bit_to=False) + self._store(user, dirty_bit=False) + return user + + def test_save(self): + user = self._user(deserialization_error="old", deserialization_exception="old.Error") + + self._run() + + self.assertFalse(MyUser.objects.get(id=user.id)._morango_dirty_bit) + self._assert_store(user.id, dirty_bit=False) + + def test_save__failure(self): + users = [ + self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) + for _ in range(2) + ] + + with self.assertLogs(self.logger_name, "WARNING") as logs: + self._run() + + saved = MyUser.objects.get(username="same") + (failed,) = [user for user in users if user.id != saved.id] + self._assert_store(saved.id, dirty_bit=False) + self._assert_store(failed.id, dirty_bit=True, exception="django.db.utils.IntegrityError") + self.assertEqual(len(logs.output), 1) + self.assertIn(failed.id, logs.output[0]) + + def test_save__signals_muted(self): + self._user() + receiver = mock.Mock() + signals.pre_save.connect(receiver) + signals.post_save.connect(receiver) + self.addCleanup(signals.pre_save.disconnect, receiver) + self.addCleanup(signals.post_save.disconnect, receiver) + + self._run() + + receiver.assert_not_called() + + def test_delete(self): + user = self._app_user() + Store.objects.filter(id=user.id).update(deleted=True, dirty_bit=True) + + self._run() + + self.assertFalse(MyUser.objects.filter(id=user.id).exists()) + self._assert_store(user.id, dirty_bit=False) + self.assertFalse(DeletedModels.objects.exists()) + + def test_delete__app_row_missing(self): + user = self._user(deleted=True) + + self._run() + + self._assert_store(user.id, dirty_bit=False) + + def test_propagate_delete(self): + user = self._user(dirty_bit=False, deleted=True) + log = self._summary_log(user.id) + + self._run() + + self.assertFalse(SummaryLog.objects.exists()) + self._assert_store(log.id, dirty_bit=False) + self.assertEqual(list(DeletedModels.objects.values_list("id", flat=True)), [log.id]) + self.assertFalse(HardDeletedModels.objects.exists()) + + def test_propagate_hard_delete(self): + user = self._user(dirty_bit=False, deleted=True, hard_deleted=True) + log = self._summary_log(user.id) + + self._run() + + self.assertFalse(SummaryLog.objects.exists()) + self._assert_store(log.id, dirty_bit=False) + self.assertEqual(list(DeletedModels.objects.values_list("id", flat=True)), [log.id]) + self.assertEqual(list(HardDeletedModels.objects.values_list("id", flat=True)), [log.id]) + + def test_error(self): + facility = self._facility(serialized="{bad") + + with self.assertLogs(self.logger_name, "WARNING") as logs: + self._run() + + self.assertFalse(Facility.objects.exists()) + self._assert_store(facility.id, dirty_bit=True, exception="json.decoder.JSONDecodeError") + self.assertIn(facility.id, logs.output[0]) + + def test_summary(self): + self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) + self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) + deleted_user = self._user(deleted=True) + self._summary_log(deleted_user.id) + self._facility(serialized="{bad") + + with self.assertLogs(self.logger_name, "WARNING"): + sink = self._run() + + self.assertEqual( + sink.summary(), + { + DeserializeOutcome.SAVE: 1, + DeserializeOutcome.ERROR: 2, + DeserializeOutcome.DELETE: 1, + DeserializeOutcome.PROPAGATE_DELETE: 1, + }, + ) + + def _count_queries(self, sink, tasks): + with CaptureQueriesContext(connection) as context: + sink.consume(tasks) + return len( + [ + query + for query in context.captured_queries + if not query["sql"].upper().startswith(("SAVEPOINT", "RELEASE SAVEPOINT")) + ] + ) + + def _batched_chunk(self, deleted_user_id, size): + """A chunk of one deleted log and `size` logs that propagate a deletion""" + self._summary_log(deleted_user_id, deleted=True) + for _ in range(size): + self._summary_log(deleted_user_id) + tasks = self._transform() + self.assertEqual(len(tasks), size + 1) + return tasks + + def test_consume__batched(self): + deleted_user = self._user(dirty_bit=False, deleted=True) + sink = DeserializeSink() + + small = self._count_queries(sink, self._batched_chunk(deleted_user.id, 10)) + large = self._count_queries(sink, self._batched_chunk(deleted_user.id, 500)) + + # SQLite splits the 500 `DeletedModels` inserts into batches of 499 + self.assertEqual(large - small, 1 if connection.vendor == "sqlite" else 0) + self.assertEqual(DeletedModels.objects.count(), 510) + self.assertFalse(Store.objects.filter(dirty_bit=True).exists()) From 97067303bb3c006fc369e49e382003a6299dc3e7 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:26:37 -0700 Subject: [PATCH 08/13] Guard sink saves against FK targets that failed or were cascade-deleted FK constraints are deferred, so saving a record whose target the sink failed or removed earlier in the run would only fail at commit. Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 19 ++- .../tests/sync/stream/test_deserialize.py | 108 +++++++++++++++++- 2 files changed, 125 insertions(+), 2 deletions(-) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index f65799f5..6cc4ccfd 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -480,11 +480,28 @@ def _save_all( return with mute_signals(signals.pre_save, signals.post_save): for task in tasks: - if self._save(task): + error = self._removed_target_error(task) + if error is not None: + self._record_error(task, error) + errored.append(task) + elif self._save(task): clean.append(task) else: errored.append(task) + def _removed_target_error(self, task: DeserializeTask) -> Optional[Exception]: + """ + The error for a task whose FK target was removed after the task was validated, because the + target failed in the sink or a delete cascaded to it. FK constraints are deferred, so saving + the task would only fail when the transaction commits. + """ + for field, target_id in fk_references(task.app_model): + if target_id in self.failed_ids: + return failed_target_error(task, field, target_id) + if row_key(field.related_model, target_id) in self.cascaded_keys: + return deleted_target_error(field, target_id) + return None + def _write_stores(self, clean: List[DeserializeTask], errored: List[DeserializeTask]) -> None: """Clears the store records that were written, and saves the errors of those that weren't""" if clean: diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 42b38b33..14a72305 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -1289,7 +1289,9 @@ def consume(self, task: DeserializeTask) -> None: self.tasks.append(task) -class DeserializeSinkTestCase(DeserializeFixturesMixin, TestCase): +class DeserializeSinkFixturesMixin(DeserializeFixturesMixin): + """Runs the transforms into `DeserializeSink`, and checks what it wrote""" + logger_name = "morango.sync.stream.deserialize" def _transform(self): @@ -1329,6 +1331,8 @@ def _app_user(self, username="user"): self._store(user, dirty_bit=False) return user + +class DeserializeSinkTestCase(DeserializeSinkFixturesMixin, TestCase): def test_save(self): user = self._user(deserialization_error="old", deserialization_exception="old.Error") @@ -1465,3 +1469,105 @@ def test_consume__batched(self): self.assertEqual(large - small, 1 if connection.vendor == "sqlite" else 0) self.assertEqual(DeletedModels.objects.count(), 510) self.assertFalse(Store.objects.filter(dirty_bit=True).exists()) + + +class DeserializeSinkGuardsTestCase(DeserializeSinkFixturesMixin, TestCase): + """Saves whose FK target the sink failed or removed after the save was validated""" + + def _app_facility(self, name, parent=None): + facility = Facility(name=name, parent=parent) + facility.save(update_dirty_bit_to=False) + self._store( + facility, + dirty_bit=False, + _self_ref_fk=facility.parent_id or "", + _self_ref_order=0 if parent is None else 1, + ) + return facility + + def _error(self, store_id): + return Store.objects.get(id=store_id).deserialization_error + + def test_failed_target__self_ref(self): + parent = self._app_facility("parent") + child = self._facility(parent=parent, order=1) + tasks = self._transform() + sink = DeserializeSink() + sink.failed_ids.add(parent.id) + + with self.assertLogs(self.logger_name, "WARNING"): + sink.consume(tasks) + + self.assertFalse(Facility.objects.filter(id=child.id).exists()) + self._assert_store(child.id, dirty_bit=True, exception=MorangoDirtyParent.path()) + self.assertIn(child.id, sink.failed_ids) + + def test_failed_target__cross_model(self): + user = self._app_user() + log = self._summary_log(user.id) + tasks = self._transform() + sink = DeserializeSink() + sink.failed_ids.add(user.id) + + with self.assertLogs(self.logger_name, "WARNING"): + sink.consume(tasks) + + self.assertFalse(SummaryLog.objects.exists()) + self._assert_store( + log.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" + ) + self.assertIn(user.id, self._error(log.id)) + self.assertIn(log.id, sink.failed_ids) + + def test_failed_save(self): + """The log was validated against the user before the user's save failed""" + self._app_user(username="same") + user = self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) + log = self._summary_log(user.id) + + with self.assertLogs(self.logger_name, "WARNING"): + self._run() + + self._assert_store(user.id, dirty_bit=True, exception="django.db.utils.IntegrityError") + self.assertFalse(SummaryLog.objects.exists()) + self._assert_store( + log.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" + ) + self.assertIn("failed to deserialize", self._error(log.id)) + + def _cascade_fixture(self): + """A deleted root, its clean child, and a dirty grandchild validated against the child""" + root = self._app_facility("root") + child = self._app_facility("child", parent=root) + Store.objects.filter(id=root.id).update(deleted=True, dirty_bit=True) + grandchild = self._facility(parent=child, order=2) + tasks = {task.id: task for task in self._transform()} + self.assertEqual(tasks[grandchild.id].outcome, DeserializeOutcome.SAVE) + return root, child, grandchild, tasks + + def _assert_cascaded(self, child, grandchild): + self.assertFalse(Facility.objects.exists()) + self._assert_store( + grandchild.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" + ) + self.assertIn("was deleted", self._error(grandchild.id)) + self.assertIn(child.id, self._error(grandchild.id)) + + def test_cascaded_target__same_chunk(self): + root, child, grandchild, tasks = self._cascade_fixture() + sink = DeserializeSink() + + with self.assertLogs(self.logger_name, "WARNING"): + sink.consume([tasks[root.id], tasks[grandchild.id]]) + + self._assert_cascaded(child, grandchild) + + def test_cascaded_target__later_chunk(self): + root, child, grandchild, tasks = self._cascade_fixture() + sink = DeserializeSink() + + sink.consume([tasks[root.id]]) + with self.assertLogs(self.logger_name, "WARNING"): + sink.consume([tasks[grandchild.id]]) + + self._assert_cascaded(child, grandchild) From f98268b40a70f4b4e9612a7b4f78d0da85d34688 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:28:57 -0700 Subject: [PATCH 09/13] Add streaming deserialize_from_store pipeline entry point Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/stream/deserialize.py | 30 ++++ .../tests/sync/stream/test_deserialize.py | 163 ++++++++++++++++-- 2 files changed, 182 insertions(+), 11 deletions(-) diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index 6cc4ccfd..26c8b8e5 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -27,6 +27,8 @@ from morango.models.core import Store from morango.models.core import SyncableModel from morango.registry import syncable_models +from morango.sync.db import begin_transaction +from morango.sync.stream.core import Buffer from morango.sync.stream.core import Sink from morango.sync.stream.core import Transform from morango.sync.stream.source import MorangoSource @@ -554,3 +556,31 @@ def log_deserialization_error(task: DeserializeTask, error: Exception) -> None: def store_model_partition_fn(task: DeserializeTask) -> str: return task.store.model_name + + +CHUNK_SIZE = 500 + + +def deserialize_from_store( + profile: str, + sync_filter: Optional[Filter] = None, + skip_errored: bool = False, +) -> None: + """ + Constructs and executes the deserialization pipeline, streaming dirty store records through + transforms that resolve each record's outcome, into a sink that writes them in chunks. + + :param profile: The Morango model profile + :param sync_filter: The Filter object for this sync + :param skip_errored: Whether to skip Store records with deserialization errors + """ + with begin_transaction(sync_filter, isolated=True): + sink = DeserializeSink() + ( + StoreModelSource(profile, sync_filter=sync_filter, skip_errored=skip_errored) + .pipe(AppModelDeserialize(sync_filter=sync_filter)) + .pipe(AppModelValidate(sync_filter=sync_filter)) + .pipe(Buffer(size=CHUNK_SIZE, partition_fn=store_model_partition_fn)) + .end(sink) + ) + logger.info("Deserialization done: %s", sink.summary()) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 14a72305..2b15758f 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -10,6 +10,7 @@ from django.db.utils import IntegrityError from django.test import SimpleTestCase from django.test import TestCase +from django.test import TransactionTestCase from django.test.utils import CaptureQueriesContext from facility_profile.models import ConditionalLog from facility_profile.models import Facility @@ -30,6 +31,7 @@ from morango.sync.stream.deserialize import AppModelDeserialize from morango.sync.stream.deserialize import AppModelValidate from morango.sync.stream.deserialize import delete_app_models +from morango.sync.stream.deserialize import deserialize_from_store from morango.sync.stream.deserialize import DeserializeOutcome from morango.sync.stream.deserialize import DeserializeSink from morango.sync.stream.deserialize import DeserializeTask @@ -1331,6 +1333,17 @@ def _app_user(self, username="user"): self._store(user, dirty_bit=False) return user + def _app_facility(self, name, parent=None): + facility = Facility(name=name, parent=parent) + facility.save(update_dirty_bit_to=False) + self._store( + facility, + dirty_bit=False, + _self_ref_fk=facility.parent_id or "", + _self_ref_order=0 if parent is None else 1, + ) + return facility + class DeserializeSinkTestCase(DeserializeSinkFixturesMixin, TestCase): def test_save(self): @@ -1474,17 +1487,6 @@ def test_consume__batched(self): class DeserializeSinkGuardsTestCase(DeserializeSinkFixturesMixin, TestCase): """Saves whose FK target the sink failed or removed after the save was validated""" - def _app_facility(self, name, parent=None): - facility = Facility(name=name, parent=parent) - facility.save(update_dirty_bit_to=False) - self._store( - facility, - dirty_bit=False, - _self_ref_fk=facility.parent_id or "", - _self_ref_order=0 if parent is None else 1, - ) - return facility - def _error(self, store_id): return Store.objects.get(id=store_id).deserialization_error @@ -1571,3 +1573,142 @@ def test_cascaded_target__later_chunk(self): sink.consume([tasks[grandchild.id]]) self._assert_cascaded(child, grandchild) + + +class DeserializeFromStoreTestCase(DeserializeSinkFixturesMixin, TransactionTestCase): + """ + Runs the whole pipeline in its own transaction, which commits at the end of each run. FK + constraints are deferred, so a save that breaks one would fail the commit. + """ + + def setUp(self): + patcher = mock.patch.object(Facility, "clean_dereferences_parent", False) + patcher.start() + self.addCleanup(patcher.stop) + + def _deserialize(self, **kwargs): + with self.assertLogs(self.logger_name, "INFO") as logs: + deserialize_from_store(self.profile, **kwargs) + return logs + + def test_failed_parent_save__cross_model(self): + self._app_user(username="same") + user = self._store(MyUser(id=uuid.uuid4().hex, username="same", password="password")) + log = self._summary_log(user.id) + + self._deserialize() + + self._assert_store(user.id, dirty_bit=True, exception="django.db.utils.IntegrityError") + self._assert_store( + log.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" + ) + self.assertFalse(MyUser.objects.filter(id=user.id).exists()) + self.assertFalse(SummaryLog.objects.exists()) + + def test_failed_parent_save__self_ref(self): + root = self._facility() + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + save = Facility.save + + def failing_save(facility, *args, **kwargs): + if facility.id == root.id: + raise IntegrityError("root failed") + return save(facility, *args, **kwargs) + + with mock.patch.object(Facility, "save", failing_save): + self._deserialize() + + self.assertFalse(Facility.objects.exists()) + self._assert_store(root.id, dirty_bit=True, exception="django.db.utils.IntegrityError") + self._assert_store(child.id, dirty_bit=True, exception=MorangoDirtyParent.path()) + self._assert_store(grandchild.id, dirty_bit=True, exception=MorangoDirtyParent.path()) + + def _cascade_fixture(self): + """A deleted root, its clean child, and a dirty grandchild""" + root = self._app_facility("root") + child = self._app_facility("child", parent=root) + Store.objects.filter(id=root.id).update(deleted=True, dirty_bit=True) + grandchild = self._facility(parent=child, order=2) + return root, child, grandchild + + def _assert_cascade(self, root, grandchild): + self._deserialize() + + self.assertFalse(Facility.objects.exists()) + self._assert_store(root.id, dirty_bit=False) + self._assert_store( + grandchild.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" + ) + self.assertIn("was deleted", Store.objects.get(id=grandchild.id).deserialization_error) + + # the next run fails the grandchild in validation, since the child's row is gone + self._deserialize() + + self._assert_store( + grandchild.id, dirty_bit=True, exception="django.core.exceptions.ValidationError" + ) + self.assertNotIn("was deleted", Store.objects.get(id=grandchild.id).deserialization_error) + + def test_cascaded_target__same_chunk(self): + root, _, grandchild = self._cascade_fixture() + self._assert_cascade(root, grandchild) + + @mock.patch("morango.sync.stream.deserialize.CHUNK_SIZE", 1) + def test_cascaded_target__later_chunk(self): + """The grandchild is validated before the root's chunk is written""" + root, _, grandchild = self._cascade_fixture() + self._assert_cascade(root, grandchild) + + def test_self_ref_chain(self): + root = self._facility() + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + + self._deserialize() + + self.assertEqual(Facility.objects.count(), 3) + for facility in (root, child, grandchild): + self._assert_store(facility.id, dirty_bit=False) + + def test_self_ref_chain__dereferences_parent(self): + """A child can't load a parent that is in the same chunk, so each run saves one level""" + root = self._facility() + child = self._facility(parent=root, order=1) + grandchild = self._facility(parent=child, order=2) + + with mock.patch.object(Facility, "clean_dereferences_parent", True): + for saved in ([root], [root, child], [root, child, grandchild]): + self._deserialize() + self.assertEqual( + set(Facility.objects.values_list("id", flat=True)), + {facility.id for facility in saved}, + ) + + self.assertFalse(Store.objects.filter(dirty_bit=True).exists()) + + def test_skip_errored(self): + user = self._user(deserialization_error="old", deserialization_exception="old.Error") + + self._deserialize(skip_errored=True) + + self._assert_store(user.id, dirty_bit=True, exception="old.Error") + self.assertFalse(MyUser.objects.exists()) + + def test_sync_filter(self): + included = self._user() + excluded = self._user() + + self._deserialize(sync_filter=Filter(included.partition)) + + self.assertEqual(list(MyUser.objects.values_list("id", flat=True)), [included.id]) + self._assert_store(excluded.id, dirty_bit=True) + + def test_summary_logged(self): + self._user() + self._facility(serialized="{bad") + + logs = self._deserialize() + + self.assertIn("'save': 1", logs.output[-1]) + self.assertIn("'error': 1", logs.output[-1]) From c839a54ea329a8f2216d3192e48ee12e7005ae96 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:29:29 -0700 Subject: [PATCH 10/13] Freeze legacy deserialization parity as literal expectations Co-Authored-By: Claude Opus 5.5 (1M context) --- .../tests/sync/stream/test_deserialize.py | 102 ++++++++++-------- 1 file changed, 59 insertions(+), 43 deletions(-) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 2b15758f..8f149c49 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -39,7 +39,6 @@ from morango.sync.stream.deserialize import row_key from morango.sync.stream.deserialize import store_model_partition_fn from morango.sync.stream.deserialize import StoreModelSource -from morango.utils import exception_path from ...helpers import StoreFactory @@ -1050,112 +1049,126 @@ def test_transforms_only_read(self): class LegacyParityTestCase(DeserializeFixturesMixin, TestCase): """ - Characterizes legacy `_deserialize_from_store` errors and propagated deletions against the - transforms, on the same store records. The transforms don't write, so both can run against the - same database state. Runs with the default `Facility.clean_dereferences_parent`, which is the - legacy test behavior, unless a test patches it. + Characterizes the end state that legacy `_deserialize_from_store` stores for errors and + propagated deletions. The expected values were captured from legacy, with the default + `Facility.clean_dereferences_parent` at the time patched in where a test relies on it. """ - def _legacy_result(self, store_id): - """The legacy result of a store record that did not save, in the transforms' terms""" - if HardDeletedModels.objects.filter(id=store_id).exists(): - return DeserializeOutcome.PROPAGATE_HARD_DELETE, None - if DeletedModels.objects.filter(id=store_id).exists(): - return DeserializeOutcome.PROPAGATE_DELETE, None - exception = Store.objects.get(id=store_id).deserialization_exception - self.assertIsNotNone(exception) - return DeserializeOutcome.ERROR, exception - - def _assert_parity(self, *store_ids): - results = self._run() + def _deserialize(self): _deserialize_from_store(self.profile) - for store_id in store_ids: - with self.subTest(store_id=store_id): - outcome, error_type = results[store_id] - error_path = exception_path(error_type) if error_type else None - self.assertEqual((outcome, error_path), self._legacy_result(store_id)) + def _assert_stored(self, store_id, dirty_bit, exception, deleted=False, hard_deleted=False): + store = Store.objects.get(id=store_id) + self.assertEqual( + ( + store.dirty_bit, + store.deserialization_exception, + DeletedModels.objects.filter(id=store_id).exists(), + HardDeletedModels.objects.filter(id=store_id).exists(), + ), + (dirty_bit, exception, deleted, hard_deleted), + ) + @mock.patch.object(Facility, "clean_dereferences_parent", True) def test_parent_failed(self): root = self._facility(serialized="{bad") child = self._facility(parent=root, order=1) grandchild = self._facility(parent=child, order=2) - self._assert_parity(root.id, child.id, grandchild.id) + self._deserialize() + + self._assert_stored(root.id, True, "json.decoder.JSONDecodeError") + self._assert_stored(child.id, True, MorangoDirtyParent.path()) + self._assert_stored(grandchild.id, True, MorangoDirtyParent.path()) + @mock.patch.object(Facility, "clean_dereferences_parent", True) def test_parent_missing(self): missing_parent = Facility(id=uuid.uuid4().hex, name="missing") child = self._facility(parent=missing_parent) - self._assert_parity(child.id) + self._deserialize() + + self._assert_stored(child.id, True, MorangoMissingParent.path()) def test_target_missing(self): log = self._summary_log(uuid.uuid4().hex) - self._assert_parity(log.id) + self._deserialize() + + self._assert_stored(log.id, True, "django.db.utils.IntegrityError") @mock.patch.object(Facility, "clean_dereferences_parent", False) def test_own_error_with_parent_accepted(self): root = self._facility() child = self._facility(parent=root, order=1, name="x" * 200) - self._assert_parity(child.id) + self._deserialize() + + self._assert_stored(child.id, True, "django.core.exceptions.ValidationError") def test_target_deleted_in_run(self): user = self._user(deleted=True) log = self._summary_log(user.id) - self._assert_parity(log.id) + self._deserialize() + + self._assert_stored(log.id, True, None, deleted=True) def test_target_deleted_earlier(self): user = self._user(dirty_bit=False, deleted=True) log = self._summary_log(user.id) - self._assert_parity(log.id) + self._deserialize() + + self._assert_stored(log.id, True, None, deleted=True) def test_target_hard_deleted_earlier(self): user = self._user(dirty_bit=False, deleted=True, hard_deleted=True) log = self._summary_log(user.id) - self._assert_parity(log.id) + self._deserialize() + + self._assert_stored(log.id, True, None, deleted=True, hard_deleted=True) + @mock.patch.object(Facility, "clean_dereferences_parent", True) def test_parent_deleted_earlier(self): root = self._facility(dirty_bit=False, deleted=True) child = self._facility(parent=root, order=1) - self._assert_parity(child.id) + self._deserialize() + + self._assert_stored(child.id, False, None, deleted=True) + @mock.patch.object(Facility, "clean_dereferences_parent", True) def test_parent_hard_deleted_earlier(self): root = self._facility(dirty_bit=False, deleted=True, hard_deleted=True) child = self._facility(parent=root, order=1) - self._assert_parity(child.id) + self._deserialize() + + self._assert_stored(child.id, False, None, deleted=True, hard_deleted=True) @mock.patch.object(Facility, "clean_dereferences_parent", False) def test_parent_deleted_earlier__parent_not_loaded(self): root = self._facility(dirty_bit=False, deleted=True) child = self._facility(parent=root, order=1) - self._assert_parity(child.id) + self._deserialize() + + self._assert_stored(child.id, True, "django.core.exceptions.ValidationError") - def test_dirty_parent_not_in_run__differs(self): + def test_dirty_parent_not_in_run(self): """ - Characterizes an intentional difference: legacy never deserializes children of a parent - whose store record is dirty. The transforms save the child, which validates against the - parent's older app row. + Legacy never deserializes children of a parent whose store record is dirty, and records + `MorangoMissingParent` when the parent is left out of the run. """ parent = Facility.objects.create(name="parent") self._store(parent, _self_ref_order=0, deserialization_error="failed earlier") child = self._facility(parent=parent, order=1) - results = self._run(skip_errored=True) _deserialize_from_store(self.profile, skip_erroring=True) - self.assertEqual(results[child.id], (DeserializeOutcome.SAVE, None)) - self.assertEqual( - self._legacy_result(child.id), - (DeserializeOutcome.ERROR, exception_path(MorangoMissingParent)), - ) + self._assert_stored(child.id, True, MorangoMissingParent.path()) @mock.patch.object(Facility, "clean_dereferences_parent", False) def test_parent_deleted_in_run__parent_not_loaded(self): @@ -1163,7 +1176,10 @@ def test_parent_deleted_in_run__parent_not_loaded(self): child = self._facility(parent=root, order=1) grandchild = self._facility(parent=child, order=2) - self._assert_parity(child.id, grandchild.id) + self._deserialize() + + self._assert_stored(child.id, True, "django.core.exceptions.ValidationError") + self._assert_stored(grandchild.id, True, MorangoDirtyParent.path()) class DeleteAppModelsTestCase(TestCase): From 1f4ffaffa35511bc42ee52c9317ac3296a33e46e Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:35:06 -0700 Subject: [PATCH 11/13] Switch deserialization callers to the streaming pipeline Defaults the test Facility to not load its parent during validation, like models that don't load related objects. With sink lag, a dirty chain of models that do takes one run per level, so the test that deserializes a whole chain at once relies on the new default. Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/controller.py | 6 ++-- morango/sync/operations.py | 5 ++-- tests/testapp/facility_profile/models.py | 5 ++-- .../tests/sync/stream/test_deserialize.py | 29 ++++++++++++------- tests/testapp/tests/sync/test_controller.py | 2 +- tests/testapp/tests/sync/test_operations.py | 20 ++++++------- 6 files changed, 39 insertions(+), 28 deletions(-) diff --git a/morango/sync/controller.py b/morango/sync/controller.py index 5fe180a8..f0e3f911 100644 --- a/morango/sync/controller.py +++ b/morango/sync/controller.py @@ -5,8 +5,8 @@ from morango.constants import transfer_stages from morango.constants import transfer_statuses from morango.registry import session_middleware -from morango.sync.operations import _deserialize_from_store from morango.sync.operations import OperationLogger +from morango.sync.stream.deserialize import deserialize_from_store from morango.sync.stream.serialize import serialize_into_store from morango.sync.utils import SyncSignalGroup from morango.utils import _assert @@ -33,7 +33,9 @@ def deserialize_from_store(self, skip_erroring=False, sync_filter=None): with OperationLogger("Deserializing records", "Deserialization complete"): # we first serialize to avoid deserialization merge conflicts serialize_into_store(self.profile, sync_filter=sync_filter) - _deserialize_from_store(self.profile, filter=sync_filter, skip_erroring=skip_erroring) + deserialize_from_store( + self.profile, sync_filter=sync_filter, skip_errored=skip_erroring + ) def create_network_connection(self, base_url, **kwargs): from morango.sync.syncsession import NetworkSyncConnection diff --git a/morango/sync/operations.py b/morango/sync/operations.py index f8452443..1d4a3e2b 100644 --- a/morango/sync/operations.py +++ b/morango/sync/operations.py @@ -56,6 +56,7 @@ from morango.sync.context import LocalSessionContext from morango.sync.context import NetworkSessionContext from morango.sync.db import begin_transaction +from morango.sync.stream.deserialize import deserialize_from_store from morango.sync.stream.serialize import serialize_into_store from morango.sync.utils import mute_signals from morango.sync.utils import validate_and_create_buffer_data @@ -1166,8 +1167,8 @@ def handle(self, context): serialize_into_store( context.sync_session.profile, sync_filter=context.filter ) - _deserialize_from_store( - context.sync_session.profile, filter=context.filter + deserialize_from_store( + context.sync_session.profile, sync_filter=context.filter ) except OperationalError as e: # if we run into a transaction isolation error, we return a pending status to force diff --git a/tests/testapp/facility_profile/models.py b/tests/testapp/facility_profile/models.py index 37457471..706f58af 100644 --- a/tests/testapp/facility_profile/models.py +++ b/tests/testapp/facility_profile/models.py @@ -25,8 +25,9 @@ class Facility(FacilityDataSyncableModel): morango_model_name = "facility" # when True, clean_fields loads the parent object, which raises ObjectDoesNotExist if the - # parent row is not in the app table; used to test non-validation error handling - clean_dereferences_parent = True + # parent row is not in the app table; used to test non-validation error handling. Defaults to + # False like models that don't load related objects, and tests patch it to cover those that do + clean_dereferences_parent = False name = models.CharField(max_length=100) now_date = models.DateTimeField(default=timezone.now) diff --git a/tests/testapp/tests/sync/stream/test_deserialize.py b/tests/testapp/tests/sync/stream/test_deserialize.py index 8f149c49..1f5ea2b9 100644 --- a/tests/testapp/tests/sync/stream/test_deserialize.py +++ b/tests/testapp/tests/sync/stream/test_deserialize.py @@ -25,7 +25,6 @@ from morango.models.core import HardDeletedModels from morango.models.core import Store from morango.models.core import SyncableModel -from morango.sync.operations import _deserialize_from_store from morango.sync.stream.core import Buffer from morango.sync.stream.core import Sink from morango.sync.stream.deserialize import AppModelDeserialize @@ -1047,15 +1046,21 @@ def test_transforms_only_read(self): ) -class LegacyParityTestCase(DeserializeFixturesMixin, TestCase): +class DeserializeParityTestCase(DeserializeFixturesMixin, TestCase): """ - Characterizes the end state that legacy `_deserialize_from_store` stores for errors and - propagated deletions. The expected values were captured from legacy, with the default + Checks the end state that `deserialize_from_store` stores for errors and propagated deletions + against the values captured from legacy `_deserialize_from_store`, with the default `Facility.clean_dereferences_parent` at the time patched in where a test relies on it. + + The exceptions are: + - A record that propagates the deletion of its FK target, which legacy left dirty. It now ends + clean, since its deletion is tracked. + - A child of a parent that is dirty but left out of the run, which legacy failed with + `MorangoMissingParent`. It is now saved against the parent's older app row. """ def _deserialize(self): - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) def _assert_stored(self, store_id, dirty_bit, exception, deleted=False, hard_deleted=False): store = Store.objects.get(id=store_id) @@ -1112,7 +1117,7 @@ def test_target_deleted_in_run(self): self._deserialize() - self._assert_stored(log.id, True, None, deleted=True) + self._assert_stored(log.id, False, None, deleted=True) def test_target_deleted_earlier(self): user = self._user(dirty_bit=False, deleted=True) @@ -1120,7 +1125,7 @@ def test_target_deleted_earlier(self): self._deserialize() - self._assert_stored(log.id, True, None, deleted=True) + self._assert_stored(log.id, False, None, deleted=True) def test_target_hard_deleted_earlier(self): user = self._user(dirty_bit=False, deleted=True, hard_deleted=True) @@ -1128,7 +1133,7 @@ def test_target_hard_deleted_earlier(self): self._deserialize() - self._assert_stored(log.id, True, None, deleted=True, hard_deleted=True) + self._assert_stored(log.id, False, None, deleted=True, hard_deleted=True) @mock.patch.object(Facility, "clean_dereferences_parent", True) def test_parent_deleted_earlier(self): @@ -1160,15 +1165,17 @@ def test_parent_deleted_earlier__parent_not_loaded(self): def test_dirty_parent_not_in_run(self): """ Legacy never deserializes children of a parent whose store record is dirty, and records - `MorangoMissingParent` when the parent is left out of the run. + `MorangoMissingParent` when the parent is left out of the run. The child is now saved, + since it validates against the parent's older app row. """ parent = Facility.objects.create(name="parent") self._store(parent, _self_ref_order=0, deserialization_error="failed earlier") child = self._facility(parent=parent, order=1) - _deserialize_from_store(self.profile, skip_erroring=True) + deserialize_from_store(self.profile, skip_errored=True) - self._assert_stored(child.id, True, MorangoMissingParent.path()) + self.assertTrue(Facility.objects.filter(id=child.id).exists()) + self._assert_stored(child.id, False, None) @mock.patch.object(Facility, "clean_dereferences_parent", False) def test_parent_deleted_in_run__parent_not_loaded(self): diff --git a/tests/testapp/tests/sync/test_controller.py b/tests/testapp/tests/sync/test_controller.py index 1de23018..f436a84b 100644 --- a/tests/testapp/tests/sync/test_controller.py +++ b/tests/testapp/tests/sync/test_controller.py @@ -614,7 +614,7 @@ def test_unique_violation_logs_warning(self): """A warning should be logged when deserialization fails due to a unique constraint.""" self._create_conflicting_user_store_records() - with self.assertLogs("morango.sync.operations", level="WARNING") as cm: + with self.assertLogs("morango.sync.stream.deserialize", level="WARNING") as cm: self.mc.deserialize_from_store() unique_warnings = [ diff --git a/tests/testapp/tests/sync/test_operations.py b/tests/testapp/tests/sync/test_operations.py index 2eb264d0..979c32c2 100644 --- a/tests/testapp/tests/sync/test_operations.py +++ b/tests/testapp/tests/sync/test_operations.py @@ -30,7 +30,6 @@ from morango.sync.controller import MorangoProfileController from morango.sync.controller import SessionController from morango.sync.operations import _dequeue_into_store -from morango.sync.operations import _deserialize_from_store from morango.sync.operations import _queue_into_buffer_v1 from morango.sync.operations import _queue_into_buffer_v2 from morango.sync.operations import _update_legacy_self_ref_order @@ -41,6 +40,7 @@ from morango.sync.operations import ReceiverDequeueOperation from morango.sync.operations import ReceiverDeserializeOperation from morango.sync.operations import ReceiverQueueOperation +from morango.sync.stream.deserialize import deserialize_from_store from morango.sync.syncsession import TransferClient from ..helpers import create_buffer_and_store_dummy_data @@ -1257,7 +1257,7 @@ def test_successful_deserialization(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization() @@ -1267,7 +1267,7 @@ def test_deserialization_with_missing_username(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization( user_deserialized=False, @@ -1282,7 +1282,7 @@ def test_deserialization_with_excessively_long_username(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization( user_deserialized=False, @@ -1297,7 +1297,7 @@ def test_deserialization_with_invalid_content_id(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization(log1_deserialized=False) @@ -1307,14 +1307,14 @@ def test_deserialization_with_invalid_log_user_id(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization(log1_deserialized=False) def test_deserialization__conditional__no_filter(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile) + deserialize_from_store(self.profile) self.assert_deserialization() conditional_log = ConditionalLog.objects.get(pk=self.serialized_conditional["id"]) @@ -1323,7 +1323,7 @@ def test_deserialization__conditional__no_filter(self): def test_deserialization__conditional__with_filter(self): self.serialize_all_to_store() - _deserialize_from_store(self.profile, filter=Filter(self.serialized_facility["id"])) + deserialize_from_store(self.profile, sync_filter=Filter(self.serialized_facility["id"])) self.assert_deserialization( user_deserialized=False, @@ -1341,7 +1341,7 @@ def test_deserialization__skip_erroring(self): ) conditional_log_store.save() - _deserialize_from_store(self.profile, skip_erroring=True) + deserialize_from_store(self.profile, skip_errored=True) self.assert_deserialization( conditional_deserialized=False, @@ -1353,6 +1353,6 @@ def test_deserialization__skip_erroring__empty_string(self): conditional_log_store.deserialization_error = "" conditional_log_store.save() - _deserialize_from_store(self.profile, skip_erroring=True) + deserialize_from_store(self.profile, skip_errored=True) self.assert_deserialization() From e6717187e8b70f490a96dd47517f1f43d42bc011 Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:38:39 -0700 Subject: [PATCH 12/13] Remove legacy deserialization from operations Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/sync/operations.py | 381 +------------------------------------ 1 file changed, 1 insertion(+), 380 deletions(-) diff --git a/morango/sync/operations.py b/morango/sync/operations.py index 1d4a3e2b..70da04d7 100644 --- a/morango/sync/operations.py +++ b/morango/sync/operations.py @@ -1,24 +1,13 @@ -import functools import itertools import json import logging import uuid -from collections import defaultdict -from django.core import exceptions from django.db import connection -from django.db import transaction -from django.db.models import CharField -from django.db.models import F from django.db.models import Exists from django.db.models import OuterRef -from django.db.models import Q from django.db.models import Subquery -from django.db.models import signals -from django.db.models import Value -from django.db.models.fields import BooleanField -from django.db.models.functions import NullIf, Cast -from django.db.utils import IntegrityError +from django.db.models.functions import Cast from django.db.utils import OperationalError from django.utils import timezone from rest_framework.exceptions import ValidationError @@ -29,17 +18,13 @@ from morango.constants.capabilities import ASYNC_OPERATIONS from morango.constants.capabilities import FSIC_V2_FORMAT from morango.constants.capabilities import SELF_REF_ORDER -from morango.errors import MorangoDirtyParent from morango.errors import MorangoInvalidFSICPartition from morango.errors import MorangoLimitExceeded -from morango.errors import MorangoMissingParent from morango.errors import MorangoResumeSyncError from morango.errors import MorangoSkipOperation from morango.models.certificates import Filter from morango.models.core import Buffer from morango.models.core import DatabaseMaxCounter -from morango.models.core import DeletedModels -from morango.models.core import HardDeletedModels from morango.models.core import InstanceIDModel from morango.models.core import RecordMaxCounter from morango.models.core import RecordMaxCounterBuffer @@ -52,17 +37,14 @@ from morango.models.fsic_utils import expand_fsic_for_use from morango.registry import syncable_models from morango.sync.backends.utils import load_backend -from morango.sync.backends.utils import TemporaryTable from morango.sync.context import LocalSessionContext from morango.sync.context import NetworkSessionContext from morango.sync.db import begin_transaction from morango.sync.stream.deserialize import deserialize_from_store from morango.sync.stream.serialize import serialize_into_store -from morango.sync.utils import mute_signals from morango.sync.utils import validate_and_create_buffer_data from morango.utils import _assert from morango.utils import self_referential_fk -from morango.utils import exception_path from morango.utils import SETTINGS logger = logging.getLogger(__name__) @@ -92,367 +74,6 @@ def _join_with_logical_operator(lst, operator): return "(({items}))".format(items=op.join(lst)) -def _validate_missing_store_foreign_keys(from_model_name, to_model_name, temp_table): - """ - Performs validation on a bulk set of foreign keys (FKs), given a temp table with two columns, - `from_pk` and `to_pk`, the primary key (PK) pair to validate. Any store record matching - `from_pk`, while missing a store record matching `to_pk`, will be updated with a deserialization - error and its PK returned within a list. - - :param from_model_name: A str name of the model which has the FK, for logging purposes - :param to_model_name: A str name of the model referenced by the FK, for logging purposes - :param temp_table: A temp table object for querying against in the DB - :type temp_table: morango.sync.backends.utils.TemporaryTable - :return: A list of store PKs that have broken FKs - """ - invalid_pks = [] - select_sql = """ - SELECT t.from_field, t.from_pk, t.to_pk - FROM {temp_table} t - WHERE NOT EXISTS ( - SELECT 1 - FROM {store} s - WHERE s.{pk_field} = t.to_pk - ) - """.format( - temp_table=temp_table.sql_name, - store=Store._meta.db_table, - pk_field=Store._meta.pk.column, - ) - - store_deserialization_error_fields = [ - f for f in Store._meta.fields if f.name in ("deserialization_error", "deserialization_exception") - ] - store_update_fields = [Store._meta.pk, *store_deserialization_error_fields] - - from_pk_field = temp_table.get_field("from_pk") - to_pk_field = temp_table.get_field("to_pk") - update_values = [] - with connection.cursor() as c: - c.execute(select_sql) - for from_field, from_pk, to_pk in c.fetchall(): - err = IntegrityError( - "{from_model_name}.{from_field} references non-existent {to_model_name} instance with id '{to_pk}'".format( - from_model_name=from_model_name, - from_field=from_field, - to_model_name=to_model_name, - to_pk=to_pk_field.to_python(to_pk), - ) - ) - logger.warning( - "Error deserializing instance with id {from_pk}: {err}".format( - from_pk=from_pk, - err=str(err), - ) - ) - update_values.extend([from_pk, str(err), exception_path(err)]) - invalid_pks.append(from_pk_field.to_python(from_pk)) - if update_values: - # update Store with errors - DBBackend._bulk_update( - c, Store._meta.db_table, store_update_fields, update_values - ) - return invalid_pks - - -def _handle_deleted_store_foreign_keys(temp_table): - """ - Handles store records with foreign key (FK) references to other deleted store records to prevent - their deserialization and update the `DeletedModels` and `HardDeletedModels` tracking, given a - temp table with two columns, `from_pk` and `to_pk`, the primary key (PK) pair - :param temp_table: A temp table object for querying against in the DB - :type temp_table: morango.sync.backends.utils.TemporaryTable - :return: A list of store PKs that have FKs to deleted records - """ - select_sql = """ - SELECT t.from_pk, s.profile, s.deleted, s.hard_deleted - FROM {temp_table} t - INNER JOIN {store} s ON s.{pk_field} = t.to_pk - WHERE s.deleted OR s.hard_deleted - """.format( - temp_table=temp_table.sql_name, - store=Store._meta.db_table, - pk_field=Store._meta.pk.column, - ) - - from_pk_field = temp_table.get_field("from_pk") - deleted_pks = [] - deleted_values = [] - hard_deleted_values = [] - with connection.cursor() as c: - c.execute(select_sql) - # find all the store PKs which have FKs to deleted store PKs - for from_pk, profile, deleted, hard_deleted in c.fetchall(): - if deleted or hard_deleted: - deleted_values.extend([from_pk, profile]) - if hard_deleted: - hard_deleted_values.extend([from_pk, profile]) - deleted_pks.append(from_pk_field.to_python(from_pk)) - # update deleted tracking models - if deleted_values: - DBBackend._bulk_full_record_upsert( - c, - DeletedModels._meta.db_table, - DeletedModels._meta.fields, - deleted_values, - ) - if hard_deleted_values: - DBBackend._bulk_full_record_upsert( - c, - HardDeletedModels._meta.db_table, - HardDeletedModels._meta.fields, - hard_deleted_values, - ) - return deleted_pks - - -def _validate_store_foreign_keys(from_model_name, fk_references): - """ - Validates the foreign key (FK) references of a model whose name is `from_model_name`, through - bulk processing using a temporary table within the database for holding the FK references - - :param from_model_name: A str name of the model which has the FKs, for logging purposes - :param fk_references: A dictionary of lists containing `morango.models.core.ForeignKeyReference` - keyed by the - :return: A tuple of two lists containing store PKs with broken FKs and FKs to deleted records - """ - exclude_pks = [] - deleted_pks = [] - - for to_model_name, to_fk_references in fk_references.items(): - with TemporaryTable( - connection, - "fks", - from_field=CharField(max_length=255), - from_pk=UUIDField(), - to_pk=UUIDField(), - ) as temp_table: - # insert all the FK references into a temp table in the database - temp_table.bulk_insert([fks._asdict() for fks in to_fk_references]) - # now pass the temp table to validate against broken FKs - exclude_pks.extend( - _validate_missing_store_foreign_keys( - from_model_name, to_model_name, temp_table - ) - ) - # find any FKs referencing deleted records - deleted_pks.extend(_handle_deleted_store_foreign_keys(temp_table)) - - return exclude_pks, deleted_pks - - -def _save_deserialized_record(store_model, app_model, model_name, excluded_list=None): - """ - Attempt to save one deserialized app model into the app table. - - On success: clears dirty_bit and deserialization_error on the Store record. - On failure: records the error on the Store record, keeps dirty_bit True, - and logs a warning. - - :returns: True if save succeeded, False otherwise - """ - - try: - with transaction.atomic(): - with mute_signals(signals.pre_save, signals.post_save): - app_model.save(update_dirty_bit_to=False) - store_model.dirty_bit = False - store_model.unset_deserialization_error() - store_model.save( - update_fields=["dirty_bit", "deserialization_error", "deserialization_exception"] - ) - return True - except ( - exceptions.ValidationError, - exceptions.ObjectDoesNotExist, - ValueError, - IntegrityError, - ) as e: - if excluded_list is not None: - excluded_list.append(store_model.id) - store_model.set_deserialization_error(e) - store_model.save(update_fields=["deserialization_error", "deserialization_exception"]) - logger.warning( - "Failed to deserialize Store record %s for %s: %s", - store_model.id, - model_name, - e, - ) - return False - - -def _deserialize_from_store(profile, skip_erroring=False, filter=None): - """ - Takes data from the store and integrates into the application. - - ALGORITHM: On a per syncable model basis, we iterate through each class model and we go through 2 possible cases: - - 1. For class models that have a self referential foreign key, we iterate down the dependency tree deserializing model by model. - 2. For other models, we deserialize and validate each record, then save individually so that DB-level errors - (e.g. unique constraint violations) are caught and recorded per-record rather than silently lost or crashing - the entire deserialization. - - If a model fails to deserialize/validate, we exclude it from being marked as clean in the store. - """ - - fk_cache = {} - excluded_list = [] - deleted_list = [] - - with begin_transaction(filter, isolated=True): - # iterate through classes which are in foreign key dependency order - for model in syncable_models.get_models(profile): - deferred_fks = defaultdict(list) - store_models = Store.objects.filter(profile=profile) - - model_condition = Q(model_name=model.morango_model_name) - for klass in model.morango_model_dependencies: - model_condition |= Q(model_name=klass.morango_model_name) - - store_models = store_models.filter(model_condition) - - if filter: - # create Q objects for filtering by prefixes - prefix_condition = functools.reduce( - lambda x, y: x | y, - [Q(partition__startswith=prefix) for prefix in filter], - ) - store_models = store_models.filter(prefix_condition) - - # if requested, skip any records that previously errored, to be faster - if skip_erroring: - # previously, deserialization_error was not nullable, and set to empty string if - # there wasn't an error, so this nullifies empty strings to ensure the filter - # applies either way - store_models = ( - store_models - .annotate( - n_deserialization_error=NullIf( - F("deserialization_error"), - Value(""), - output_field=BooleanField() - ) - ) - .filter(n_deserialization_error__isnull=True) - ) - - # handle cases where a class has a single FK reference to itself - if self_referential_fk(model): - clean_parents = store_models.filter(dirty_bit=False).char_ids_list() - dirty_children = ( - store_models.filter(dirty_bit=True) - # handle parents or if the model has no parent - .filter(Q(_self_ref_fk__in=clean_parents) | Q(_self_ref_fk="")) - ) - - # keep iterating until size of dirty_children is 0 - while len(dirty_children) > 0: - for store_model in dirty_children: - try: - app_model, _ = store_model._deserialize_store_model( - fk_cache, sync_filter=filter - ) - except ( - exceptions.ValidationError, - exceptions.ObjectDoesNotExist, - ValueError, - ) as e: - excluded_list.append(store_model.id) - store_model.set_deserialization_error(e) - store_model.save(update_fields=["deserialization_error", "deserialization_exception"]) - continue - if app_model: - _save_deserialized_record( - store_model, app_model, model.__name__, excluded_list - ) - else: - store_model.dirty_bit = False - store_model.unset_deserialization_error() - store_model.save( - update_fields=["dirty_bit", "deserialization_error", "deserialization_exception"] - ) - - # update lists with new clean parents and dirty children - clean_parents = store_models.filter(dirty_bit=False).char_ids_list() - dirty_children = store_models.filter( - dirty_bit=True, _self_ref_fk__in=clean_parents - ).exclude(id__in=excluded_list) - - # A. Mark records that were skipped due to missing parents with error info - # A(i). The ones that have a parent Store entry but it's dirty - dirty_parents = store_models.filter(dirty_bit=True).char_ids_list() - store_models.filter( - dirty_bit=True, _self_ref_fk__in=dirty_parents - ).exclude(id__in=excluded_list).update( - deserialization_error="Parent is dirty; could not deserialize.", - deserialization_exception=MorangoDirtyParent.path(), - ) - # A(ii). The ones that don't even have Store entries for parent at all - all_parents = store_models.char_ids_list() - store_models.filter(dirty_bit=True).exclude( - _self_ref_fk__in=all_parents - ).exclude(id__in=excluded_list).update( - deserialization_error="Parent does not exist in Store; could not deserialize.", - deserialization_exception=MorangoMissingParent.path(), - ) - - else: - # collect all initially valid app models and validate their FKs - app_models = [] - for store_model in store_models.filter(dirty_bit=True): - try: - ( - app_model, - model_deferred_fks, - ) = store_model._deserialize_store_model( - fk_cache, - defer_fks=True, - sync_filter=filter, - ) - if app_model: - app_models.append((store_model, app_model)) - for fk_model, fk_refs in model_deferred_fks.items(): - # validate that the FK references aren't to anything already in the - # excluded list, which should only contain models which failed to - # deserialize for reasons other than broken FKs at this point - for fk_ref in fk_refs: - if fk_ref.to_pk in excluded_list: - raise exceptions.ValidationError( - "{} with id {} failed to deserialize".format( - fk_model, fk_ref.to_pk - ) - ) - deferred_fks[fk_model].extend(fk_refs) - except ( - exceptions.ValidationError, - exceptions.ObjectDoesNotExist, - ValueError, - ) as e: - # if the app model did not validate, we leave the store dirty bit set - excluded_list.append(store_model.id) - store_model.set_deserialization_error(e) - store_model.save(update_fields=["deserialization_error", "deserialization_exception"]) - - # validate app model FKs - model_excluded_pks, model_deleted_pks = _validate_store_foreign_keys( - model.__name__, deferred_fks - ) - excluded_list.extend(model_excluded_pks) - deleted_list.extend(model_deleted_pks) - - # save each app model individually so we can catch per-record - # DB-level errors (e.g. unique constraint violations) - for store_model, app_model in app_models: - if ( - app_model.pk in excluded_list - or app_model.pk in deleted_list - ): - continue - _save_deserialized_record( - store_model, app_model, model.__name__ - ) - - def _queue_into_buffer_v1(transfersession): """ Takes a chunk of data from the store to be put into the buffer to be sent to another morango instance. This is the legacy From af3a4f4232ae90a340347cfc630a67d1135bd81e Mon Sep 17 00:00:00 2001 From: Blaine Jester Date: Tue, 13 Oct 2026 16:47:21 -0700 Subject: [PATCH 13/13] Rebuild Store._deserialize_store_model on the streaming transforms Keeps the legacy signature and return shape for manual use. The deferred FK references are no longer produced, so defer_fks warns and the second element is always empty. Co-Authored-By: Claude Opus 5.5 (1M context) --- morango/models/core.py | 88 ++++++------------------- morango/sync/stream/deserialize.py | 34 ++++++++++ tests/testapp/tests/models/test_core.py | 82 +++++++++++++++++++++++ 3 files changed, 135 insertions(+), 69 deletions(-) diff --git a/morango/models/core.py b/morango/models/core.py index 76a46c77..05ed0b9a 100644 --- a/morango/models/core.py +++ b/morango/models/core.py @@ -2,6 +2,7 @@ import json import logging import uuid +import warnings from collections import defaultdict from collections import namedtuple from functools import reduce @@ -16,12 +17,10 @@ from django.db.models import Func from django.db.models import Max from django.db.models import Q -from django.db.models import signals from django.db.models import TextField from django.db.models import Value from django.db.models.deletion import Collector from django.db.models.expressions import CombinedExpression -from django.db.models.fields.related import ForeignKey from django.db.models.functions import Cast from django.db.models.functions import NullIf from django.utils import timezone @@ -41,7 +40,6 @@ from morango.models.utils import get_0_4_system_parameters from morango.models.utils import get_0_5_mac_address from morango.models.utils import get_0_5_system_id -from morango.registry import syncable_models from morango.utils import _assert from morango.utils import exception_path from morango.utils import SETTINGS @@ -501,81 +499,33 @@ def unset_deserialization_error(self): self.deserialization_error = None self.deserialization_exception = None - def _deserialize_store_model(self, fk_cache, defer_fks=False, sync_filter=None): # noqa: C901 + def _deserialize_store_model(self, fk_cache, defer_fks=False, sync_filter=None): """ - When deserializing a store model, we look at the deleted flags to know if we should delete the app model. - Upon loading the app model in memory we validate the app models fields, if any errors occurs we follow - foreign key relationships to see if the related model has been deleted to propagate that deletion to the target app model. + Deserializes and validates this store record into its app model, without saving either. + + A deleted record deletes its app model. A record that references a deleted record is + tracked as deleted in `DeletedModels`, and in `HardDeletedModels` for a hard deletion. A + record that fails is logged, and its error is raised. + We return: - None => if the model was deleted successfully - model => if the model validates successfully + (None, {}) => if the record was deleted, or its deletion propagated + (model, {}) => if the model validates successfully - :param fk_cache: A cache for foreign key lookups + :param fk_cache: A cache for foreign key lookups, shared across calls :type fk_cache: dict - :param defer_fks: Whether to defer foreign key lookups + :param defer_fks: No longer supported and ignored :type defer_fks: bool :param sync_filter: The current sync's filter, if any :type sync_filter: Filter|None """ - deferred_fks = {} - klass_model = syncable_models.get_model(self.profile, self.model_name) - # if store model marked as deleted, attempt to delete in app layer - if self.deleted: - # Don't differentiate between deletion and hard deletion here, - # as we don't want to add additional tracking for models in either case, - # just to actually delete them. - # Import here to avoid circular import, as the utils module - # imports core models. - from morango.sync.utils import mute_signals - - with mute_signals(signals.post_delete): - klass_model.syncing_objects.filter(id=self.id).delete() - return None, deferred_fks - else: - # load model into memory - app_model = klass_model.deserialize( - json.loads(self.serialized), sync_filter=sync_filter - ) - app_model._morango_source_id = self.source_id - app_model._morango_partition = self.partition - app_model._morango_dirty_bit = False + # Import here to avoid circular import, as the deserialize module imports core models. + from morango.sync.stream.deserialize import deserialize_store_model - try: - # validate and return the model - if defer_fks: - deferred_fks = app_model.deferred_clean_fields(sync_filter=sync_filter) - else: - app_model.cached_clean_fields(fk_cache, sync_filter=sync_filter) - return app_model, deferred_fks - - except (exceptions.ValidationError, exceptions.ObjectDoesNotExist) as e: - logger.warning( - "Error deserializing instance of {model} with id {id}: {error}".format( - model=klass_model.__name__, id=app_model.id, error=e - ) - ) - - if not defer_fks and isinstance(e, exceptions.ObjectDoesNotExist): - # check FKs in store to see if any of those models were deleted or hard_deleted to propagate to this model - fk_ids = [ - getattr(app_model, field.attname) - for field in app_model._meta.fields - if isinstance(field, ForeignKey) - ] - for fk_id in fk_ids: - try: - st_model = Store.objects.get(id=fk_id) - if st_model.deleted: - # if hard deleted, propagate to store model - if st_model.hard_deleted: - app_model._update_hard_deleted_models() - app_model._update_deleted_models() - return None, {} - except Store.DoesNotExist: - pass - - # if we got here, it means the validation error wasn't handled by propagating deletion, so re-raise it - raise e + if defer_fks: + warnings.warn( + "defer_fks is no longer supported and is ignored", DeprecationWarning, stacklevel=2 + ) + return deserialize_store_model(self, fk_cache, sync_filter=sync_filter), {} class Buffer(AbstractStore): diff --git a/morango/sync/stream/deserialize.py b/morango/sync/stream/deserialize.py index 26c8b8e5..e3afa22a 100644 --- a/morango/sync/stream/deserialize.py +++ b/morango/sync/stream/deserialize.py @@ -558,6 +558,40 @@ def store_model_partition_fn(task: DeserializeTask) -> str: return task.store.model_name +def deserialize_store_model( + store: Store, fk_cache: Dict, sync_filter: Optional[Filter] = None +) -> Optional[SyncableModel]: + """ + Deserializes and validates a single store record outside of a pipeline run. Deleted records are + applied: a deleted record's app model is deleted, and a record whose FK target was deleted is + tracked as deleted. Neither the store record nor a valid app model is saved. + + :param store: The store record to deserialize + :param fk_cache: Dictionary cache for FK references, shared across calls + :param sync_filter: The Filter object for the current sync, if any + :return: The validated app model, or None if the record was deleted + :raises Exception: The error that the record failed with, after logging it + """ + task = DeserializeTask(store, fk_cache) + AppModelDeserialize(sync_filter=sync_filter).transform(task) + AppModelValidate(sync_filter=sync_filter).transform(task) + + outcome = task.outcome + if outcome == DeserializeOutcome.ERROR: + error = task.errors[0] + log_deserialization_error(task, error) + raise error + if outcome == DeserializeOutcome.DELETE: + # without sink lag, a later call validates against the app tables where the cascaded rows + # are already gone, so the cascaded keys aren't needed + delete_app_models(task.model, [task.id]) + return None + if outcome in (DeserializeOutcome.PROPAGATE_DELETE, DeserializeOutcome.PROPAGATE_HARD_DELETE): + record_propagated_deletions([task]) + return None + return task.app_model + + CHUNK_SIZE = 500 diff --git a/tests/testapp/tests/models/test_core.py b/tests/testapp/tests/models/test_core.py index cdc68d2f..aaa33bcf 100644 --- a/tests/testapp/tests/models/test_core.py +++ b/tests/testapp/tests/models/test_core.py @@ -3,17 +3,20 @@ import factory import mock from django.core.exceptions import ValidationError +from django.core.serializers.json import DjangoJSONEncoder from django.test import override_settings from django.test import TestCase from django.utils import timezone from facility_profile.models import Facility from facility_profile.models import MyUser +from facility_profile.models import SummaryLog from morango.constants import transfer_stages from morango.constants import transfer_statuses from morango.models.certificates import Filter from morango.models.core import Buffer from morango.models.core import DatabaseMaxCounter +from morango.models.core import DeletedModels from morango.models.core import Store from morango.models.core import SyncSession from morango.models.core import TransferSession @@ -607,3 +610,82 @@ def test_chains_onto_an_existing_queryset(self): queryset = Store.objects.filter(id=self.empty.id).filter_has_deserialization_error() self.assertEqual(list(queryset), []) + + +class StoreDeserializeStoreModelTestCase(TestCase): + """Tests for Store._deserialize_store_model(), which deserializes a single record by hand""" + + def _store(self, app_model, **kwargs): + store_kwargs = dict( + id=app_model.id, + profile="facilitydata", + model_name=app_model.morango_model_name, + partition=app_model.id, + source_id=uuid.uuid4().hex, + serialized=DjangoJSONEncoder().encode(app_model.serialize()), + last_saved_instance=uuid.uuid4().hex, + last_saved_counter=1, + dirty_bit=True, + ) + store_kwargs.update(kwargs) + return StoreFactory(**store_kwargs) + + def _user(self, username="user", **kwargs): + user = MyUser(id=uuid.uuid4().hex, username=username, password="password") + return self._store(user, **kwargs) + + def test_valid(self): + store = self._user() + + app_model, deferred_fks = store._deserialize_store_model({}) + + self.assertEqual(deferred_fks, {}) + self.assertIsInstance(app_model, MyUser) + self.assertEqual(app_model.username, "user") + self.assertFalse(MyUser.objects.exists()) + self.assertTrue(Store.objects.get(id=store.id).dirty_bit) + + def test_deleted(self): + user = MyUser(username="user", password="password") + user.save(update_dirty_bit_to=False) + store = self._store(user, deleted=True) + + self.assertEqual(store._deserialize_store_model({}), (None, {})) + self.assertFalse(MyUser.objects.exists()) + + def test_target_deleted(self): + user = self._user(dirty_bit=False, deleted=True) + store = self._store(SummaryLog(id=uuid.uuid4().hex, user_id=user.id)) + + self.assertEqual(store._deserialize_store_model({}), (None, {})) + self.assertTrue(DeletedModels.objects.filter(id=store.id).exists()) + self.assertTrue(Store.objects.get(id=store.id).dirty_bit) + + def test_invalid(self): + store = self._user(username="x" * 256) + + with self.assertLogs("morango.sync.stream.deserialize", "WARNING") as logs: + with self.assertRaises(ValidationError): + store._deserialize_store_model({}) + + self.assertIn(store.id, logs.output[0]) + + def test_defer_fks(self): + store = self._user() + + with self.assertWarns(DeprecationWarning): + app_model, deferred_fks = store._deserialize_store_model({}, defer_fks=True) + + self.assertEqual(app_model.id, store.id) + self.assertEqual(deferred_fks, {}) + + def test_fk_cache(self): + """The user is validated but never saved, so the log only validates through the cache""" + fk_cache = {} + user = self._user() + log = self._store(SummaryLog(id=uuid.uuid4().hex, user_id=user.id)) + + user._deserialize_store_model(fk_cache) + app_model, _ = log._deserialize_store_model(fk_cache) + + self.assertEqual(app_model.id, log.id)