refactor: inheritable studioeditableblock's callbacks for editing & d…

…uplication

Solves openedx#33715

Instead of applying duplicated logic after getattr, that logic is
incorporated into inheritable methods of the StudioEditable block.

Risks:
- Assumes all cms blocks that are duplicated inherit from
StudioEditableBlock.

cms/djangoapps/contentstore/utils.py

@@ -15,7 +15,6 @@
 from django.utils import translation
 from django.utils.translation import gettext as _
 from help_tokens.core import HelpUrlExpert
-from lti_consumer.models import CourseAllowPIISharingInLTIFlag
 from opaque_keys.edx.keys import CourseKey, UsageKey
 from opaque_keys.edx.locator import LibraryLocator
 from openedx_events.content_authoring.data import DuplicatedXBlockData
@@ -27,9 +26,7 @@
 from cms.djangoapps.contentstore.toggles import exam_setting_view_enabled
 from common.djangoapps.course_action_state.models import CourseRerunUIStateManager
 from common.djangoapps.course_modes.models import CourseMode
-from common.djangoapps.edxmako.services import MakoService
 from common.djangoapps.student import auth
-from common.djangoapps.student.auth import has_studio_read_access, has_studio_write_access, STUDIO_EDIT_ROLES
 from common.djangoapps.student.models import CourseEnrollment
 from common.djangoapps.student.roles import (
     CourseInstructorRole,
@@ -45,7 +42,6 @@
     get_namespace_choices,
     generate_milestone_namespace
 )
-from common.djangoapps.xblock_django.user_service import DjangoXBlockUserService
 from openedx.core import toggles as core_toggles
 from openedx.core.djangoapps.credit.api import get_credit_requirements, is_credit_course
 from openedx.core.djangoapps.discussions.config.waffle import ENABLE_PAGES_AND_RESOURCES_MICROFRONTEND
@@ -79,12 +75,12 @@
     use_tagging_taxonomy_list_page,
 )
 from cms.djangoapps.models.settings.course_grading import CourseGradingModel
-from xmodule.library_tools import LibraryToolsService
 from xmodule.modulestore import ModuleStoreEnum  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
 from xmodule.modulestore.exceptions import ItemNotFoundError  # lint-amnesty, pylint: disable=wrong-import-order
-from xmodule.partitions.partitions_service import get_all_partitions_for_course  # lint-amnesty, pylint: disable=wrong-import-order
-from xmodule.services import SettingsService, ConfigurationService, TeamsConfigurationService
+from xmodule.partitions.partitions_service import get_all_partitions_for_course
+from xmodule.services import load_services_for_studio
+from xmodule.util.duplicate import handle_children_duplication  # lint-amnesty, pylint: disable=wrong-import-order
 
 
 log = logging.getLogger(__name__)
@@ -1055,23 +1051,17 @@ def duplicate_block(
         )
 
         children_handled = False
-
-        if hasattr(dest_block, 'studio_post_duplicate'):
+        if hasattr(dest_block, "studio_post_duplicate"):
             # Allow an XBlock to do anything fancy it may need to when duplicated from another block.
-            # These blocks may handle their own children or parenting if needed. Let them return booleans to
-            # let us know if we need to handle these or not.
-            load_services_for_studio(dest_block.runtime, user)
-            children_handled = dest_block.studio_post_duplicate(store, source_item)
-
-        # Children are not automatically copied over (and not all xblocks have a 'children' attribute).
-        # Because DAGs are not fully supported, we need to actually duplicate each child as well.
-        if source_item.has_children and not shallow and not children_handled:
-            dest_block.children = dest_block.children or []
-            for child in source_item.children:
-                dupe = duplicate_block(dest_block.location, child, user=user, is_child=True)
-                if dupe not in dest_block.children:  # _duplicate_block may add the child for us.
-                    dest_block.children.append(dupe)
-            store.update_item(dest_block, user.id)
+            load_services_for_studio(source_item.runtime, user)
+            children_handled = dest_block.studio_post_duplicate(
+                store, source_item, store, user, duplication_function=duplicate_block, shallow=shallow
+            )
+
+        if not children_handled:
+            handle_children_duplication(
+                store, source_item, store, user, duplication_function=duplicate_block, shallow=shallow
+            )
 
         # pylint: disable=protected-access
         if 'detached' not in source_item.runtime.load_block_type(category)._class_tags:
@@ -1173,25 +1163,6 @@ def gather_block_attributes(source_item, display_name=None, is_child=False):
     return duplicate_metadata, asides_to_create
 
 
