Assignment handling within single-user syncing

kolibri/core/auth/apps.py

@@ -13,3 +13,10 @@ class KolibriAuthConfig(AppConfig):
     def ready(self):
         from .signals import cascade_delete_membership  # noqa: F401
         from .signals import cascade_delete_user  # noqa: F401
+
+        from kolibri.core.auth.sync_event_hook_utils import (
+            register_sync_event_handlers,
+        )  # noqa: F401
+        from morango.api.viewsets import session_controller  # noqa: F401
+
+        register_sync_event_handlers(session_controller)

kolibri/core/auth/hooks.py

@@ -0,0 +1,29 @@
+from kolibri.plugins.hooks import define_hook
+from kolibri.plugins.hooks import KolibriHook
+
+
+@define_hook
+class FacilityDataSyncHook(KolibriHook):
+    """
+    A hook to allow plugins to register callbacks for sync events they're interested in.
+    """
+
+    def pre_transfer(
+        self,
+        dataset_id,
+        local_is_single_user,
+        remote_is_single_user,
+        single_user_id,
+        context,
+    ):
+        pass
+
+    def post_transfer(
+        self,
+        dataset_id,
+        local_is_single_user,
+        remote_is_single_user,
+        single_user_id,
+        context,
+    ):
+        pass

kolibri/core/auth/management/commands/sync.py

@@ -23,12 +23,14 @@
 from kolibri.core.auth.management.utils import get_facility
 from kolibri.core.auth.management.utils import run_once
 from kolibri.core.auth.models import dataset_cache
+from kolibri.core.auth.sync_event_hook_utils import register_sync_event_handlers
 from kolibri.core.logger.utils.data import bytes_for_humans
 from kolibri.core.tasks.exceptions import UserCancelledError
 from kolibri.core.tasks.management.commands.base import AsyncCommand
 from kolibri.core.utils.lock import db_lock
 from kolibri.utils import conf
 
+
 DATA_PORTAL_SYNCING_BASE_URL = conf.OPTIONS["Urls"]["DATA_PORTAL_SYNCING_BASE_URL"]
 TRANSFER_MESSAGE = "{records_transferred}/{records_total}, {transfer_total}"
 
@@ -218,6 +220,8 @@ def handle_async(self, *args, **options):  # noqa C901
             client_cert, server_cert, chunk_size=chunk_size
         )
 
+        register_sync_event_handlers(sync_session_client.controller)
+
         try:
             # pull from server
             if not no_pull:
@@ -407,7 +411,7 @@ def _handle_push(
 
         # allow server timeout since remotely integrating data can take a while and the request
         # could timeout. In that case, we'll assume everything is good.
-        sync_client.finalize(allow_server_timeout=True)
+        sync_client.finalize()
 
     def _update_all_progress(self, progress_fraction, progress):
         """
@@ -475,11 +479,14 @@ def handler(transfer_session):
             """
             :type transfer_session: morango.models.core.TransferSession
             """
-            progress = (
-                100
-                * transfer_session.records_transferred
-                / float(transfer_session.records_total)
-            )
+            try:
+                progress = (
+                    100
+                    * transfer_session.records_transferred
+                    / float(transfer_session.records_total)
+                )
+            except ZeroDivisionError:
+                progress = 100
             tracker.update_progress(
                 increment=math.ceil(progress - tracker.progress),
                 message=stats_msg(transfer_session),

kolibri/core/auth/sync_event_hook_utils.py

@@ -0,0 +1,69 @@
+import json
+
+from morango.sync.context import LocalSessionContext
+
+from kolibri.core.auth.constants.morango_sync import ScopeDefinitions
+from kolibri.core.auth.hooks import FacilityDataSyncHook
+
+
+def _get_our_cert(context):
+    ss = context.sync_session
+    return ss.server_certificate if ss.is_server else ss.client_certificate
+
+
+def _get_their_cert(context):
+    ss = context.sync_session
+    return ss.client_certificate if ss.is_server else ss.server_certificate
+
+
+def _this_side_using_single_user_cert(context):
+    return _get_our_cert(context).scope_definition_id == ScopeDefinitions.SINGLE_USER
+
+
+def _other_side_using_single_user_cert(context):
+    return _get_their_cert(context).scope_definition_id == ScopeDefinitions.SINGLE_USER
+
+
+def _get_user_id_for_single_user_sync(context):
+    if _other_side_using_single_user_cert(context):
+        cert = _get_their_cert(context)
+    elif _this_side_using_single_user_cert(context):
+        cert = _get_our_cert(context)
+    else:
+        return None
+    return json.loads(cert.scope_params)["user_id"]
+
+
+def _extract_kwargs_from_context(context):
+    return {
+        "dataset_id": _get_our_cert(context).get_root().id,
+        "local_is_single_user": _this_side_using_single_user_cert(context),
+        "remote_is_single_user": _other_side_using_single_user_cert(context),
+        "single_user_id": _get_user_id_for_single_user_sync(context),
+        "context": context,
+    }
+
+
+def _pre_transfer_handler(context):
+    assert context is not None
+
+    kwargs = _extract_kwargs_from_context(context)
+
+    if isinstance(context, LocalSessionContext):
+        for hook in FacilityDataSyncHook.registered_hooks:
+            hook.pre_transfer(**kwargs)
+
+
+def _post_transfer_handler(context):
+    assert context is not None
+
+    kwargs = _extract_kwargs_from_context(context)
+
+    if isinstance(context, LocalSessionContext):
+        for hook in FacilityDataSyncHook.registered_hooks:
+            hook.post_transfer(**kwargs)
+
+
+def register_sync_event_handlers(session_controller):
+    session_controller.signals.initializing.completed.connect(_pre_transfer_handler)
+    session_controller.signals.cleanup.completed.connect(_post_transfer_handler)

kolibri/core/auth/test/sync_utils.py

@@ -57,7 +57,7 @@ def manage(self, *args):
 
     def create_model(self, model, **kwargs):
         kwarg_text = ",".join(
-            '{key}=\\"{value}\\"'.format(key=key, value=value)
+            "{key}={value}".format(key=key, value=repr(value))
             for key, value in kwargs.items()
         )
         self.pipe_shell(