Merge branch 'master' into master

cms/djangoapps/contentstore/tests/test_libraries.py

@@ -15,8 +15,8 @@
 from xmodule.x_module import STUDIO_VIEW
 
 from cms.djangoapps.contentstore.tests.utils import AjaxEnabledTestClient, parse_json
-from cms.djangoapps.contentstore.utils import reverse_library_url, reverse_url, reverse_usage_url
-from cms.djangoapps.contentstore.views.block import _duplicate_block
+from cms.djangoapps.contentstore.utils import reverse_library_url, reverse_url, \
+    reverse_usage_url, duplicate_block
 from cms.djangoapps.contentstore.views.preview import _load_preview_block
 from cms.djangoapps.contentstore.views.tests.test_library import LIBRARY_REST_URL
 from cms.djangoapps.course_creators.views import add_user_with_status_granted
@@ -947,7 +947,7 @@ def test_persistent_overrides(self, duplicate):
         if duplicate:
             # Check that this also works when the RCB is duplicated.
             self.lc_block = modulestore().get_item(
-                _duplicate_block(self.course.location, self.lc_block.location, self.user)
+                duplicate_block(self.course.location, self.lc_block.location, self.user)
             )
             self.problem_in_course = modulestore().get_item(self.lc_block.children[0])
         else:
@@ -1006,7 +1006,7 @@ def test_duplicated_version(self):
 
         # Duplicate self.lc_block:
         duplicate = store.get_item(
-            _duplicate_block(self.course.location, self.lc_block.location, self.user)
+            duplicate_block(self.course.location, self.lc_block.location, self.user)
         )
         # The duplicate should have identical children to the original:
         self.assertEqual(len(duplicate.children), 1)

cms/djangoapps/contentstore/utils.py

@@ -5,20 +5,28 @@
 from collections import defaultdict
 import logging
 from contextlib import contextmanager
-from datetime import datetime
+from datetime import datetime, timezone
+from uuid import uuid4
 
 from django.conf import settings
 from django.urls import reverse
 from django.utils import translation
 from django.utils.translation import gettext as _
+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
+from openedx_events.content_authoring.signals import XBLOCK_DUPLICATED
 from pytz import UTC
+from xblock.fields import Scope
 
 from cms.djangoapps.contentstore.toggles import exam_setting_view_enabled
+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
 from common.djangoapps.student.models import CourseEnrollment
 from common.djangoapps.student.roles import CourseInstructorRole, CourseStaffRole
+from common.djangoapps.xblock_django.user_service import DjangoXBlockUserService
 from openedx.core.djangoapps.course_apps.toggles import proctoring_settings_modal_view_enabled
 from openedx.core.djangoapps.discussions.config.waffle import ENABLE_PAGES_AND_RESOURCES_MICROFRONTEND
 from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration
@@ -29,8 +37,6 @@
 from openedx.features.content_type_gating.models import ContentTypeGatingConfig
 from openedx.features.content_type_gating.partitions import CONTENT_TYPE_GATING_SCHEME
 from cms.djangoapps.contentstore.toggles import (
-    use_new_text_editor,
-    use_new_video_editor,
     use_new_advanced_settings_page,
     use_new_course_outline_page,
     use_new_export_page,
@@ -44,10 +50,13 @@
     use_new_updates_page,
     use_new_video_uploads_page,
 )
+from cms.djangoapps.contentstore.toggles import use_new_text_editor, use_new_video_editor
+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
 
 log = logging.getLogger(__name__)
 
@@ -935,3 +944,202 @@ def update_course_discussions_settings(course_key):
     course = store.get_course(course_key)
     course.discussions_settings['provider_type'] = provider
     store.update_item(course, course.published_by)
