feat: paste tags when pasting xblocks with tag data (#34270)

cms/djangoapps/contentstore/helpers.py

@@ -18,7 +18,6 @@
 from xmodule.contentstore.content import StaticContent
 from xmodule.contentstore.django import contentstore
 from xmodule.exceptions import NotFoundError
-from xmodule.library_content_block import LibraryContentBlock
 from xmodule.modulestore.django import modulestore
 from xmodule.xml_block import XmlMixin
 
@@ -367,14 +366,18 @@ def _import_xml_node_to_parent(
     new_xblock = store.update_item(temp_xblock, user_id, allow_not_found=True)
     parent_xblock.children.append(new_xblock.location)
     store.update_item(parent_xblock, user_id)
-    if isinstance(new_xblock, LibraryContentBlock):
-        # Special case handling for library content. If we need this for other blocks in the future, it can be made into
-        # an API, and we'd call new_block.studio_post_paste() instead of this code.
-        # In this case, we want to pull the children from the library and let library_tools assign their IDs.
-        new_xblock.sync_from_library(upgrade_to_latest=False)
-    else:
+
+    children_handled = False
+    if hasattr(new_xblock, 'studio_post_paste'):
+        # Allow an XBlock to do anything fancy it may need to when pasted from the clipboard.
+        # 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.
+        children_handed = new_xblock.studio_post_paste(store, node)
+
+    if not children_handled:
         for child_node in child_nodes:
             _import_xml_node_to_parent(child_node, new_xblock, store, user_id=user_id)
+
     return new_xblock
 
 

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

@@ -73,6 +73,7 @@
 from common.djangoapps.xblock_django.user_service import DjangoXBlockUserService
 from lms.djangoapps.lms_xblock.mixin import NONSENSICAL_ACCESS_RESTRICTION
 from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration
+from openedx.core.djangoapps.content_tagging import api as tagging_api
 
 from ..component import component_handler, get_component_templates
 from cms.djangoapps.contentstore.xblock_storage_handlers.view_handlers import (
@@ -1106,6 +1107,59 @@ def test_duplicate_library_content_block(self):  # pylint: disable=too-many-stat
         assert dupe_html_2.display_name == "HTML 2 Title (Lib Update)"
         assert dupe_html_2.data == "HTML 2 Content (Lib Update)"
 
+    def test_duplicate_tags(self):
+        """
+        Test that duplicating a tagged XBlock also duplicates its content tags.
+        """
+        source_course = CourseFactory()
+        user = UserFactory.create()
+        source_chapter = BlockFactory(
+            parent=source_course, category="chapter", display_name="Source Chapter"
+        )
+        source_block = BlockFactory(parent=source_chapter, category="html", display_name="Child")
+
+        # Create a couple of taxonomies with tags
+        taxonomyA = tagging_api.create_taxonomy(name="A", export_id="A")
+        taxonomyB = tagging_api.create_taxonomy(name="B", export_id="B")
+        tagging_api.set_taxonomy_orgs(taxonomyA, all_orgs=True)
+        tagging_api.set_taxonomy_orgs(taxonomyB, all_orgs=True)
+        tagging_api.add_tag_to_taxonomy(taxonomyA, "one")
+        tagging_api.add_tag_to_taxonomy(taxonomyA, "two")
+        tagging_api.add_tag_to_taxonomy(taxonomyB, "three")
+        tagging_api.add_tag_to_taxonomy(taxonomyB, "four")
+
+        # Tag the chapter
+        tagging_api.tag_object(str(source_chapter.location), taxonomyA, ["one", "two"])
+        tagging_api.tag_object(str(source_chapter.location), taxonomyB, ["three", "four"])
+
+        # Tag the child block
+        tagging_api.tag_object(str(source_block.location), taxonomyA, ["two"],)
+
+        # Duplicate the chapter (and its children)
+        dupe_location = duplicate_block(
+            parent_usage_key=source_course.location,
+            duplicate_source_usage_key=source_chapter.location,
+            user=user,
+        )
+        dupe_chapter = self.store.get_item(dupe_location)
+        self.assertEqual(len(dupe_chapter.get_children()), 1)
+        dupe_block = dupe_chapter.get_children()[0]
+
+        # Check that the duplicated blocks also duplicated tags
+        expected_chapter_tags = [
+            f'<ObjectTag> {str(dupe_chapter.location)}: A=one',
+            f'<ObjectTag> {str(dupe_chapter.location)}: A=two',
+            f'<ObjectTag> {str(dupe_chapter.location)}: B=four',
+            f'<ObjectTag> {str(dupe_chapter.location)}: B=three',
+        ]
+        dupe_chapter_tags = [str(object_tag) for object_tag in tagging_api.get_object_tags(str(dupe_chapter.location))]
+        assert dupe_chapter_tags == expected_chapter_tags
+        expected_block_tags = [
+            f'<ObjectTag> {str(dupe_block.location)}: A=two',
+        ]
+        dupe_block_tags = [str(object_tag) for object_tag in tagging_api.get_object_tags(str(dupe_block.location))]
+        assert dupe_block_tags == expected_block_tags
+
 
 @ddt.ddt
 class TestMoveItem(ItemTest):

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

@@ -6,13 +6,15 @@
 import ddt
 from opaque_keys.edx.keys import UsageKey
 from rest_framework.test import APIClient
+from openedx_tagging.core.tagging.models import Tag
 from organizations.models import Organization
 from xmodule.modulestore.django import contentstore, modulestore
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase, upload_file_to_course
 from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory, ToyCourseFactory
 
 from cms.djangoapps.contentstore.utils import reverse_usage_url
 from openedx.core.djangoapps.content_libraries import api as library_api
+from openedx.core.djangoapps.content_tagging import api as tagging_api
 
 CLIPBOARD_ENDPOINT = "/api/content-staging/v1/clipboard/"
 XBLOCK_ENDPOINT = "/xblock/"
@@ -141,6 +143,69 @@ def test_copy_and_paste_component(self, block_args):
         # The new block should store a reference to where it was copied from
         assert dest_block.copied_from_block == str(source_block.location)
 
+    def test_copy_and_paste_unit_with_tags(self):
+        """
+        Test copying a unit (vertical) with tags from one course into another
+        """
+        course_key, client = self._setup_course()
+        dest_course = CourseFactory.create(display_name='Destination Course')
+        with self.store.bulk_operations(dest_course.id):
+            dest_chapter = BlockFactory.create(parent=dest_course, category='chapter', display_name='Section')
+            dest_sequential = BlockFactory.create(parent=dest_chapter, category='sequential', display_name='Subsection')
+
+        unit_key = course_key.make_usage_key("vertical", "vertical_test")
+        # Add tags to the unit
+        taxonomy_all_org = tagging_api.create_taxonomy("test_taxonomy", "Test Taxonomy")
+
+        tagging_api.set_taxonomy_orgs(taxonomy_all_org, all_orgs=True)
+        Tag.objects.create(taxonomy=taxonomy_all_org, value="tag_1")
+        Tag.objects.create(taxonomy=taxonomy_all_org, value="tag_2")
+        tagging_api.tag_object(
+            object_id=str(unit_key),
+            taxonomy=taxonomy_all_org,
+            tags=["tag_1", "tag_2"],
+        )
+
+        taxonomy_all_org_removed = tagging_api.create_taxonomy("test_taxonomy_removed", "Test Taxonomy Removed")
+        tagging_api.set_taxonomy_orgs(taxonomy_all_org_removed, all_orgs=True)
+        Tag.objects.create(taxonomy=taxonomy_all_org_removed, value="tag_1")
+        Tag.objects.create(taxonomy=taxonomy_all_org_removed, value="tag_2")
+        tagging_api.tag_object(
+            object_id=str(unit_key),
+            taxonomy=taxonomy_all_org_removed,
+            tags=["tag_1", "tag_2"],
+        )
+        tagging_api.get_object_tags(str(unit_key))
+
+        taxonomy_no_org = tagging_api.create_taxonomy("test_taxonomy_no_org", "Test Taxonomy No Org")
+        Tag.objects.create(taxonomy=taxonomy_no_org, value="tag_1")
+        Tag.objects.create(taxonomy=taxonomy_no_org, value="tag_2")
+        tagging_api.tag_object(
+            object_id=str(unit_key),
+            taxonomy=taxonomy_no_org,
+            tags=["tag_1", "tag_2"],
+        )
+
+        # Copy the unit
+        copy_response = client.post(CLIPBOARD_ENDPOINT, {"usage_key": str(unit_key)}, format="json")
+        assert copy_response.status_code == 200
+
+        taxonomy_all_org_removed.delete()
+
+        # Paste the unit
+        paste_response = client.post(XBLOCK_ENDPOINT, {
+            "parent_locator": str(dest_sequential.location),
+            "staged_content": "clipboard",
+        }, format="json")
+        assert paste_response.status_code == 200
+        dest_unit_key = UsageKey.from_string(paste_response.json()["locator"])
+
+        # Only tags from the taxonomy that is associated with the dest org should be copied
+        tags = list(tagging_api.get_object_tags(str(dest_unit_key)))
+        assert len(tags) == 2
+        assert str(tags[0]) == f'<ObjectTag> {dest_unit_key}: test_taxonomy=tag_1'
+        assert str(tags[1]) == f'<ObjectTag> {dest_unit_key}: test_taxonomy=tag_2'
+
     def test_paste_with_assets(self):
         """
         When pasting into a different course, any required static assets should

cms/lib/xblock/tagging/tagged_block_mixin.py

@@ -1,15 +1,40 @@
-# lint-amnesty, pylint: disable=missing-module-docstring
-from urllib.parse import quote
+"""
+Content tagging functionality for XBlocks.
+"""
+from urllib.parse import quote, unquote
 
 
 class TaggedBlockMixin:
     """
     Mixin containing XML serializing and parsing functionality for tagged blocks
     """
 
-    def serialize_tag_data(self):
+    def __init__(self, *args, **kwargs):
         """
-        Serialize block's tag data to include in the xml, escaping special characters
+        Initialize the tagged xblock.
+        """
+        # We store tags internally, without an XBlock field, because we don't want tags to be stored to the modulestore.
+        # But we do want them persisted on duplicate, copy, or export/import.
+        self.tags_v1 = ""
+
+    @property
+    def tags_v1(self) -> str:
+        """
+        Returns the serialized tags.
+        """
+        return self._tags_v1
+
+    @tags_v1.setter
+    def tags_v1(self, tags: str) -> None:
+        """
+        Returns the serialized tags.
+        """
+        self._tags_v1 = tags
+
+    @classmethod
+    def serialize_tag_data(cls, usage_id):
+        """
+        Serialize a block's tag data to include in the xml, escaping special characters
 
         Example tags:
             LightCast Skills Taxonomy: ["Typing", "Microsoft Office"]
@@ -21,7 +46,7 @@ def serialize_tag_data(self):
         # This import is done here since we import and use TaggedBlockMixin in the cms settings, but the
         # content_tagging app wouldn't have loaded yet, so importing it outside causes an error
         from openedx.core.djangoapps.content_tagging.api import get_object_tags
-        content_tags = get_object_tags(self.scope_ids.usage_id)
+        content_tags = get_object_tags(usage_id)
 
         serialized_tags = []
         taxonomies_and_tags = {}
@@ -45,13 +70,63 @@ def add_tags_to_node(self, node):
         """
         Serialize and add tag data (if any) to node
         """
-        tag_data = self.serialize_tag_data()
+        tag_data = self.serialize_tag_data(self.scope_ids.usage_id)
         if tag_data:
             node.set('tags-v1', tag_data)
 
+    def add_tags_from_field(self):
+        """
+        Parse tags_v1 data and create tags for this block.
+        """
+        # This import is done here since we import and use TaggedBlockMixin in the cms settings, but the
+        # content_tagging app wouldn't have loaded yet, so importing it outside causes an error
+        from openedx.core.djangoapps.content_tagging.api import set_object_tags
+
+        tag_data = self.tags_v1
+        if not tag_data:
+            return
+
+        serialized_tags = tag_data.split(';')
+        taxonomy_and_tags_dict = {}
+        for serialized_tag in serialized_tags:
+            taxonomy_export_id, tags = serialized_tag.split(':')
+            tags = tags.split(',')
+            tag_values = [unquote(tag) for tag in tags]
+            taxonomy_and_tags_dict[taxonomy_export_id] = tag_values
+
+        set_object_tags(self.usage_key, taxonomy_and_tags_dict)
+
     def add_xml_to_node(self, node):
         """
         Include the serialized tag data in XML when adding to node
         """
         super().add_xml_to_node(node)
         self.add_tags_to_node(node)
+
+    def studio_post_duplicate(self, store, source_item) -> bool:
+        """
+        Duplicates content tags from the source_item.
+
+        Returns False to indicate the children have not been handled.
+        """
+        if hasattr(super(), 'studio_post_duplicate'):
+            super().studio_post_duplicate()
+
+        self.tags_v1 = self.serialize_tag_data(source_item.scope_ids.usage_id)
+        self.add_tags_from_field()
+        return False
+
+    def studio_post_paste(self, store, source_node) -> bool:
+        """
+        Copies content tags from the source_node.
+
+        Returns False to indicate the children have not been handled.
+        """
+        if hasattr(super(), 'studio_post_paste'):
+            super().studio_post_paste()
+
+        if 'tags-v1' in source_node.attrib:
+            self.tags_v1 = str(source_node.attrib['tags-v1'])
+
+        self.add_tags_from_field()
+        return False

openedx/core/djangoapps/content_tagging/api.py

@@ -13,7 +13,8 @@
 from organizations.models import Organization
 
 from .models import TaxonomyOrg
-from .types import ObjectTagByObjectIdDict, TaxonomyDict
+from .types import ContentKey, ObjectTagByObjectIdDict, TagValuesByTaxonomyExportIdDict, TaxonomyDict
+from .utils import check_taxonomy_context_key_org, get_context_key_from_key
 
 
 def create_taxonomy(
@@ -161,16 +162,42 @@ def get_all_object_tags(
 
     for object_id, block_tags in groupby(all_object_tags, lambda x: x.object_id):
         grouped_object_tags[object_id] = {}
-        for taxonomy_id, taxonomy_tags in groupby(block_tags, lambda x: x.tag.taxonomy_id):
+        for taxonomy_id, taxonomy_tags in groupby(block_tags, lambda x: x.tag.taxonomy_id if x.tag else 0):
             object_tags_list = list(taxonomy_tags)
             grouped_object_tags[object_id][taxonomy_id] = object_tags_list
 
             if taxonomy_id not in taxonomies:
+                assert object_tags_list[0].tag
+                assert object_tags_list[0].tag.taxonomy
                 taxonomies[taxonomy_id] = object_tags_list[0].tag.taxonomy
 
     return grouped_object_tags, taxonomies
 
 
+def set_object_tags(
+    content_key: ContentKey,
+    object_tags: TagValuesByTaxonomyExportIdDict,
+) -> None:
+    """
+    Sets the tags for the given content object.
+    """
+    context_key = get_context_key_from_key(content_key)
+
+    for taxonomy_export_id, tags_values in object_tags.items():
+        taxonomy = oel_tagging.get_taxonomy_by_export_id(taxonomy_export_id)
+        if not taxonomy:
+            continue
+
+        if not check_taxonomy_context_key_org(taxonomy, context_key):
+            continue
+
+        oel_tagging.tag_object(
+            object_id=str(content_key),
+            taxonomy=taxonomy,
+            tags=tags_values,
+        )
+
+
 # Expose the oel_tagging APIs
 
 get_taxonomy = oel_tagging.get_taxonomy
@@ -181,3 +208,4 @@ def get_all_object_tags(
 resync_object_tags = oel_tagging.resync_object_tags
 get_object_tags = oel_tagging.get_object_tags
 tag_object = oel_tagging.tag_object
+add_tag_to_taxonomy = oel_tagging.add_tag_to_taxonomy

openedx/core/djangoapps/content_tagging/rules.py

@@ -20,10 +20,9 @@
 )
 
 from .models import TaxonomyOrg
-from .utils import get_context_key_from_key_string, TaggingRulesCache
+from .utils import check_taxonomy_context_key_org, get_context_key_from_key_string, rules_cache
 
 
-rules_cache = TaggingRulesCache()
 UserType = Union[django.contrib.auth.models.User, django.contrib.auth.models.AnonymousUser]
 
 
@@ -288,19 +287,12 @@ def can_change_object_tag(
     """
     if oel_tagging.can_change_object_tag(user, perm_obj):
         if perm_obj and perm_obj.taxonomy and perm_obj.object_id:
-            # can_change_object_tag_objectid already checked that object_id is valid and has an org,
-            # so these statements will not fail. But we need to assert to keep the type checker happy.
             try:
                 context_key = get_context_key_from_key_string(perm_obj.object_id)
-                assert context_key.org
-            except (ValueError, AssertionError):
+            except ValueError:
                 return False  # pragma: no cover
 
-            is_all_org, taxonomy_orgs = TaxonomyOrg.get_organizations(perm_obj.taxonomy)
-            if not is_all_org:
-                # Ensure the object_id's org is among the allowed taxonomy orgs
-                object_org = rules_cache.get_orgs([context_key.org])
-                return bool(object_org) and object_org[0] in taxonomy_orgs
+            return check_taxonomy_context_key_org(perm_obj.taxonomy, context_key)
 
         return True
     return False