refactor: remove ContentObjectTag model and related functions

cms/djangoapps/contentstore/views/component.py

@@ -29,7 +29,7 @@
 from cms.djangoapps.contentstore.xblock_storage_handlers.view_handlers import load_services_for_studio
 from openedx.core.lib.xblock_utils import get_aside_from_xblock, is_xblock_aside
 from openedx.core.djangoapps.discussions.models import DiscussionsConfiguration
-from openedx.core.djangoapps.content_tagging.api import get_content_tags
+from openedx.core.djangoapps.content_tagging.api import get_object_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
 
@@ -533,7 +533,7 @@ def get_unit_tags(usage_key):
     which already provides this grouping + sorting logic.
     """
     # Get content tags from content tagging API
-    content_tags = get_content_tags(usage_key)
+    content_tags = get_object_tags(str(usage_key))
 
     # Group content tags by taxonomy
     taxonomy_dict = {}

openedx/core/djangoapps/content_tagging/api.py

@@ -4,12 +4,11 @@
 from __future__ import annotations
 
 import openedx_tagging.core.tagging.api as oel_tagging
-from django.db.models import Q, QuerySet, Exists, OuterRef
+from django.db.models import QuerySet, Exists, OuterRef
 from openedx_tagging.core.tagging.models import Taxonomy
 from organizations.models import Organization
 
-from .models import ContentObjectTag, TaxonomyOrg
-from .types import ContentKey
+from .models import TaxonomyOrg
 
 
 def create_taxonomy(
@@ -125,56 +124,6 @@ def get_unassigned_taxonomies(enabled=True) -> QuerySet:
     )
 
 
-def get_content_tags(
-    object_key: ContentKey,
-    taxonomy_id: int | None = None,
-) -> QuerySet:
-    """
-    Generates a list of content tags for a given object.
-
-    Pass taxonomy to limit the returned object_tags to a specific taxonomy.
-    """
-    return oel_tagging.get_object_tags(
-        object_id=str(object_key),
-        taxonomy_id=taxonomy_id,
-        object_tag_class=ContentObjectTag,
-    )
-
-
-def tag_content_object(
-    object_key: ContentKey,
-    taxonomy: Taxonomy,
-    tags: list,
-) -> QuerySet:
-    """
-    This is the main API to use when you want to add/update/delete tags from a content object (e.g. an XBlock or
-    course).
-
-    It works one "Taxonomy" at a time, i.e. one field at a time, so you can set call it with taxonomy=Keywords,
-    tags=["gravity", "newton"] to replace any "Keywords" [Taxonomy] tags on the given content object with "gravity" and
-    "newton". Doing so to change the "Keywords" Taxonomy won't affect other Taxonomy's tags (other fields) on the
-    object, such as "Language: [en]" or "Difficulty: [hard]".
-
-    If it's a free-text taxonomy, then the list should be a list of tag values.
-    Otherwise, it should be a list of existing Tag IDs.
-
-    Raises ValueError if the proposed tags are invalid for this taxonomy.
-    Preserves existing (valid) tags, adds new (valid) tags, and removes omitted (or invalid) tags.
-    """
-    if not taxonomy.system_defined:
-        # We require that this taxonomy is linked to the content object's "org" or linked to "all orgs" (None):
-        org_short_name = object_key.org  # type: ignore
-        if not taxonomy.taxonomyorg_set.filter(Q(org__short_name=org_short_name) | Q(org=None)).exists():
-            raise ValueError(f"The specified Taxonomy is not enabled for the content object's org ({org_short_name})")
-    oel_tagging.tag_object(
-        taxonomy=taxonomy,
-        tags=tags,
-        object_id=str(object_key),
-        object_tag_class=ContentObjectTag,
-    )
-    return get_content_tags(object_key, taxonomy_id=taxonomy.id)
-
-
 # Expose the oel_tagging APIs
 
 get_taxonomy = oel_tagging.get_taxonomy
@@ -183,3 +132,5 @@ def tag_content_object(
 get_object_tag_counts = oel_tagging.get_object_tag_counts
 delete_object_tags = oel_tagging.delete_object_tags
 resync_object_tags = oel_tagging.resync_object_tags
+get_object_tags = oel_tagging.get_object_tags
+tag_object = oel_tagging.tag_object

openedx/core/djangoapps/content_tagging/migrations/0008_remove_content_object_tag.py

@@ -0,0 +1,16 @@
+# Generated by Django 3.2.23 on 2024-01-30 21:15
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('content_tagging', '0007_system_defined_org_2'),
+    ]
+
+    operations = [
+        migrations.DeleteModel(
+            name='ContentObjectTag',
+        ),
+    ]

openedx/core/djangoapps/content_tagging/models/__init__.py

@@ -3,5 +3,4 @@
 """
 from .base import (
     TaxonomyOrg,
-    ContentObjectTag,
 )