+
+
+def duplicate_block(
+    parent_usage_key,
+    duplicate_source_usage_key,
+    user,
+    dest_usage_key=None,
+    display_name=None,
+    shallow=False,
+    is_child=False
+):
+    """
+    Duplicate an existing xblock as a child of the supplied parent_usage_key. You can
+    optionally specify what usage key the new duplicate block will use via dest_usage_key.
+
+    If shallow is True, does not copy children. Otherwise, this function calls itself
+    recursively, and will set the is_child flag to True when dealing with recursed child
+    blocks.
+    """
+    store = modulestore()
+    with store.bulk_operations(duplicate_source_usage_key.course_key):
+        source_item = store.get_item(duplicate_source_usage_key)
+        if not dest_usage_key:
+            # Change the blockID to be unique.
+            dest_usage_key = source_item.location.replace(name=uuid4().hex)
+
+        category = dest_usage_key.block_type
+
+        duplicate_metadata, asides_to_create = gather_block_attributes(
+            source_item, display_name=display_name, is_child=is_child,
+        )
+
+        dest_block = store.create_item(
+            user.id,
+            dest_usage_key.course_key,
+            dest_usage_key.block_type,
+            block_id=dest_usage_key.block_id,
+            definition_data=source_item.get_explicitly_set_fields_by_scope(Scope.content),
+            metadata=duplicate_metadata,
+            runtime=source_item.runtime,
+            asides=asides_to_create
+        )
+
+        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.
+            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)
+
+        # pylint: disable=protected-access
+        if 'detached' not in source_item.runtime.load_block_type(category)._class_tags:
+            parent = store.get_item(parent_usage_key)
+            # If source was already a child of the parent, add duplicate immediately afterward.
+            # Otherwise, add child to end.
+            if source_item.location in parent.children:
+                source_index = parent.children.index(source_item.location)
+                parent.children.insert(source_index + 1, dest_block.location)
+            else:
+                parent.children.append(dest_block.location)
+            store.update_item(parent, user.id)
+
+        # .. event_implemented_name: XBLOCK_DUPLICATED
+        XBLOCK_DUPLICATED.send_event(
+            time=datetime.now(timezone.utc),
+            xblock_info=DuplicatedXBlockData(
+                usage_key=dest_block.location,
+                block_type=dest_block.location.block_type,
+                source_usage_key=duplicate_source_usage_key,
+            )
+        )
+
+        return dest_block.location
+
+
+def update_from_source(*, source_block, destination_block, user_id):
+    """
+    Update a block to have all the settings and attributes of another source.
+
+    Copies over all attributes and settings of a source block to a destination
+    block. Blocks must be the same type. This function does not modify or duplicate
+    children.
+
+    This function is useful when a block, originally copied from a source block, drifts
+    and needs to be updated to match the original.
+
+    The modulestore function copy_from_template will copy a block's children recursively,
+    replacing the target block's children. It does not, however, update any of the target
+    block's settings. copy_from_template, then, is useful for cases like the Library
+    Content Block, where the children are the same across all instances, but the settings
+    may differ.
+
+    By contrast, for cases where we're copying a block that has drifted from its source,
+    we need to update the target block's settings, but we don't want to replace its children,
+    or, at least, not only replace its children. update_from_source is useful for these cases.
+
+    This function is meant to be imported by pluggable django apps looking to manage duplicated
+    sections of a course. It is placed here for lack of a more appropriate location, since this
+    code has not yet been brought up to the standards in OEP-45.
+    """
+    duplicate_metadata, asides = gather_block_attributes(source_block, display_name=source_block.display_name)
+    for key, value in duplicate_metadata.items():
+        setattr(destination_block, key, value)
+    for key, value in source_block.get_explicitly_set_fields_by_scope(Scope.content).items():
+        setattr(destination_block, key, value)
+    modulestore().update_item(
+        destination_block,
+        user_id,
+        metadata=duplicate_metadata,
+        asides=asides,
+    )
+
+
+def gather_block_attributes(source_item, display_name=None, is_child=False):
+    """
+    Gather all the attributes of the source block that need to be copied over to a new or updated block.
+    """
+    # Update the display name to indicate this is a duplicate (unless display name provided).
+    # Can't use own_metadata(), b/c it converts data for JSON serialization -
+    # not suitable for setting metadata of the new block
+    duplicate_metadata = {}
+    for field in source_item.fields.values():
+        if field.scope == Scope.settings and field.is_set_on(source_item):
+            duplicate_metadata[field.name] = field.read_from(source_item)
+
+    if is_child:
+        display_name = display_name or source_item.display_name or source_item.category
+
+    if display_name is not None:
+        duplicate_metadata['display_name'] = display_name
+    else:
+        if source_item.display_name is None:
+            duplicate_metadata['display_name'] = _("Duplicate of {0}").format(source_item.category)
+        else:
+            duplicate_metadata['display_name'] = _("Duplicate of '{0}'").format(source_item.display_name)
+
+    asides_to_create = []
+    for aside in source_item.runtime.get_asides(source_item):
+        for field in aside.fields.values():
+            if field.scope in (Scope.settings, Scope.content,) and field.is_set_on(aside):
+                asides_to_create.append(aside)
+                break
+
+    for aside in asides_to_create:
+        for field in aside.fields.values():
+            if field.scope not in (Scope.settings, Scope.content,):
+                field.delete_from(aside)
+    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
+
+
+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)