feat: Paste Components (OLX) into any Unit in Studio (#31969)

* feat: Implement paste button

* chore: improve docs and add tests for python API

* fix: drive-by fix to use a better API for comparing XML

* feat: track which XBlock something was copied from

* feat: add tests

* feat: enable import linter so content_staging's public API is respected

* fix: error seen when trying to paste drag-and-drop-v2 blocks

* fix: use strip_text=True consistently for XML comparisons

* refactor: rename get_user_clipboard_status to get_user_clipboard

* feat: Better error reporting when pasting in Studio

* chore: convert new test suite to pytest assertions

* refactor: push READY status check into the API per review suggestion

* fix: use strip_text=True consistently for XML comparisons

* fix: store "copied_from_block" as a string to avoid Reference field issues

* fix: minor lint error

* refactor: move data types to data.py per OEP-49

cms/djangoapps/contentstore/views/block.py

@@ -72,6 +72,7 @@ from ..utils import (
 from .helpers import (
     create_xblock,
     get_parent_xblock,
+    import_staged_content_from_user_clipboard,
     is_unit,
     usage_key_with_run,
     xblock_primary_child_category,
@@ -169,6 +170,8 @@ def xblock_handler(request, usage_key_string=None):
                 :display_name: name for new xblock, optional
                 :boilerplate: template name for populating fields, optional and only used
                      if duplicate_source_locator is not present
+                :staged_content: use "clipboard" to paste from the OLX user's clipboard. (Incompatible with all other
+                     fields except parent_locator)
               The locator (unicode representation of a UsageKey) for the created xblock (minus children) is returned.
     """
     if usage_key_string:
@@ -701,6 +704,19 @@ def _create_block(request):
     if not has_studio_write_access(request.user, usage_key.course_key):
         raise PermissionDenied()
 
+    if request.json.get('staged_content') == "clipboard":
+        # Paste from the user's clipboard (content_staging app clipboard, not browser clipboard) into 'usage_key':
+        try:
+            created_xblock = import_staged_content_from_user_clipboard(parent_key=usage_key, request=request)
+        except Exception:  # pylint: disable=broad-except
+            log.exception("Could not paste component into location {}".format(usage_key))
+            return JsonResponse({"error": _('There was a problem pasting your component.')}, status=400)
+        if created_xblock is None:
+            return JsonResponse({"error": _('Your clipboard is empty or invalid.')}, status=400)
+        return JsonResponse(
+            {'locator': str(created_xblock.location), 'courseKey': str(created_xblock.location.course_key)}
+        )
+
     category = request.json['category']
     if isinstance(usage_key, LibraryUsageLocator):
         # Only these categories are supported at this time.

cms/djangoapps/contentstore/views/component.py

@@ -27,6 +27,11 @@ from common.djangoapps.xblock_django.models import XBlockStudioConfigurationFlag
 from cms.djangoapps.contentstore.toggles import use_new_problem_editor
 from openedx.core.lib.xblock_utils import get_aside_from_xblock, is_xblock_aside
 from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration
+try:
+    # Technically this is a django app plugin, so we should not error if it's not installed:
+    import openedx.core.djangoapps.content_staging.api as content_staging_api
+except ImportError:
+    content_staging_api = None
 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
 
@@ -185,6 +190,12 @@ def container_handler(request, usage_key_string):
                     break
                 index += 1
 
+            # Get the status of the user's clipboard so they can paste components if they have something to paste
+            if content_staging_api:
+                user_clipboard = content_staging_api.get_user_clipboard_json(request.user.id, request)
+            else:
+                user_clipboard = {"content": None}
+
             return render_to_response('container.html', {
                 'language_code': request.LANGUAGE_CODE,
                 'context_course': course,  # Needed only for display of menus at top of page.
@@ -205,7 +216,9 @@ def container_handler(request, usage_key_string):
                 'xblock_info': xblock_info,
                 'draft_preview_link': preview_lms_link,
                 'published_preview_link': lms_link,
-                'templates': CONTAINER_TEMPLATES
+                'templates': CONTAINER_TEMPLATES,
+                # Status of the user's clipboard, exactly as would be returned from the "GET clipboard" REST API.
+                'user_clipboard': user_clipboard,
             })
     else:
         return HttpResponseBadRequest("Only supports HTML requests")

cms/djangoapps/contentstore/views/helpers.py

@@ -3,20 +3,31 @@ Helper methods for Studio views.
 """
 
 import urllib
+from lxml import etree
 from uuid import uuid4
 
 from django.http import HttpResponse
 from django.utils.translation import gettext as _
 from opaque_keys.edx.keys import UsageKey
+from opaque_keys.edx.locator import DefinitionLocator, LocalId
 from xblock.core import XBlock
+from xblock.fields import ScopeIds
+from xblock.runtime import IdGenerator
 from xmodule.modulestore.django import modulestore
 from xmodule.tabs import StaticTab
 
+from cms.djangoapps.contentstore.views.preview import _load_preview_block
 from cms.djangoapps.models.settings.course_grading import CourseGradingModel
 from common.djangoapps.student import auth
 from common.djangoapps.student.roles import CourseCreatorRole, OrgContentCreatorRole
 from openedx.core.toggles import ENTRANCE_EXAMS
 
+try:
+    # Technically this is a django app plugin, so we should not error if it's not installed:
+    import openedx.core.djangoapps.content_staging.api as content_staging_api
+except ImportError:
+    content_staging_api = None
+
 from ..utils import reverse_course_url, reverse_library_url, reverse_usage_url
 
 __all__ = ['event']
@@ -271,6 +282,76 @@ def create_xblock(parent_locator, user, category, display_name, boilerplate=None
         return created_block
 
 
+class ImportIdGenerator(IdGenerator):
+    """
+    Modulestore's IdGenerator doesn't work for importing single blocks as OLX,
+    so we implement our own
+    """
+    def __init__(self, context_key):
+        super().__init__()
+        self.context_key = context_key
+
+    def create_aside(self, definition_id, usage_id, aside_type):
+        """ Generate a new aside key """
+        raise NotImplementedError()
+
+    def create_usage(self, def_id) -> UsageKey:
+        """ Generate a new UsageKey for an XBlock """
+        # Note: Split modulestore will detect this temporary ID and create a new block ID when the XBlock is saved.
+        return self.context_key.make_usage_key(def_id.block_type, LocalId())
+
+    def create_definition(self, block_type, slug=None) -> DefinitionLocator:
+        """ Generate a new definition_id for an XBlock """
+        # Note: Split modulestore will detect this temporary ID and create a new definition ID when the XBlock is saved.
+        return DefinitionLocator(block_type, LocalId(block_type))
+
+
+def import_staged_content_from_user_clipboard(parent_key: UsageKey, request):
+    """
+    Import a block (and any children it has) from "staged" OLX.
+    Does not deal with permissions or REST stuff - do that before calling this.
+
+    Returns the newly created block on success or None if the clipboard is
+    empty.
+    """
+    if not content_staging_api:
+        raise RuntimeError("The required content_staging app is not installed")
+    user_clipboard = content_staging_api.get_user_clipboard(request.user.id)
+    if not user_clipboard:
+        # Clipboard is empty or expired/error/loading
+        return None
+    block_type = user_clipboard.content.block_type
+    olx_str = content_staging_api.get_staged_content_olx(user_clipboard.content.id)
+    node = etree.fromstring(olx_str)
+    store = modulestore()
+    with store.bulk_operations(parent_key.course_key):
+        parent_descriptor = store.get_item(parent_key)
+        # Some blocks like drag-and-drop only work here with the full XBlock runtime loaded:
+        parent_xblock = _load_preview_block(request, parent_descriptor)
+        runtime = parent_xblock.runtime
+        # Generate the new ID:
+        id_generator = ImportIdGenerator(parent_key.context_key)
+        def_id = id_generator.create_definition(block_type, user_clipboard.source_usage_key.block_id)
+        usage_id = id_generator.create_usage(def_id)
+        keys = ScopeIds(None, block_type, def_id, usage_id)
+        # parse_xml is a really messy API. We pass both 'keys' and 'id_generator' and, depending on the XBlock, either
+        # one may be used to determine the new XBlock's usage key, and the other will be ignored. e.g. video ignores
+        # 'keys' and uses 'id_generator', but the default XBlock parse_xml ignores 'id_generator' and uses 'keys'.
+        # For children of this block, obviously only id_generator is used.
+        xblock_class = runtime.load_block_type(block_type)
+        temp_xblock = xblock_class.parse_xml(node, runtime, keys, id_generator)
+        if xblock_class.has_children and temp_xblock.children:
+            raise NotImplementedError("We don't yet support pasting XBlocks with children")
+        temp_xblock.parent = parent_key
+        # Store a reference to where this block was copied from, in the 'copied_from_block' field (AuthoringMixin)
+        temp_xblock.copied_from_block = str(user_clipboard.source_usage_key)
+        # Save the XBlock into modulestore. We need to save the block and its parent for this to work:
+        new_xblock = store.update_item(temp_xblock, request.user.id, allow_not_found=True)
+        parent_xblock.children.append(new_xblock.location)
+        store.update_item(parent_xblock, request.user.id)
+        return new_xblock
+
+
 def is_item_in_course_tree(item):
     """
     Check that the item is in the course tree.

cms/djangoapps/contentstore/views/tests/test_clipboard_paste.py

@@ -0,0 +1,62 @@
+"""
+Test the import_staged_content_from_user_clipboard() method, which is used to
+allow users to paste XBlocks that were copied using the staged_content/clipboard
+APIs.
+"""
+from opaque_keys.edx.keys import UsageKey
+from rest_framework.test import APIClient
+from xmodule.modulestore.django import modulestore
+from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
+from xmodule.modulestore.tests.factories import ToyCourseFactory
+
+CLIPBOARD_ENDPOINT = "/api/content-staging/v1/clipboard/"
+XBLOCK_ENDPOINT = "/xblock/"
+
+
+class ClipboardPasteTestCase(ModuleStoreTestCase):
+    """
+    Test Clipboard Paste functionality
+    """
+
+    def _setup_course(self):
+        """ Set up the "Toy Course" and an APIClient for testing clipboard functionality. """
+        # Setup:
+        course_key = ToyCourseFactory.create().id  # See xmodule/modulestore/tests/sample_courses.py
+        client = APIClient()
+        client.login(username=self.user.username, password=self.user_password)
+        return (course_key, client)
+
+    def test_copy_and_paste_video(self):
+        """
+        Test copying a video from the course, and pasting it into the same unit
+        """
+        course_key, client = self._setup_course()
+
+        # Check how many blocks are in the vertical currently
+        parent_key = course_key.make_usage_key("vertical", "vertical_test")  # This is the vertical that holds the video
+        orig_vertical = modulestore().get_item(parent_key)
+        assert len(orig_vertical.children) == 4
+
+        # Copy the video
+        video_key = course_key.make_usage_key("video", "sample_video")
+        copy_response = client.post(CLIPBOARD_ENDPOINT, {"usage_key": str(video_key)}, format="json")
+        assert copy_response.status_code == 200
+
+        # Paste the video
+        paste_response = client.post(XBLOCK_ENDPOINT, {
+            "parent_locator": str(parent_key),
+            "staged_content": "clipboard",
+        }, format="json")
+        assert paste_response.status_code == 200
+        new_block_key = UsageKey.from_string(paste_response.json()["locator"])
+
+        # Now there should be an extra block in the vertical:
+        updated_vertical = modulestore().get_item(parent_key)
+        assert len(updated_vertical.children) == 5
+        assert updated_vertical.children[-1] == new_block_key
+        # And it should match the original:
+        orig_video = modulestore().get_item(video_key)
+        new_video = modulestore().get_item(new_block_key)
+        assert new_video.youtube_id_1_0 == orig_video.youtube_id_1_0
+        # The new block should store a reference to where it was copied from
+        assert new_video.copied_from_block == str(video_key)

cms/lib/xblock/authoring_mixin.py

@@ -8,6 +8,7 @@ import logging
 from django.conf import settings
 from web_fragments.fragment import Fragment
 from xblock.core import XBlock, XBlockMixin
+from xblock.fields import String, Scope
 
 log = logging.getLogger(__name__)
 
@@ -43,3 +44,10 @@ class AuthoringMixin(XBlockMixin):
         fragment.add_javascript_url(self._get_studio_resource_url('/js/xblock/authoring.js'))
         fragment.initialize_js('VisibilityEditorInit')
         return fragment
+
+    copied_from_block = String(
+        # Note: used by the content_staging app. This field is not needed in the LMS.
+        help="ID of the block that this one was copied from, if any. Used when copying and pasting blocks in Studio.",
+        scope=Scope.settings,
+        enforce_type=True,
+    )

cms/static/js/views/pages/container.js

@@ -22,12 +22,14 @@ function($, _, Backbone, gettext, BasePage, ViewUtils, ContainerView, XBlockView
             'click .move-button': 'showMoveXBlockModal',
             'click .delete-button': 'deleteXBlock',
             'click .show-actions-menu-button': 'showXBlockActionsMenu',
-            'click .new-component-button': 'scrollToNewComponentButtons'
+            'click .new-component-button': 'scrollToNewComponentButtons',
+            'click .paste-component-button': 'pasteComponent',
         },
 
         options: {
             collapsedClass: 'is-collapsed',
-            canEdit: true // If not specified, assume user has permission to make changes
+            canEdit: true, // If not specified, assume user has permission to make changes
+            clipboardData: { content: null },
         },
 
         view: 'container_preview',
@@ -100,6 +102,7 @@ function($, _, Backbone, gettext, BasePage, ViewUtils, ContainerView, XBlockView
             }
 
             this.listenTo(Backbone, 'move:onXBlockMoved', this.onXBlockMoved);
+            this.clipboardBroadcastChannel = new BroadcastChannel("studio_clipboard_channel");
         },
 
         getViewParameters: function() {
@@ -147,6 +150,11 @@ function($, _, Backbone, gettext, BasePage, ViewUtils, ContainerView, XBlockView
 
                     // Re-enable Backbone events for any updated DOM elements
                     self.delegateEvents();
+
+                    // Show/hide the paste button
+                    if (!self.isLibraryPage) {
+                        self.initializePasteButton();
+                    }
                 },
                 block_added: options && options.block_added
             });
@@ -182,6 +190,59 @@ function($, _, Backbone, gettext, BasePage, ViewUtils, ContainerView, XBlockView
             }
         },
 
+        initializePasteButton() {
+            if (this.options.canEdit) {
+                // We should have the user's clipboard status.
+                const data = this.options.clipboardData;
+                this.refreshPasteButton(data);
+                // Refresh the status when something is copied on another tab:
+                this.clipboardBroadcastChannel.onmessage = (event) => { this.refreshPasteButton(event.data); };
+            } else {
+                this.$(".paste-component").hide();
+            }
+        },
+
+        /**
+         * Given the latest information about the user's clipboard, hide or show the Paste button as appropriate.
+         */
+        refreshPasteButton(data) {
+            // 'data' is the same data returned by the "get clipboard status" API endpoint
+            // i.e. /api/content-staging/v1/clipboard/
+            if (this.options.canEdit && data.content) {
+                // TODO: check if this is suitable for pasting into a unit
+                this.$(".paste-component").show();
+            } else {
+                this.$(".paste-component").hide();
+            }
+        },
+
+        /** The user has clicked on the "Paste Component button" */
+        pasteComponent(event) {
+            event.preventDefault();
+            // Get the ID of the container (usually a unit/vertical) that we're pasting into:
+            const parentElement = this.findXBlockElement(event.target);
+            const parentLocator = parentElement.data('locator');
+            // Create a placeholder XBlock while we're pasting:
+            const $placeholderEl = $(this.createPlaceholderElement());
+            const addComponentsPanel = $(event.target).closest('.paste-component').prev();
+            const listPanel = addComponentsPanel.prev();
+            const scrollOffset = ViewUtils.getScrollOffset(addComponentsPanel);
+            const placeholderElement = $placeholderEl.appendTo(listPanel);
+
+            // Start showing a "Pasting" notification:
+            ViewUtils.runOperationShowingMessage(gettext('Pasting'), () => {
+                return $.postJSON(this.getURLRoot() + '/', {
+                    parent_locator: parentLocator,
+                    staged_content: "clipboard",
+                }).then((data) => {
+                    this.onNewXBlock(placeholderElement, scrollOffset, false, data);
+                }).fail(() => {
+                    // Remove the placeholder if the paste failed
+                    placeholderElement.remove();
+                });
+            });
+        },
+
         editXBlock: function(event, options) {
             event.preventDefault();
 
@@ -307,6 +368,8 @@ function($, _, Backbone, gettext, BasePage, ViewUtils, ContainerView, XBlockView
                     const status = data.content?.status;
                     if (status === "ready") {
                         // The XBlock has been copied and is ready to use.
+                        this.refreshPasteButton(data); // Update our UI
+                        this.clipboardBroadcastChannel.postMessage(data); // And notify any other open tabs
                         return data;
                     } else if (status === "loading") {
                         // The clipboard is being loaded asynchonously.
@@ -316,6 +379,8 @@ function($, _, Backbone, gettext, BasePage, ViewUtils, ContainerView, XBlockView
                             $.getJSON(clipboardEndpoint, (pollData) => {
                                 const newStatus = pollData.content?.status;
                                 if (newStatus === "ready") {
+                                    this.refreshPasteButton(data);
+                                    this.clipboardBroadcastChannel.postMessage(pollData);
                                     deferred.resolve(pollData);
                                 } else if (newStatus === "loading") {
                                     setTimeout(checkStatus, 1_000);

cms/static/sass/elements/_modules.scss

@@ -320,6 +320,22 @@
     }
   }
 }
+// New "Paste component" menu, shown on the Unit page to users with a component in their clipboard
+.paste-component {
+  margin: $baseline ($baseline/2);
+
+  .paste-component-button {
+    display: block;
+    width: 100%;
+    // Override what we're extending from ui-btn-flat-outline:
+    &.button {
+      font-size: 1.5rem;
+      padding: 10px 0;
+    }
+
+    @extend %ui-btn-flat-outline;
+  }
+}
 
 // outline UI
 // --------------------

cms/templates/container.html

@@ -48,7 +48,8 @@ from openedx.core.djangolib.markup import HTML, Text
             {
                 isUnitPage: ${is_unit_page | n, dump_js_escaped_json},
                 canEdit: true,
-                outlineURL: "${outline_url | n, js_escaped_string}"
+                outlineURL: "${outline_url | n, js_escaped_string}",
+                clipboardData: ${user_clipboard | n, dump_js_escaped_json},
             }
         );
         require(["js/models/xblock_info", "js/views/xblock", "js/views/utils/xblock_utils", "common/js/components/utils/view_utils"], function (XBlockInfo, XBlockView, XBlockUtils, ViewUtils) {

lms/templates/studio_render_children_view.html

@@ -1,3 +1,4 @@
+<%! from django.utils.translation import gettext as _ %>
 % if can_reorder:
     <ol class="reorderable-container">
 % endif
@@ -9,4 +10,10 @@
 % endif
 % if can_add:
 <div class="add-xblock-component new-component-item adding"></div>
+<div class="paste-component" style="display: none;">
+    <button type="button" class="button paste-component-button">
+        <span class="icon fa fa-paste" aria-hidden="true"></span>
+        ${_('Paste Component')}
+    </button>
+</div>
 % endif

openedx/core/djangoapps/content_staging/api.py

@@ -1,4 +1,77 @@
 """
 Public python API for content staging
 """
-# Currently, there is no public API.
+from __future__ import annotations
+
+from django.http import HttpRequest
+
+from .data import StagedContentData, StagedContentStatus, UserClipboardData
+from .models import UserClipboard as _UserClipboard, StagedContent as _StagedContent
+from .serializers import UserClipboardSerializer as _UserClipboardSerializer
+
+
+def get_user_clipboard(user_id: int, only_ready: bool = True) -> UserClipboardData | None:
+    """
+    Get the details of the user's clipboard.
+
+    By default, will only return a value if the clipboard is READY to use for
+    pasting etc. Pass only_ready=False to get the clipboard data regardless.
+
+    To get the actual OLX content, use get_staged_content_olx(content.id)
+    """
+    try:
+        clipboard = _UserClipboard.objects.get(user_id=user_id)
+    except _UserClipboard.DoesNotExist:
+        # This user does not have any content on their clipboard.
+        return None
+    content = clipboard.content
+    if only_ready and content.status != StagedContentStatus.READY:
+        # The clipboard content is LOADING, ERROR, or EXPIRED
+        return None
+    return UserClipboardData(
+        content=StagedContentData(
+            id=content.id,
+            user_id=content.user_id,
+            created=content.created,
+            purpose=content.purpose,
+            status=content.status,
+            block_type=content.block_type,
+            display_name=content.display_name,
+        ),
+        source_usage_key=clipboard.source_usage_key,
+    )
+
+
+def get_user_clipboard_json(user_id: int, request: HttpRequest = None):
+    """
+    Get the detailed status of the user's clipboard, in exactly the same format
+    as returned from the
+        /api/content-staging/v1/clipboard/
+    REST API endpoint. This version of the API is meant for "preloading" that
+    REST API endpoint so it can be embedded in a larger response sent to the
+    user's browser. If you just want to get the clipboard data from python, use
+    get_user_clipboard() instead, since it's fully typed.
+
+    (request is optional; including it will make the "olx_url" absolute instead
+    of relative.)
+    """
+    try:
+        clipboard = _UserClipboard.objects.get(user_id=user_id)
+    except _UserClipboard.DoesNotExist:
+        # This user does not have any content on their clipboard.
+        return {"content": None, "source_usage_key": "", "source_context_title": ""}
+    serializer = _UserClipboardSerializer(clipboard, context={'request': request})
+    return serializer.data
+
+
+def get_staged_content_olx(staged_content_id: int) -> str | None:
+    """
+    Get the OLX (as a string) for the given StagedContent.
+
+    Does not check permissions!
+    """
+    try:
+        sc = _StagedContent.objects.get(pk=staged_content_id)
+        return sc.olx
+    except _StagedContent.DoesNotExist:
+        return None

openedx/core/djangoapps/content_staging/data.py

@@ -0,0 +1,50 @@
+"""
+Public python data types for content staging
+"""
+from attrs import field, frozen, validators
+from datetime import datetime
+
+from django.db.models import TextChoices
+from django.utils.translation import gettext_lazy as _
+from opaque_keys.edx.keys import UsageKey
+
+
+class StagedContentStatus(TextChoices):
+    """ The status of this staged content. """
+    # LOADING: We are actively (asynchronously) writing the OLX and related data into the staging area.
+    # It is not ready to be read.
+    LOADING = "loading", _("Loading")
+    # READY: The content is staged and ready to be read.
+    READY = "ready", _("Ready")
+    # The content has expired and this row can be deleted, along with any associated data.
+    EXPIRED = "expired", _("Expired")
+    # ERROR: The content could not be staged.
+    ERROR = "error", _("Error")
+
+
+# Value of the "purpose" field on StagedContent objects used for clipboards.
+CLIPBOARD_PURPOSE = "clipboard"
+# There may be other valid values of "purpose" which aren't defined within this app.
+
+
+@frozen
+class StagedContentData:
+    """
+    Read-only data model representing StagedContent
+
+    (OLX content that isn't part of any course at the moment)
+    """
+    id: int = field(validator=validators.instance_of(int))
+    user_id: int = field(validator=validators.instance_of(int))
+    created: datetime = field(validator=validators.instance_of(datetime))
+    purpose: str = field(validator=validators.instance_of(str))
+    status: StagedContentStatus = field(validator=validators.in_(StagedContentStatus), converter=StagedContentStatus)
+    block_type: str = field(validator=validators.instance_of(str))
+    display_name: str = field(validator=validators.instance_of(str))
+
+
+@frozen
+class UserClipboardData:
+    """ Read-only data model for User Clipboard data (copied OLX) """
+    content: StagedContentData = field(validator=validators.instance_of(StagedContentData))
+    source_usage_key: UsageKey = field(validator=validators.instance_of(UsageKey))

openedx/core/djangoapps/content_staging/models.py

@@ -12,6 +12,8 @@ from opaque_keys.edx.keys import LearningContextKey
 
 from openedx.core.djangoapps.content.course_overviews.api import get_course_overview_or_none
 
+from .data import CLIPBOARD_PURPOSE, StagedContentStatus
+
 log = logging.getLogger(__name__)
 
 User = get_user_model()
@@ -32,25 +34,13 @@ class StagedContent(models.Model):
     class Meta:
         verbose_name_plural = _("Staged Content")
 
-    class Status(models.TextChoices):
-        """ The status of this staged content. """
-        # LOADING: We are actively (asynchronously) writing the OLX and related data into the staging area.
-        # It is not ready to be read.
-        LOADING = "loading", _("Loading")
-        # READY: The content is staged and ready to be read.
-        READY = "ready", _("Ready")
-        # The content has expired and this row can be deleted, along with any associated data.
-        EXPIRED = "expired", _("Expired")
-        # ERROR: The content could not be staged.
-        ERROR = "error", _("Error")
-
     id = models.AutoField(primary_key=True)
     # The user that created and owns this staged content. Only this user can read it.
     user = models.ForeignKey(User, null=False, on_delete=models.CASCADE)
     created = models.DateTimeField(null=False, auto_now_add=True)
     # What this StagedContent is for (e.g. "clipboard" for clipboard)
     purpose = models.CharField(max_length=64)
-    status = models.CharField(max_length=20, choices=Status.choices)
+    status = models.CharField(max_length=20, choices=StagedContentStatus.choices)
 
     block_type = models.CharField(
         max_length=100,
@@ -83,9 +73,6 @@ class UserClipboard(models.Model):
     is some OLX content that can be used in a course, such as an XBlock, a Unit,
     or a Subsection.
     """
-    # value of the "purpose" field on underlying StagedContent objects
-    PURPOSE = "clipboard"
-
     # The user that copied something. Clipboards are user-specific and
     # previously copied items are not kept.
     user = models.OneToOneField(User, on_delete=models.CASCADE, primary_key=True)
@@ -114,9 +101,9 @@ class UserClipboard(models.Model):
         # These could probably be replaced with constraints in Django 4.1+
         if self.user.id != self.content.user.id:
             raise ValidationError("User ID mismatch.")
-        if self.content.purpose != UserClipboard.PURPOSE:
+        if self.content.purpose != CLIPBOARD_PURPOSE:
             raise ValidationError(
-                f"StagedContent.purpose must be '{UserClipboard.PURPOSE}' to use it as clipboard content."
+                f"StagedContent.purpose must be '{CLIPBOARD_PURPOSE}' to use it as clipboard content."
             )
 
     def save(self, *args, **kwargs):

openedx/core/djangoapps/content_staging/serializers.py

@@ -16,7 +16,7 @@ class StagedContentSerializer(serializers.ModelSerializer):
         model = StagedContent
         fields = [
             'id',
-            'user',
+            'user_id',
             'created',
             'purpose',
             'status',

openedx/core/djangoapps/content_staging/tasks.py

@@ -7,7 +7,8 @@ import logging
 from celery import shared_task
 from celery_utils.logged_task import LoggedTask
 
-from .models import StagedContent, UserClipboard
+from .data import CLIPBOARD_PURPOSE
+from .models import StagedContent
 
 log = logging.getLogger(__name__)
 
@@ -21,5 +22,5 @@ def delete_expired_clipboards(staged_content_ids: list[int]):
     for pk in staged_content_ids:
         # Due to signal handlers deleting asset file objects from S3 or similar,
         # this may be "slow" relative to database speed.
-        StagedContent.objects.get(purpose=UserClipboard.PURPOSE, pk=pk).delete()
+        StagedContent.objects.get(purpose=CLIPBOARD_PURPOSE, pk=pk).delete()
     log.info(f"Successfully deleted StagedContent entries ({','.join(str(x) for x in staged_content_ids)})")

openedx/core/djangoapps/content_staging/tests/test_clipboard.py

@@ -6,12 +6,26 @@ from xml.etree import ElementTree
 
 from rest_framework.test import APIClient
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
-
 from xmodule.modulestore.tests.factories import ToyCourseFactory
 
+from openedx.core.djangoapps.content_staging import api as python_api
+
 
 CLIPBOARD_ENDPOINT = "/api/content-staging/v1/clipboard/"
 
+# OLX of the video in the toy course using course_key.make_usage_key("video", "sample_video")
+SAMPLE_VIDEO_OLX = """
+    <video
+        url_name="sample_video"
+        display_name="default"
+        youtube="0.75:JMD_ifUUfsU,1.00:OEoXaMPEzfM,1.25:AKqURZnYqpk,1.50:DYpADpL7jAY"
+        youtube_id_0_75="JMD_ifUUfsU"
+        youtube_id_1_0="OEoXaMPEzfM"
+        youtube_id_1_25="AKqURZnYqpk"
+        youtube_id_1_5="DYpADpL7jAY"
+    />
+"""
+
 
 class ClipboardTestCase(ModuleStoreTestCase):
     """
@@ -22,6 +36,7 @@ class ClipboardTestCase(ModuleStoreTestCase):
         """
         When a user has no content on their clipboard, we get an empty 200 response
         """
+        ## Test the REST API:
         client = APIClient()
         client.login(username=self.user.username, password=self.user_password)
         response = client.get(CLIPBOARD_ENDPOINT)
@@ -32,6 +47,13 @@ class ClipboardTestCase(ModuleStoreTestCase):
             "source_usage_key": "",
             "source_context_title": ""
         })
+        ## The Python method for getting the API response should be identical:
+        self.assertEqual(
+            response.json(),
+            python_api.get_user_clipboard_json(self.user.id, response.wsgi_request),
+        )
+        # And the pure python API should return None
+        self.assertEqual(python_api.get_user_clipboard(self.user.id), None)
 
     def _setup_course(self):
         """ Set up the "Toy Course" and an APIClient for testing clipboard functionality. """
@@ -49,7 +71,7 @@ class ClipboardTestCase(ModuleStoreTestCase):
 
     def test_copy_video(self):
         """
-        Test copying a video from the course
+        Test copying a video from the course, and retrieve it using the REST API
         """
         course_key, client = self._setup_course()
 
@@ -74,24 +96,36 @@ class ClipboardTestCase(ModuleStoreTestCase):
         olx_response = client.get(olx_url)
         self.assertEqual(olx_response.status_code, 200)
         self.assertEqual(olx_response.get("Content-Type"), "application/vnd.openedx.xblock.v1.video+xml")
-        self.assertXmlEqual(
-            olx_response.content.decode(),
-            """
-            <video
-                url_name="sample_video"
-                display_name="default"
-                youtube="0.75:JMD_ifUUfsU,1.00:OEoXaMPEzfM,1.25:AKqURZnYqpk,1.50:DYpADpL7jAY"
-                youtube_id_0_75="JMD_ifUUfsU"
-                youtube_id_1_0="OEoXaMPEzfM"
-                youtube_id_1_25="AKqURZnYqpk"
-                youtube_id_1_5="DYpADpL7jAY"
-            />
-            """
-        )
+        self.assertXmlEqual(olx_response.content.decode(), SAMPLE_VIDEO_OLX)
 
         # Now if we GET the clipboard again, the GET response should exactly equal the last POST response:
         self.assertEqual(client.get(CLIPBOARD_ENDPOINT).json(), response_data)
 
+    def test_copy_video_python_get(self):
+        """
+        Test copying a video from the course, and retrieve it using the python API
+        """
+        course_key, client = self._setup_course()
+
+        # Copy the video
+        video_key = course_key.make_usage_key("video", "sample_video")
+        response = client.post(CLIPBOARD_ENDPOINT, {"usage_key": str(video_key)}, format="json")
+        self.assertEqual(response.status_code, 200)
+
+        # Get the clipboard status using python:
+        clipboard_data = python_api.get_user_clipboard(self.user.id)
+        self.assertIsNotNone(clipboard_data)
+        self.assertEqual(clipboard_data.source_usage_key, video_key)
+        # source_context_title is not in the python API because it's easy to retrieve a course's name from python code.
+        self.assertEqual(clipboard_data.content.block_type, "video")
+        # To ensure API stability, we are hard-coding these expected values:
+        self.assertEqual(clipboard_data.content.purpose, "clipboard")
+        self.assertEqual(clipboard_data.content.status, "ready")
+        self.assertEqual(clipboard_data.content.display_name, "default")
+        # Test the actual OLX in the clipboard:
+        olx_data = python_api.get_staged_content_olx(clipboard_data.content.id)
+        self.assertXmlEqual(olx_data, SAMPLE_VIDEO_OLX)
+
     def test_copy_html(self):
         """
         Test copying an HTML from the course
@@ -156,6 +190,8 @@ class ClipboardTestCase(ModuleStoreTestCase):
         html_clip_data = response.json()
         self.assertEqual(html_clip_data["source_usage_key"], str(html_key))
         self.assertEqual(html_clip_data["content"]["block_type"], "html")
+        ## The Python method for getting the API response should be identical:
+        self.assertEqual(html_clip_data, python_api.get_user_clipboard_json(self.user.id, response.wsgi_request))
 
         # The OLX link from the video will no longer work:
         self.assertEqual(client.get(old_olx_url).status_code, 404)
@@ -196,4 +232,7 @@ class ClipboardTestCase(ModuleStoreTestCase):
 
     def assertXmlEqual(self, xml_str_a: str, xml_str_b: str) -> bool:
         """ Assert that the given XML strings are equal, ignoring attribute order and some whitespace variations. """
-        self.assertEqual(ElementTree.canonicalize(xml_str_a), ElementTree.canonicalize(xml_str_b))
+        self.assertEqual(
+            ElementTree.canonicalize(xml_str_a, strip_text=True),
+            ElementTree.canonicalize(xml_str_b, strip_text=True),
+        )

openedx/core/djangoapps/content_staging/views.py

@@ -22,6 +22,7 @@ from xmodule import block_metadata_utils
 from xmodule.modulestore.django import modulestore
 from xmodule.modulestore.exceptions import ItemNotFoundError
 
+from .data import CLIPBOARD_PURPOSE, StagedContentStatus
 from .models import StagedContent, UserClipboard
 from .serializers import UserClipboardSerializer, PostToClipboardSerializer
 from .tasks import delete_expired_clipboards
@@ -42,7 +43,7 @@ class StagedContentOLXEndpoint(APIView):
         staged_content = get_object_or_404(StagedContent, pk=id)
         if staged_content.user.id != request.user.id:
             raise PermissionDenied("Users can only access their own staged content")
-        if staged_content.status != StagedContent.Status.READY:
+        if staged_content.status != StagedContentStatus.READY:
             # If the status is LOADING, the OLX may not be generated/valid yet.
             # If the status is ERROR or EXPIRED, this row is no longer usable.
             raise NotFound("The requested content is not available.")
@@ -117,19 +118,19 @@ class ClipboardEndpoint(APIView):
             # Mark all of the user's existing StagedContent rows as EXPIRED
             to_expire = StagedContent.objects.filter(
                 user=request.user,
-                purpose=UserClipboard.PURPOSE,
+                purpose=CLIPBOARD_PURPOSE,
             ).exclude(
-                status=StagedContent.Status.EXPIRED,
+                status=StagedContentStatus.EXPIRED,
             )
             for sc in to_expire:
                 expired_ids.append(sc.id)
-                sc.status = StagedContent.Status.EXPIRED
+                sc.status = StagedContentStatus.EXPIRED
                 sc.save()
             # Insert a new StagedContent row for this
             staged_content = StagedContent.objects.create(
                 user=request.user,
-                purpose=UserClipboard.PURPOSE,
-                status=StagedContent.Status.READY,
+                purpose=CLIPBOARD_PURPOSE,
+                status=StagedContentStatus.READY,
                 block_type=usage_key.block_type,
                 olx=block_data.olx_str,
                 display_name=block_metadata_utils.display_name_with_default(block),

openedx/core/djangoapps/olx_rest_api/test_views.py

@@ -1,8 +1,7 @@
 """
 Test for the OLX REST API app.
 """
-import re
-from xml.dom import minidom
+from xml.etree import ElementTree
 
 from openedx.core.djangolib.testing.utils import skip_unless_cms
 from common.djangoapps.student.roles import CourseStaffRole
@@ -40,17 +39,12 @@ class OlxRestApiTestCase(SharedModuleStoreTestCase):
 
     # Helper methods:
 
-    def assertXmlEqual(self, xml_str_a, xml_str_b):
-        """
-        Assert that the given XML strings are equal,
-        ignoring attribute order and some whitespace variations.
-        """
-        def clean(xml_str):
-            # Collapse repeated whitespace:
-            xml_str = re.sub(r'(\s)\s+', r'\1', xml_str)
-            xml_bytes = xml_str.encode('utf8')
-            return minidom.parseString(xml_bytes).toprettyxml()
-        assert clean(xml_str_a) == clean(xml_str_b)
+    def assertXmlEqual(self, xml_str_a: str, xml_str_b: str) -> bool:
+        """ Assert that the given XML strings are equal, ignoring attribute order and some whitespace variations. """
+        self.assertEqual(
+            ElementTree.canonicalize(xml_str_a, strip_text=True),
+            ElementTree.canonicalize(xml_str_b, strip_text=True),
+        )
 
     def get_olx_response_for_block(self, block_id):
         return self.client.get(f'/api/olx-export/v1/xblock/{block_id}/')

setup.cfg

@@ -162,6 +162,7 @@ isolated_apps =
     openedx.core.djangoapps.agreements
     openedx.core.djangoapps.bookmarks
     openedx.core.djangoapps.content_libraries
+    openedx.core.djangoapps.content_staging
     openedx.core.djangoapps.olx_rest_api
     openedx.core.djangoapps.xblock
     openedx.core.lib.xblock_serializer

xmodule/modulestore/split_mongo/split.py

@@ -1953,9 +1953,12 @@ class SplitMongoModuleStore(SplitBulkWriteMixin, ModuleStoreWriteBase):
         the definition, structure, nor course if they didn't change.
         """
         partitioned_fields = self.partition_xblock_fields_by_scope(block)
+        definition_locator = getattr(block, "definition_locator", None)
+        if definition_locator is None and not allow_not_found:
+            raise AttributeError("block is missing expected definition_locator from caching descriptor system")
         return self._update_item_from_fields(
             user_id, block.location.course_key, BlockKey.from_usage_key(block.location),
-            partitioned_fields, block.definition_locator, allow_not_found, force, **kwargs
+            partitioned_fields, definition_locator, allow_not_found, force, **kwargs
         ) or block
 
     def _update_item_from_fields(self, user_id, course_key, block_key, partitioned_fields,    # pylint: disable=too-many-statements
@@ -2162,7 +2165,8 @@ class SplitMongoModuleStore(SplitBulkWriteMixin, ModuleStoreWriteBase):
         partitioned_fields = self.partition_xblock_fields_by_scope(xblock)
         new_def_data = self._serialize_fields(xblock.category, partitioned_fields[Scope.content])
         is_updated = False
-        if xblock.definition_locator is None or isinstance(xblock.definition_locator.definition_id, LocalId):
+        current_definition_locator = getattr(xblock, "definition_locator", xblock.scope_ids.def_id)
+        if current_definition_locator is None or isinstance(current_definition_locator.definition_id, LocalId):
             xblock.definition_locator = self.create_definition_from_data(
                 course_key, new_def_data, xblock.category, user_id
             )

xmodule/tests/test_unit_block.py

@@ -2,10 +2,9 @@
 Tests for the Unit XBlock
 """
 
-import re
 import unittest
-from xml.dom import minidom
 from unittest.mock import patch
+from xml.etree import ElementTree
 
 from web_fragments.fragment import Fragment
 from xblock.core import XBlock
@@ -74,14 +73,9 @@ class UnitBlockTests(XmlTest, unittest.TestCase):
         """
         assert XBlockCompletionMode.get_mode(UnitBlock) == XBlockCompletionMode.AGGREGATOR
 
-    def assertXmlEqual(self, xml_str_a, xml_str_b):
-        """
-        Assert that the given XML strings are equal,
-        ignoring attribute order and some whitespace variations.
-        """
-        def clean(xml_str):
-            # Collapse repeated whitespace:
-            xml_str = re.sub(r'(\s)\s+', r'\1', xml_str)
-            xml_bytes = xml_str.encode('utf8')
-            return minidom.parseString(xml_bytes).toprettyxml()
-        assert clean(xml_str_a) == clean(xml_str_b)
+    def assertXmlEqual(self, xml_str_a: str, xml_str_b: str) -> bool:
+        """ Assert that the given XML strings are equal, ignoring attribute order and some whitespace variations. """
+        self.assertEqual(
+            ElementTree.canonicalize(xml_str_a, strip_text=True),
+            ElementTree.canonicalize(xml_str_b, strip_text=True),
+        )