openedx/core/djangoapps/content_tagging/models/base.py

@@ -3,13 +3,10 @@
 """
 from __future__ import annotations
 
-from django.core.exceptions import ValidationError
 from django.db import models
 from django.db.models import Q, QuerySet
 from django.utils.translation import gettext as _
-from opaque_keys import InvalidKeyError
-from opaque_keys.edx.keys import LearningContextKey, UsageKey
-from openedx_tagging.core.tagging.models import ObjectTag, Taxonomy
+from openedx_tagging.core.tagging.models import Taxonomy
 from organizations.models import Organization
 
 
@@ -80,33 +77,3 @@ def get_organizations(
         if rels.filter(org=None).exists():
             return list(Organization.objects.all())
         return [rel.org for rel in rels]
-
-
-class ContentObjectTag(ObjectTag):
-    """
-    ObjectTag that requires an LearningContextKey or BlockUsageLocator as the object ID.
-    """
-
-    class Meta:
-        proxy = True
-
-    @property
-    def object_key(self) -> UsageKey | LearningContextKey:
-        """
-        Returns the object ID parsed as a UsageKey or LearningContextKey.
-        Raises InvalidKeyError object_id cannot be parse into one of those key types.
-
-        Returns None if there's no object_id.
-        """
-        try:
-            return LearningContextKey.from_string(self.object_id)
-        except InvalidKeyError:
-            return UsageKey.from_string(self.object_id)
-
-    def clean(self):
-        super().clean()
-        # Make sure that object_id is a valid key
-        try:
-            self.object_key
-        except InvalidKeyError as err:
-            raise ValidationError("object_id is not a valid opaque key string.") from err

openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py

@@ -108,6 +108,7 @@ def _setUp_library(self):
             title="Library Org A",
             description="This is a library from Org A",
         )
+        self.libraryA = str(self.content_libraryA.key)
 
     def _setUp_users(self):
         """
@@ -125,10 +126,16 @@ def _setUp_users(self):
 
         self.staffA = User.objects.create(
             username="staffA",
-            email="userA@example.com",
+            email="staffA@example.com",
         )
         update_org_role(self.staff, OrgStaffRole, self.staffA, [self.orgA.short_name])
 
