Merge pull request #284 from bjester/op-mod-con

Allow sync operations to modify the sync filter

Author: bjester

.gitignore

@@ -96,6 +96,7 @@ Pipfile
 
 # Jetbrains IDE
 .idea
+/*.iml
 
 *.db
 

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 List of the most important changes for each release.
 
+## 0.8.6
+- Allows sync operations to modify the context, particularly the sync filter, as long as the original is a subset of the new filter
+- Deprecates usage of `Filter` for parameter replacement. Use `Filter.from_template` instead
+- Adds additional utilities to `Filter` class and defensive logic for various scenarios
+
 ## 0.8.5
 - Prevents MPTT corruption that occurs with concurrent certificate creation
 

morango/__init__.py

@@ -1 +1 @@
-__version__ = "0.8.5"
+__version__ = "0.8.6"

morango/models/certificates.py

@@ -4,13 +4,16 @@
 and a ``public_key`` used for verifying that a certificate(s) was properly signed.
 """
 import json
-import string
 import logging
+import string
+from contextlib import contextmanager
 
 import mptt.models
 from django.core.management import call_command
+from django.db import connection
 from django.db import models
 from django.db import transaction
+from django.db.utils import OperationalError
 from django.utils import timezone
 
 from .fields.crypto import Key
@@ -24,11 +27,9 @@
 from morango.errors import CertificateSignatureInvalid
 from morango.errors import NonceDoesNotExist
 from morango.errors import NonceExpired
-from morango.utils import _assert
-from django.db import transaction, connection
 from morango.sync.backends.utils import load_backend
-from contextlib import contextmanager
-from django.db.utils import OperationalError
+from morango.utils import _assert
+
 
 class Certificate(mptt.models.MPTTModel, UUIDModelMixin):
 
@@ -360,45 +361,89 @@ def get_description(self, params):
 
 
 class Filter(object):
-    def __init__(self, template, params={}):
-        # ensure params have been deserialized
-        if isinstance(params, str):
-            params = json.loads(params)
-        self._template = template
-        self._params = params
-        self._filter_string = string.Template(template).safe_substitute(params)
-        self._filter_tuple = tuple(self._filter_string.split()) or ("",)
+    def __init__(self, filter_str, params=None):
+        """
+        :param filter_str: The partition filter string, which may have multiple separated by newlines
+        :type filter_str: str
+        :param params: DEPRECATED: USE Filter.from_template() INSTEAD
+        :type params: dict|str
+        """
+        if params is not None:
+            logging.warning("DEPRECATED: Constructing a filter with a template and params is deprecated. Use Filter.from_template() instead")
+            filter_str = str(Filter.from_template(filter_str, params=params))
+
+        self._filter_tuple = tuple(filter_str.split()) or ("",)
 
     def is_subset_of(self, other):
-        for partition in self._filter_tuple:
-            if not partition.startswith(other._filter_tuple):
+        """
+        :param other: The other Filter
+        :type other: Filter
+        :return: A boolean on whether this Filter is captured within the other Filter
+        :rtype: bool
+        """
+        for partition in self:
+            if not other.contains_partition(partition):
                 return False
         return True
 
     def contains_partition(self, partition):
+        """Returns True if the partition starts with as least one of the partitions in this Filter"""
         return partition.startswith(self._filter_tuple)
 
+    def contains_exact_partition(self, partition):
+        """Returns True if the partition exactly matches one of the partitions in this Filter"""
+        return partition in self._filter_tuple
+
+    def copy(self):
+        return Filter(str(self))
+
     def __le__(self, other):
+        """Returns True if this Filter is a subset of the other"""
         return self.is_subset_of(other)
 
     def __eq__(self, other):
+        """Returns True if this Filter has exactly the same partitions as the other"""
         if other is None:
             return False
-        for partition in self._filter_tuple:
-            if partition not in other._filter_tuple:
+        for partition in self:
+            if not other.contains_exact_partition(partition):
                 return False
-        for partition in other._filter_tuple:
-            if partition not in self._filter_tuple:
+        for partition in other:
+            if not self.contains_exact_partition(partition):
                 return False
         return True
 
     def __contains__(self, partition):
+        """
+        Performs a 'startswith' comparison on the partition, determining whether it matches or
+        is a subset of any partition in this Filter
+
+        :param partition: str
+        :return: A boolean
+        :rtype: bool
+        """
         return self.contains_partition(partition)
 
     def __add__(self, other):
-        return Filter(self._filter_string + "\n" + other._filter_string)
+        """
+        The Filter's addition operator overload
+        :param other: Filter or None
+        :type other: Filter|None
+        :return: The combined Filter
+        :rtype: Filter
+        """
+        if other is None:
+            return self
+        # create a list of partition filters, deduplicating them between the two filter objects
+        partitions = []
+        partitions.extend(p for p in self if p)
+        partitions.extend(p for p in other if p and p not in partitions)
+        return Filter("\n".join(partitions))
 
     def __iter__(self):
+        """
+        :rtype: tuple[str]
+        """
         return iter(self._filter_tuple)
 
     def __str__(self):
@@ -407,13 +452,48 @@ def __str__(self):
     def __len__(self):
         return len(self._filter_tuple)
 
+    @classmethod
+    def add(cls, filter_a, filter_b):
+        """
+        The Filter's addition operator overload is already defensive against None being the
+        right-hand operand, but this method is defensive against None being the left-hand operand
+
+        :param filter_a: A Filter or None
+        :type filter_a: Filter|None
+        :param filter_b: A Filter or None
+        :type filter_b: Filter|None
+        :return: The combined Filter or None
+        :rtype: Filter|None
+        """
+        if filter_a is None:
+            return filter_b
+        return filter_a + filter_b
+
+    @classmethod
+    def from_template(cls, template, params=None):
+        """
+        Create a filter from a string template, which may have params that will be replaced with
+        values passed to `params`
+
+        :param template: The partition filter template
+        :type template: str
+        :param params: The param dictionary or JSON object string
+        :type params: dict|str
+        :return: The filter with params replaced
+        :rtype: Filter
+        """
+        if isinstance(params, str):
+            params = json.loads(params)
+        params = params or {}
+        return Filter(string.Template(template).safe_substitute(params))
+
 
 class Scope(object):
     def __init__(self, definition, params):
         # turn the scope definition filter templates into Filter objects
-        rw_filter = Filter(definition.read_write_filter_template, params)
-        self.read_filter = rw_filter + Filter(definition.read_filter_template, params)
-        self.write_filter = rw_filter + Filter(definition.write_filter_template, params)
+        rw_filter = Filter.from_template(definition.read_write_filter_template, params)
+        self.read_filter = rw_filter + Filter.from_template(definition.read_filter_template, params)
+        self.write_filter = rw_filter + Filter.from_template(definition.write_filter_template, params)
 
     def is_subset_of(self, other):
         if not self.read_filter.is_subset_of(other.read_filter):

morango/sync/context.py

@@ -63,6 +63,13 @@ def prepare(self):
         """
         return self
 
+    def join(self, context):
+        """
+        Perform any processing of session context after passing it to the middleware, which will
+        receive the result returned by `prepare` after that has happened
+        """
+        pass
+
     def update(
         self,
         transfer_session=None,
@@ -83,7 +90,7 @@ def update(
         :type capabilities: str[]|None
         :type error: BaseException|None
         """
-        if transfer_session and self.transfer_session:
+        if transfer_session and self.transfer_session and transfer_session.id != self.transfer_session.id:
             raise MorangoContextUpdateError("Transfer session already exists")
         elif (
             transfer_session
@@ -92,14 +99,18 @@ def update(
         ):
             raise MorangoContextUpdateError("Sync session mismatch")
 
-        if sync_filter and self.filter:
-            raise MorangoContextUpdateError("Filter already exists")
+        if sync_filter and self.filter and sync_filter != self.filter:
+            if not self.filter.is_subset_of(sync_filter):
+                raise MorangoContextUpdateError("The existing filter must be a subset of the new filter")
+            if transfer_stages.stage(self.stage) > transfer_stages.stage(transfer_stages.INITIALIZING):
+                raise MorangoContextUpdateError("Cannot update filter after initializing stage")
 
         if is_push is not None and self.is_push is not None:
             raise MorangoContextUpdateError("Push/pull method already exists")
 
         self.transfer_session = transfer_session or self.transfer_session
-        self.filter = sync_filter or self.filter
+        if sync_filter or self.filter:
+            self.filter = Filter.add(self.filter, sync_filter)
         self.is_push = is_push if is_push is not None else self.is_push
         self.capabilities = set(capabilities or self.capabilities) & CAPABILITIES
         self.update_state(stage=stage, stage_status=stage_status)
@@ -402,50 +413,47 @@ def prepare(self):
         """
         return self.children[self._counter % len(self.children)]
 
+    def join(self, context):
+        """
+        This updates some context attributes that really only make sense logically for what an
+        operation might change during a sync
+
+        :param context: The context that was returned previously from `prepare`
+        :type context: SessionContext
+        """
+        updates = {}
+        if not self.transfer_session and context.transfer_session:
+            # if the transfer session is being resumed, we'd detect a different stage here,
+            # and thus we reset the counter, so we can be sure to start fresh at that stage
+            # on the next invocation of the middleware
+            if (
+                context.transfer_session.transfer_stage
+                and self._stage
+                and context.transfer_session.transfer_stage != self._stage
+            ):
+                self._counter = 0
+                updates.update(
+                    stage=context.transfer_session.transfer_stage,
+                    stage_status=transfer_statuses.PENDING,
+                )
+            updates.update(transfer_session=context.transfer_session)
+        if self.filter != context.filter:
+            updates.update(sync_filter=context.filter)
+        if self.capabilities != context.capabilities:
+            updates.update(capabilities=context.capabilities)
+        if self.error != context.error:
+            updates.update(error=context.error)
+        if updates:
+            self.update(**updates)
+
     def update(self, stage=None, stage_status=None, **kwargs):
-        """
-        Updates the context object and its state
-        :param stage: The str transfer stage
-        :param stage_status: The str transfer stage status
-        :param kwargs: Other arguments to update the context with
-        """
         # update ourselves, but exclude stage and stage_status
         super(CompositeSessionContext, self).update(**kwargs)
         # update children contexts directly, but exclude stage and stage_status
         self._update_attrs(**kwargs)
         # handle state changes after updating children
         self.update_state(stage=stage, stage_status=stage_status)
 
-        # During the initializing stage, we want to make sure to synchronize the transfer session
-        # object between the composite and children contexts, using whatever child context's
-        # transfer session object that was updated on the context during initialization
-        current_stage = stage or self._stage
-        if not self.transfer_session and current_stage == transfer_stages.INITIALIZING:
-            try:
-                transfer_session = next(
-                    c.transfer_session for c in self.children if c.transfer_session
-                )
-                # prepare an updates dictionary, so we can update everything at once
-                updates = dict(transfer_session=transfer_session)
-
-                # if the transfer session is being resumed, we'd detect a different stage here,
-                # and thus we reset the counter, so we can be sure to start fresh at that stage
-                # on the next invocation of the middleware
-                if (
-                    transfer_session.transfer_stage
-                    and transfer_session.transfer_stage != current_stage
-                ):
-                    self._counter = 0
-                    updates.update(
-                        stage=transfer_session.transfer_stage,
-                        stage_status=transfer_statuses.PENDING,
-                    )
-
-                # recurse into update with transfer session and possibly state updates too
-                self.update(**updates)
-            except StopIteration:
-                pass
-
     def update_state(self, stage=None, stage_status=None):
         """
         Updates the state of the transfer

morango/sync/controller.py

@@ -253,6 +253,9 @@ def _invoke_middleware(self, context, middleware):
             # invoke the middleware with the prepared context
             result = middleware(prepared_context)
 
+            # return the prepared context back
+            context.join(prepared_context)
+
             # don't update stage result if context's stage was updated during operation
             if context.stage == stage:
                 context.update(stage_status=result)