-def load_services_for_studio(runtime, user):
-    """
-    Function to set some required services used for XBlock edits and studio_view.
-    (i.e. whenever we're not loading _prepare_runtime_for_preview.) This is required to make information
-    about the current user (especially permissions) available via services as needed.
-    """
-    services = {
-        "user": DjangoXBlockUserService(user),
-        "studio_user_permissions": StudioPermissionsService(user),
-        "mako": MakoService(),
-        "settings": SettingsService(),
-        "lti-configuration": ConfigurationService(CourseAllowPIISharingInLTIFlag),
-        "teams_configuration": TeamsConfigurationService(),
-        "library_tools": LibraryToolsService(modulestore(), user.id)
-    }
-
-    runtime._services.update(services)  # lint-amnesty, pylint: disable=protected-access
-
-
 def update_course_details(request, course_key, payload, course_block):
     """
     Utils is used to update course details.
@@ -1666,24 +1637,3 @@ def get_course_videos_context(course_block, pagination_conf, course_key=None):
         # Cached state for transcript providers' credentials (org-specific)
         course_video_context['transcript_credentials'] = get_transcript_credentials_state_for_org(course.id.org)
     return course_video_context
-
-
-class StudioPermissionsService:
-    """
-    Service that can provide information about a user's permissions.
-
-    Deprecated. To be replaced by a more general authorization service.
-
-    Only used by LibraryContentBlock (and library_tools.py).
-    """
-
-    def __init__(self, user):
-        self._user = user
-
-    def can_read(self, course_key):
-        """ Does the user have read access to the given course/library? """
-        return has_studio_read_access(self._user, course_key)
-
-    def can_write(self, course_key):
-        """ Does the user have read access to the given course/library? """
-        return has_studio_write_access(self._user, course_key)

cms/djangoapps/contentstore/views/block.py

@@ -27,7 +27,8 @@
 )
 from xmodule.modulestore.django import (
     modulestore,
-)  # lint-amnesty, pylint: disable=wrong-import-order
+)
+from xmodule.services import load_services_for_studio  # lint-amnesty, pylint: disable=wrong-import-order
 
 
 from xmodule.x_module import (
@@ -46,7 +47,6 @@
 from cms.djangoapps.contentstore.xblock_storage_handlers.view_handlers import (
     handle_xblock,
     create_xblock_info,
-    load_services_for_studio,
     get_block_info,
     get_xblock,
     delete_orphans,

cms/djangoapps/contentstore/views/component.py

@@ -34,14 +34,14 @@
 from openedx.core.djangoapps.content_staging import api as content_staging_api
 from openedx.core.djangoapps.content_tagging.api import get_content_tags
 from xmodule.modulestore.django import modulestore  # lint-amnesty, pylint: disable=wrong-import-order
-from xmodule.modulestore.exceptions import ItemNotFoundError  # lint-amnesty, pylint: disable=wrong-import-order
+from xmodule.modulestore.exceptions import ItemNotFoundError
+from xmodule.services import load_services_for_studio  # lint-amnesty, pylint: disable=wrong-import-order
 from ..toggles import use_new_unit_page
 from ..utils import get_lms_link_for_item, get_sibling_urls, reverse_course_url, get_unit_url
 from ..helpers import get_parent_xblock, is_unit, xblock_type_display_name
 from cms.djangoapps.contentstore.xblock_storage_handlers.view_handlers import (
     add_container_page_publishing_info,
     create_xblock_info,
-    load_services_for_studio,
 )
 
 __all__ = [

cms/djangoapps/contentstore/views/preview.py

@@ -21,7 +21,7 @@
 from xmodule.exceptions import NotFoundError, ProcessingError
 from xmodule.modulestore.django import XBlockI18nService, modulestore
 from xmodule.partitions.partitions_service import PartitionService
-from xmodule.services import SettingsService, TeamsConfigurationService
+from xmodule.services import SettingsService, StudioPermissionsService, TeamsConfigurationService
 from xmodule.studio_editable import has_author_view
 from xmodule.util.sandboxing import SandboxService
 from xmodule.util.builtin_assets import add_webpack_js_to_fragment
@@ -45,7 +45,7 @@
     wrap_xblock_aside
 )
 
-from ..utils import get_visibility_partition_info, StudioPermissionsService
+from ..utils import get_visibility_partition_info
 from .access import get_user_role
 from .session_kv_store import SessionKeyValueStore
 

cms/djangoapps/contentstore/xblock_storage_handlers/view_handlers.py

@@ -31,7 +31,6 @@
 )
 from edx_proctoring.exceptions import ProctoredExamNotFoundException
 from help_tokens.core import HelpUrlExpert
-from lti_consumer.models import CourseAllowPIISharingInLTIFlag
 from opaque_keys.edx.locator import LibraryUsageLocator
 from pytz import UTC
 from xblock.core import XBlock
@@ -41,15 +40,13 @@
 from cms.djangoapps.contentstore.toggles import ENABLE_COPY_PASTE_UNITS, use_tagging_taxonomy_list_page
 from cms.djangoapps.models.settings.course_grading import CourseGradingModel
 from cms.lib.ai_aside_summary_config import AiAsideSummaryConfig
-from common.djangoapps.edxmako.services import MakoService
 from common.djangoapps.static_replace import replace_static_urls
 from common.djangoapps.student.auth import (
     has_studio_read_access,
     has_studio_write_access,
 )
 from common.djangoapps.util.date_utils import get_default_time_display
 from common.djangoapps.util.json_request import JsonResponse, expect_json
-from common.djangoapps.xblock_django.user_service import DjangoXBlockUserService
 from openedx.core.djangoapps.bookmarks import api as bookmarks_api
 from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration
 from openedx.core.djangoapps.video_config.toggles import PUBLIC_VIDEO_SHARE
@@ -63,8 +60,9 @@
 from xmodule.modulestore.draft_and_published import DIRECT_ONLY_CATEGORIES
 from xmodule.modulestore.exceptions import InvalidLocationError, ItemNotFoundError
 from xmodule.modulestore.inheritance import own_metadata
-from xmodule.services import ConfigurationService, SettingsService, TeamsConfigurationService
+from xmodule.services import load_services_for_studio
 from xmodule.tabs import CourseTabList
+from xmodule.util.duplicate import handle_children_duplication
 
 from ..utils import (
     ancestor_has_staff_lock,
@@ -296,46 +294,6 @@ def modify_xblock(usage_key, request):
     )
 
 
-class StudioPermissionsService:
-    """
-    Service that can provide information about a user's permissions.
-
-    Deprecated. To be replaced by a more general authorization service.
-
-    Only used by LibraryContentBlock (and library_tools.py).
-    """
-
-    def __init__(self, user):
-        self._user = user
-
-    def can_read(self, course_key):
-        """Does the user have read access to the given course/library?"""
-        return has_studio_read_access(self._user, course_key)
-
-    def can_write(self, course_key):
-        """Does the user have read access to the given course/library?"""
-        return has_studio_write_access(self._user, course_key)
-
-
-def load_services_for_studio(runtime, user):
-    """
-    Function to set some required services used for XBlock edits and studio_view.
-    (i.e. whenever we're not loading _prepare_runtime_for_preview.) This is required to make information
-    about the current user (especially permissions) available via services as needed.
-    """
-    services = {
-        "user": DjangoXBlockUserService(user),
-        "studio_user_permissions": StudioPermissionsService(user),
-        "mako": MakoService(),
-        "settings": SettingsService(),
-        "lti-configuration": ConfigurationService(CourseAllowPIISharingInLTIFlag),
-        "teams_configuration": TeamsConfigurationService(),
-        "library_tools": LibraryToolsService(modulestore(), user.id),
-    }
-
-    runtime._services.update(services)  # lint-amnesty, pylint: disable=protected-access
-
-
 def _update_with_callback(xblock, user, old_metadata=None, old_content=None):
     """
     Updates the xblock in the modulestore.
@@ -860,29 +818,17 @@ def _duplicate_block(
         )
 
         children_handled = False
-
         if hasattr(dest_block, "studio_post_duplicate"):
             # Allow an XBlock to do anything fancy it may need to when duplicated from another block.
-            # These blocks may handle their own children or parenting if needed. Let them return booleans to
-            # let us know if we need to handle these or not.
-            # TODO: Make this a proper method in the base class so we don't need getattr.
-            #       See https://github.com/openedx/edx-platform/issues/33715
-            load_services_for_studio(dest_block.runtime, user)
-            children_handled = dest_block.studio_post_duplicate(store, source_item)
-
-        # Children are not automatically copied over (and not all xblocks have a 'children' attribute).
-        # Because DAGs are not fully supported, we need to actually duplicate each child as well.
-        if source_item.has_children and not children_handled:
-            dest_block.children = dest_block.children or []
-            for child in source_item.children:
-                dupe = _duplicate_block(
-                    dest_block.location, child, user=user, is_child=True
-                )
-                if (
-                    dupe not in dest_block.children
-                ):  # _duplicate_block may add the child for us.
-                    dest_block.children.append(dupe)
-            store.update_item(dest_block, user.id)
+            load_services_for_studio(source_item.runtime, user)
+            children_handled = dest_block.studio_post_duplicate(
+                source_item, store, user, duplication_function=_duplicate_block, shallow=False
+            )
+
+        if not children_handled:
+            handle_children_duplication(
+                dest_block, source_item, store, user, duplication_function=_duplicate_block, shallow=False
+            )
 
         # pylint: disable=protected-access
         if "detached" not in source_item.runtime.load_block_type(category)._class_tags:

xmodule/library_content_block.py

@@ -8,6 +8,7 @@
 import random
 from copy import copy
 from gettext import ngettext, gettext
+from typing import Callable
 
 import bleach
 from django.conf import settings
@@ -587,15 +588,19 @@ def children_are_syncing(self, request, suffix=''):  # pylint: disable=unused-ar
             is_updating = False
         return Response(json.dumps(is_updating))
 
-    def studio_post_duplicate(self, store, source_block):
+    def studio_post_duplicate(
+        self,
+        source_item,
+        *_,
+    ) -> None:  # pylint: disable=unused-argument
         """
         Used by the studio after basic duplication of a source block. We handle the children
         ourselves, because we have to properly reference the library upstream and set the overrides.
 
         Otherwise we'll end up losing data on the next refresh.
         """
         self._validate_sync_permissions()
-        self.get_tools(to_read_library_content=True).trigger_duplication(source_block=source_block, dest_block=self)
+        self.get_tools(to_read_library_content=True).trigger_duplication(source_block=source_item, dest_block=self)
         return True  # Children have been handled.
 
     def _validate_library_version(self, validation, lib_tools, version, library_key):

xmodule/services.py

@@ -6,15 +6,23 @@
 import inspect
 import logging
 from functools import partial
+from common.djangoapps.edxmako.services import MakoService
+from common.djangoapps.xblock_django.user_service import DjangoXBlockUserService
 
 from config_models.models import ConfigurationModel
 from django.conf import settings
 from eventtracking import tracker
 from edx_when.field_data import DateLookupFieldData
+from lti_consumer.models import CourseAllowPIISharingInLTIFlag
 from xblock.reference.plugins import Service
 from xblock.runtime import KvsFieldData
 
+import xmodule
 from common.djangoapps.track import contexts
+from common.djangoapps.student.auth import (
+    has_studio_read_access,
+    has_studio_write_access,
+)
 from lms.djangoapps.courseware.masquerade import is_masquerading_as_specific_student
 from xmodule.modulestore.django import modulestore
 
@@ -308,3 +316,43 @@ def _handle_deprecated_progress_event(self, block, event):
         # in order to avoid duplicate work and possibly conflicting semantics.
         if not getattr(block, 'has_custom_completion', False):
             self.completion_service.submit_completion(block.scope_ids.usage_id, 1.0)
+
+
+def load_services_for_studio(runtime, user):
+    """
+    Function to set some required services used for XBlock edits and studio_view.
+    (i.e. whenever we're not loading _prepare_runtime_for_preview.) This is required to make information
+    about the current user (especially permissions) available via services as needed.
+    """
+    services = {
+        "user": DjangoXBlockUserService(user),
+        "studio_user_permissions": StudioPermissionsService(user),
+        "mako": MakoService(),
+        "settings": SettingsService(),
+        "lti-configuration": ConfigurationService(CourseAllowPIISharingInLTIFlag),
+        "teams_configuration": TeamsConfigurationService(),
+        "library_tools": xmodule.library_tools.LibraryToolsService(modulestore(), user.id),
+    }
+
+    runtime._services.update(services)  # lint-amnesty, pylint: disable=protected-access
+
+
+class StudioPermissionsService:
+    """
+    Service that can provide information about a user's permissions.
+
+    Deprecated. To be replaced by a more general authorization service.
+
+    Only used by LibraryContentBlock (and library_tools.py).
+    """
+
+    def __init__(self, user):
+        self._user = user
+
+    def can_read(self, course_key):
+        """Does the user have read access to the given course/library?"""
+        return has_studio_read_access(self._user, course_key)
+
+    def can_write(self, course_key):
+        """Does the user have read access to the given course/library?"""
+        return has_studio_write_access(self._user, course_key)