+        self.staffB = User.objects.create(
+            username="staffB",
+            email="[email protected]",
+        )
+        update_org_role(self.staff, OrgStaffRole, self.staffB, [self.orgB.short_name])
+
         self.content_creatorA = User.objects.create(
             username="content_creatorA",
             email="[email protected]",
@@ -1366,7 +1373,7 @@ class TestObjectTagViewSet(TestObjectTagMixin, APITestCase):
     """
 
     @ddt.data(
-        # userA and userS are staff in courseA and can tag using enabled taxonomies
+        # staffA and staff are staff in courseA and can tag using enabled taxonomies
         ("user", "tA1", ["Tag 1"], status.HTTP_403_FORBIDDEN),
         ("staffA", "tA1", ["Tag 1"], status.HTTP_200_OK),
         ("staff", "tA1", ["Tag 1"], status.HTTP_200_OK),
@@ -1451,7 +1458,7 @@ def test_tag_course_invalid(self, user_attr, taxonomy_attr):
         assert response.status_code == status.HTTP_400_BAD_REQUEST
 
     @ddt.data(
-        # userA and userS are staff in courseA (owner of xblockA) and can tag using any taxonomies
+        # staffA and staff are staff in courseA (owner of xblockA) and can tag using any taxonomies
         ("user", "tA1", ["Tag 1"], status.HTTP_403_FORBIDDEN),
         ("staffA", "tA1", ["Tag 1"], status.HTTP_200_OK),
         ("staff", "tA1", ["Tag 1"], status.HTTP_200_OK),
@@ -1534,6 +1541,129 @@ def test_tag_xblock_invalid(self, user_attr, taxonomy_attr):
         response = self.client.put(url, {"tags": ["invalid"]}, format="json")
         assert response.status_code == status.HTTP_400_BAD_REQUEST
 
+    @ddt.data(
+        # staffA and staff are staff in libraryA and can tag using enabled taxonomies
+        ("user", "tA1", ["Tag 1"], status.HTTP_403_FORBIDDEN),
+        ("staffA", "tA1", ["Tag 1"], status.HTTP_200_OK),
+        ("staff", "tA1", ["Tag 1"], status.HTTP_200_OK),
+        ("user", "tA1", [], status.HTTP_403_FORBIDDEN),
+        ("staffA", "tA1", [], status.HTTP_200_OK),
+        ("staff", "tA1", [], status.HTTP_200_OK),
+        ("user", "multiple_taxonomy", ["Tag 1", "Tag 2"], status.HTTP_403_FORBIDDEN),
+        ("staffA", "multiple_taxonomy", ["Tag 1", "Tag 2"], status.HTTP_200_OK),
+        ("staff", "multiple_taxonomy", ["Tag 1", "Tag 2"], status.HTTP_200_OK),
+        ("user", "open_taxonomy", ["tag1"], status.HTTP_403_FORBIDDEN),
+        ("staffA", "open_taxonomy", ["tag1"], status.HTTP_200_OK),
+        ("staff", "open_taxonomy", ["tag1"], status.HTTP_200_OK),
+    )
+    @ddt.unpack
+    def test_tag_library(self, user_attr, taxonomy_attr, tag_values, expected_status):
+        """
+        Tests that only staff and org level users can tag libraries
+        """
+        user = getattr(self, user_attr)
+        self.client.force_authenticate(user=user)
+
+        taxonomy = getattr(self, taxonomy_attr)
+
+        url = OBJECT_TAG_UPDATE_URL.format(object_id=self.libraryA, taxonomy_id=taxonomy.pk)
+
+        response = self.client.put(url, {"tags": tag_values}, format="json")
+
+        assert response.status_code == expected_status
+        if status.is_success(expected_status):
+            tags_by_taxonomy = response.data[str(self.libraryA)]["taxonomies"]
+            if tag_values:
+                response_taxonomy = tags_by_taxonomy[0]
+                assert response_taxonomy["name"] == taxonomy.name
+                response_tags = response_taxonomy["tags"]
+                assert [t["value"] for t in response_tags] == tag_values
+            else:
+                assert tags_by_taxonomy == []  # No tags are set from any taxonomy
+
+            # Check that re-fetching the tags returns what we set
+            new_response = self.client.get(url, format="json")
+            assert status.is_success(new_response.status_code)
+            assert new_response.data == response.data
+
+    @ddt.data(
+        "staffA",
+        "staff",
+    )
+    def test_tag_library_disabled_taxonomy(self, user_attr):
+        """
+        Nobody can use disabled taxonomies to tag objects
+        """
+        user = getattr(self, user_attr)
+        self.client.force_authenticate(user=user)
+
+        disabled_taxonomy = self.tA2
+        assert disabled_taxonomy.enabled is False
+
+        url = OBJECT_TAG_UPDATE_URL.format(object_id=self.libraryA, taxonomy_id=disabled_taxonomy.pk)
+        response = self.client.put(url, {"tags": ["Tag 1"]}, format="json")
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    @ddt.data(
+        ("staffA", "tA1"),
+        ("staff", "tA1"),
+        ("staffA", "multiple_taxonomy"),
+        ("staff", "multiple_taxonomy"),
+    )
+    @ddt.unpack
+    def test_tag_library_invalid(self, user_attr, taxonomy_attr):
+        """
+        Tests that nobody can add invalid tags to a library using a closed taxonomy
+        """
+        user = getattr(self, user_attr)
+        self.client.force_authenticate(user=user)
+
+        taxonomy = getattr(self, taxonomy_attr)
+
+        url = OBJECT_TAG_UPDATE_URL.format(object_id=self.libraryA, taxonomy_id=taxonomy.pk)
+
+        response = self.client.put(url, {"tags": ["invalid"]}, format="json")
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    @ddt.data(
+        ("staff", status.HTTP_200_OK),
+        ("staffA", status.HTTP_403_FORBIDDEN),
+        ("staffB", status.HTTP_403_FORBIDDEN),
+    )
+    @ddt.unpack
+    def test_tag_cross_org(self, user_attr, expected_status):
+        """
+        Tests that only global admins can add a taxonomy from orgA to an object from orgB
+        """
+        user = getattr(self, user_attr)
+        self.client.force_authenticate(user=user)
+
+        url = OBJECT_TAG_UPDATE_URL.format(object_id=self.courseB, taxonomy_id=self.tA1.pk)
+
+        response = self.client.put(url, {"tags": ["Tag 1"]}, format="json")
+
+        assert response.status_code == expected_status
+
+    @ddt.data(
+        ("staff", status.HTTP_200_OK),
+        ("staffA", status.HTTP_403_FORBIDDEN),
+        ("staffB", status.HTTP_403_FORBIDDEN),
+    )
+    @ddt.unpack
+    def test_tag_no_org(self, user_attr, expected_status):
+        """
+        Tests that only global admins can add a no-org taxonomy to an object
+        """
+        user = getattr(self, user_attr)
+        self.client.force_authenticate(user=user)
+
+        url = OBJECT_TAG_UPDATE_URL.format(object_id=self.courseA, taxonomy_id=self.ot1.pk)
+
+        response = self.client.put(url, {"tags": []}, format="json")
+
+        assert response.status_code == expected_status
+
     @ddt.data(
         "courseB",
         "xblockB",
@@ -1567,7 +1697,21 @@ def test_tag_unauthorized(self, objectid_attr):
 
         assert response.status_code == status.HTTP_401_UNAUTHORIZED
 
+    def test_tag_invalid_object(self):
+        """
+        Test that we cannot tag an object that is not a CouseKey, LibraryLocatorV2 or UsageKey
+        """
+        url = OBJECT_TAG_UPDATE_URL.format(object_id='invalid_key', taxonomy_id=self.tA1.pk)
+        self.client.force_authenticate(user=self.staff)
+
+        response = self.client.put(url, {"tags": ["Tag 1"]}, format="json")
+
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
     def test_get_tags(self):
+        """
+        Test that we can get tags for an object
+        """
         self.client.force_authenticate(user=self.staffA)
         taxonomy = self.multiple_taxonomy
         tag_values = ["Tag 1", "Tag 2"]
@@ -1607,7 +1751,7 @@ def test_object_tags_query_count(self):
         """
         object_key = self.courseA
         object_id = str(object_key)
-        tagging_api.tag_content_object(object_key=object_key, taxonomy=self.t1, tags=["anvil", "android"])
+        tagging_api.tag_object(object_id=object_id, taxonomy=self.t1, tags=["anvil", "android"])
         expected_tags = [
             {"value": "android", "lineage": ["ALPHABET", "android"], "can_delete_objecttag": True},
             {"value": "anvil", "lineage": ["ALPHABET", "anvil"], "can_delete_objecttag": True},