feat: paste tags when pasting xblocks with tag data

cms/djangoapps/contentstore/helpers.py

@@ -345,6 +345,14 @@ def _import_xml_node_to_parent(
     else:
         for child_node in child_nodes:
             _import_xml_node_to_parent(child_node, new_xblock, store, user_id=user_id)
+
+    # ToDo: Check the better place to call it
+    # I tried to call it in the xml_block.py in the parse_xml() function,
+    # but the usage_key is not persited yet there
+    from cms.lib.xblock.tagging.tagged_block_mixin import TaggedBlockMixin
+    if isinstance(new_xblock, TaggedBlockMixin):
+        new_xblock.add_tags_from_xml()
+
     return new_xblock
 
 

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

@@ -7,6 +7,7 @@
 from django.test import LiveServerTestCase
 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
@@ -15,6 +16,7 @@
 from cms.djangoapps.contentstore.utils import reverse_usage_url
 from openedx.core.lib.blockstore_api.tests.base import BlockstoreAppTestMixin
 from openedx.core.djangoapps.content_libraries import api as library_api
+from openedx.core.djangoapps.content_tagging import api as tagging_api
 from blockstore.apps import api as blockstore_api
 
 CLIPBOARD_ENDPOINT = "/api/content-staging/v1/clipboard/"
@@ -144,6 +146,78 @@ 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")
+
+        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")
+        # Removing a tag is causing tag_object to fail
+        # tag_removed = Tag.objects.create(taxonomy=taxonomy, value="tag_removed")
+        tagging_api.tag_object(
+            object_id=str(unit_key),
+            taxonomy=taxonomy_all_org,
+            # tags=["tag_1", "tag_2", "tag_removed"],
+            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"],
+        )
+        all_org_tags = 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
+
+        # tag_removed.delete()
+        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]) == '<ObjectTag> ' \
+            'block-v1:org.1+course_1+Destination_Course+type@vertical+block@vertical1: test_taxonomy=tag_1'
+        assert str(tags[1]) == '<ObjectTag> ' \
+            'block-v1:org.1+course_1+Destination_Course+type@vertical+block@vertical1: 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,5 +1,5 @@
 # lint-amnesty, pylint: disable=missing-module-docstring
-from urllib.parse import quote
+from urllib.parse import quote, unquote
 
 
 class TaggedBlockMixin:
@@ -55,3 +55,25 @@ def add_xml_to_node(self, node):
         """
         super().add_xml_to_node(node)
         self.add_tags_to_node(node)
+
+    def add_tags_from_xml(self):
+        """
+        Parse and add tag data from xml
+        """
+        # 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.xml_attributes.get('tags-v1', None) if self.xml_attributes else None
+        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)

openedx/core/djangoapps/content_tagging/api.py

@@ -12,7 +12,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(
@@ -166,6 +167,30 @@ def get_all_object_tags(
     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

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

openedx/core/djangoapps/content_tagging/types.py

@@ -10,7 +10,9 @@
 from openedx_tagging.core.tagging.models import ObjectTag, Taxonomy
 
 ContentKey = Union[LibraryLocatorV2, CourseKey, UsageKey]
+ContextKey = Union[LibraryLocatorV2, CourseKey]
 
 ObjectTagByTaxonomyIdDict = Dict[int, List[ObjectTag]]
 ObjectTagByObjectIdDict = Dict[str, ObjectTagByTaxonomyIdDict]
 TaxonomyDict = Dict[int, Taxonomy]
+TagValuesByTaxonomyExportIdDict = Dict[str, List[str]]

openedx/core/djangoapps/content_tagging/utils.py

@@ -7,11 +7,13 @@
 from opaque_keys import InvalidKeyError
 from opaque_keys.edx.keys import CourseKey, UsageKey
 from opaque_keys.edx.locator import LibraryLocatorV2
+from openedx_tagging.core.tagging.models import Taxonomy
 from organizations.models import Organization
 
 from openedx.core.djangoapps.content_libraries.api import get_libraries_for_user
 
-from .types import ContentKey
+from .types import ContentKey, ContextKey
+from .models import TaxonomyOrg
 
 
 def get_content_key_from_string(key_str: str) -> ContentKey:
@@ -30,11 +32,10 @@ def get_content_key_from_string(key_str: str) -> ContentKey:
                 raise ValueError("object_id must be a CourseKey, LibraryLocatorV2 or a UsageKey") from usage_key_error
 
 
-def get_context_key_from_key_string(key_str: str) -> CourseKey | LibraryLocatorV2:
+def get_context_key_from_key(content_key: ContentKey) -> ContextKey:
     """
-    Get context key from an key string
+    Get context key from an key
     """
-    content_key = get_content_key_from_string(key_str)
     # If the content key is a CourseKey or a LibraryLocatorV2, return it
     if isinstance(content_key, (CourseKey, LibraryLocatorV2)):
         return content_key
@@ -48,6 +49,31 @@ def get_context_key_from_key_string(key_str: str) -> CourseKey | LibraryLocatorV
     raise ValueError("context must be a CourseKey or a LibraryLocatorV2")
 
 
+def get_context_key_from_key_string(key_str: str) -> ContextKey:
+    """
+    Get context key from an key string
+    """
+    content_key = get_content_key_from_string(key_str)
+    return get_context_key_from_key(content_key)
+
+
+def check_taxonomy_context_key_org(taxonomy: Taxonomy, context_key: ContextKey) -> bool:
+    """
+    Returns True if the given taxonomy can tag a object with the given context_key.
+    """
+    if not context_key.org:
+        return False
+
+    is_all_org, taxonomy_orgs = TaxonomyOrg.get_organizations(taxonomy)
+
+    if is_all_org:
+        return True
+
+    # 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
+
+
 class TaggingRulesCache:
     """
     Caches data required for computing rules for the duration of the request.
@@ -57,7 +83,7 @@ def __init__(self):
         """
         Initializes the request cache.
         """
-        self.request_cache = RequestCache('openedx.core.djangoapps.content_tagging.rules')
+        self.request_cache = RequestCache('openedx.core.djangoapps.content_tagging.utils')
 
     def get_orgs(self, org_names: list[str] | None = None) -> list[Organization]:
         """
@@ -100,3 +126,6 @@ def get_library_orgs(self, user, org_names: list[str]) -> list[Organization]:
         return [
             library_orgs[org_name] for org_name in org_names if org_name in library_orgs
         ]
+
+
+rules_cache = TaggingRulesCache()