Enforce limit on number of tags per object

openedx_tagging/core/tagging/models/base.py

@@ -404,6 +404,20 @@ def _find_object_tag_index(tag_ref, object_tags) -> int:
                 -1,
             )
 
+        def _check_current_tag_count() -> None:
+            """
+            Checks if the current count of tags for the object is less than 100
+            """
+            # Exclude self.id to avoid counting the tags that are going to be updated
+            current_count = ObjectTag.objects.filter(object_id=object_id).exclude(taxonomy_id=self.id).count()
+
+            if current_count >= 100:
+                raise ValueError(
+                    _(f"Object ({object_id}) already have 100 or more tags.")
+                )
+
+        _check_current_tag_count()
+
         if not isinstance(tags, list):
             raise ValueError(_(f"Tags must be a list, not {type(tags).__name__}."))
 

openedx_tagging/core/tagging/rest_api/v1/views.py

@@ -5,6 +5,7 @@
 from django.http import Http404
 from rest_framework import mixins
 from rest_framework.exceptions import MethodNotAllowed, PermissionDenied, ValidationError
+from rest_framework.response import Response
 from rest_framework.viewsets import GenericViewSet, ModelViewSet
 
 from ...api import create_taxonomy, get_object_tags, get_taxonomies, get_taxonomy, tag_object
@@ -171,21 +172,17 @@ def perform_create(self, serializer) -> None:
 
 class ObjectTagView(mixins.RetrieveModelMixin, mixins.UpdateModelMixin, mixins.ListModelMixin, GenericViewSet):
     """
-    View to retrieve paginated ObjectTags for a provided Object ID (object_id).
+    View to retrieve ObjectTags for a provided Object ID (object_id).
 
     **Retrieve Parameters**
         * object_id (required): - The Object ID to retrieve ObjectTags for.
 
     **Retrieve Query Parameters**
         * taxonomy (optional) - PK of taxonomy to filter ObjectTags for.
-        * page (optional) - Page number of paginated results.
-        * page_size (optional) - Number of results included in each page.
 
     **Retrieve Example Requests**
         GET api/tagging/v1/object_tags/:object_id
         GET api/tagging/v1/object_tags/:object_id?taxonomy=1
-        GET api/tagging/v1/object_tags/:object_id?taxonomy=1&page=2
-        GET api/tagging/v1/object_tags/:object_id?taxonomy=1&page=2&page_size=10
 
     **Retrieve Query Returns**
         * 200 - Success
@@ -232,8 +229,7 @@ def get_queryset(self) -> models.QuerySet:
 
     def retrieve(self, request, object_id=None):
         """
-        Retrieve ObjectTags that belong to a given object_id and
-        return paginated results.
+        Retrieve ObjectTags that belong to a given object_id
 
         Note: We override `retrieve` here instead of `list` because we are
         passing in the Object ID (object_id) in the path (as opposed to passing
@@ -243,14 +239,12 @@ def retrieve(self, request, object_id=None):
         behavior we want.
         """
         object_tags = self.get_queryset()
-        paginated_object_tags = self.paginate_queryset(object_tags)
-        serializer = ObjectTagSerializer(paginated_object_tags, many=True)
-        return self.get_paginated_response(serializer.data)
+        serializer = ObjectTagSerializer(object_tags, many=True)
+        return Response(serializer.data)
 
     def update(self, request, object_id, partial=False):
         """
-        Update ObjectTags that belong to a given object_id and
-        return the list of these ObjecTags paginated.
+        Update ObjectTags that belong to a given object_id
 
         Pass a list of Tag ids or Tag values to be applied to an object id in the
         body `tag` parameter. Passing an empty list will remove all tags from

tests/openedx_tagging/core/tagging/test_api.py

@@ -54,7 +54,7 @@ def test_bad_taxonomy_class(self) -> None:
     def test_get_taxonomy(self) -> None:
         tax1 = tagging_api.get_taxonomy(1)
         assert tax1 == self.taxonomy
-        no_tax = tagging_api.get_taxonomy(10)
+        no_tax = tagging_api.get_taxonomy(200)
         assert no_tax is None
 
     def test_get_taxonomies(self) -> None:
@@ -70,7 +70,7 @@ def test_get_taxonomies(self) -> None:
             self.taxonomy,
             self.system_taxonomy,
             self.user_taxonomy,
-        ]
+        ] + self.dummy_taxonomies
         assert str(enabled[0]) == f"<Taxonomy> ({tax1.id}) Enabled"
         assert str(enabled[1]) == "<Taxonomy> (5) Import Taxonomy Test"
         assert str(enabled[2]) == "<Taxonomy> (-1) Languages"
@@ -92,7 +92,7 @@ def test_get_taxonomies(self) -> None:
             self.taxonomy,
             self.system_taxonomy,
             self.user_taxonomy,
-        ]
+        ] + self.dummy_taxonomies
 
     @override_settings(LANGUAGES=test_languages)
     def test_get_tags(self) -> None:
@@ -539,6 +539,35 @@ def test_tag_object_model_system_taxonomy_invalid(self) -> None:
             exc.exception
         )
 
+    def test_tag_object_limit(self) -> None:
+        """
+        Test that the tagging limit is enforced.
+        """
+        # The user can add up to 100 tags to a object
+        for taxonomy in self.dummy_taxonomies:
+            tagging_api.tag_object(
+                taxonomy,
+                ["Dummy Tag"],
+                "object_1",
+            )
+
+        # Adding a new tag should fail
+        with self.assertRaises(ValueError) as exc:
+            tagging_api.tag_object(
+                self.taxonomy,
+                ["Eubacteria"],
+                "object_1",
+            )
+            assert "already have 100 or more tags" in str(exc.exception)
+
+        # Updating existing tags should work
+        for taxonomy in self.dummy_taxonomies:
+            tagging_api.tag_object(
+                taxonomy,
+                ["New Dummy Tag"],
+                "object_1",
+            )
+
     def test_get_object_tags(self) -> None:
         # Alpha tag has no taxonomy
         alpha = ObjectTag(object_id="abc")

tests/openedx_tagging/core/tagging/test_models.py

@@ -93,6 +93,17 @@ def setUp(self):
             get_tag("System Tag 4"),
         ]
 
+        self.dummy_taxonomies = []
+        for i in range(100):
+            taxonomy = Taxonomy.objects.create(name=f"ZZ Dummy Taxonomy {i:03}", allow_free_text=True)
+            ObjectTag.objects.create(
+                object_id="limit_tag_count",
+                taxonomy=taxonomy,
+                _name=taxonomy.name,
+                _value="Dummy Tag",
+            )
+            self.dummy_taxonomies.append(taxonomy)
+
     def setup_tag_depths(self):
         """
         Annotate our tags with depth so we can compare them.

tests/openedx_tagging/core/tagging/test_views.py

@@ -406,7 +406,7 @@ def _object_permission(_user, object_id: str) -> bool:
             """
             Everyone have object permission on object_id "abc"
             """
-            return object_id == "abc"
+            return object_id in ("abc", "limit_tag_count")
 
         super().setUp()
 
@@ -449,28 +449,39 @@ def _object_permission(_user, object_id: str) -> bool:
             )
 
         # Free-Text Taxonomies created by taxonomy admins, each linked
-        # to 200 ObjectTags
+        # to 10 ObjectTags
         self.open_taxonomy_enabled = Taxonomy.objects.create(name="Enabled Free-Text Taxonomy", allow_free_text=True)
         self.open_taxonomy_disabled = Taxonomy.objects.create(
             name="Disabled Free-Text Taxonomy", allow_free_text=True, enabled=False
         )
-        for i in range(200):
+        for i in range(10):
             ObjectTag.objects.create(object_id="abc", taxonomy=self.open_taxonomy_enabled, _value=f"Free Text {i}")
             ObjectTag.objects.create(object_id="abc", taxonomy=self.open_taxonomy_disabled, _value=f"Free Text {i}")
 
+        self.dummy_taxonomies = []
+        for i in range(100):
+            taxonomy = Taxonomy.objects.create(name=f"Dummy Taxonomy {i}", allow_free_text=True)
+            ObjectTag.objects.create(
+                object_id="limit_tag_count",
+                taxonomy=taxonomy,
+                _name=taxonomy.name,
+                _value="Dummy Tag"
+            )
+            self.dummy_taxonomies.append(taxonomy)
+
         # Override the object permission for the test
         rules.set_perm("oel_tagging.change_objecttag_objectid", _object_permission)
 
     @ddt.data(
-        (None, "abc", status.HTTP_403_FORBIDDEN, None, None),
-        ("user", "abc", status.HTTP_200_OK, 461, 10),
-        ("staff", "abc", status.HTTP_200_OK, 461, 10),
-        (None, "non-existing-id", status.HTTP_403_FORBIDDEN, None, None),
-        ("user", "non-existing-id", status.HTTP_200_OK, 0, 0),
-        ("staff", "non-existing-id", status.HTTP_200_OK, 0, 0),
+        (None, "abc", status.HTTP_403_FORBIDDEN, None),
+        ("user", "abc", status.HTTP_200_OK, 81),
+        ("staff", "abc", status.HTTP_200_OK, 81),
+        (None, "non-existing-id", status.HTTP_403_FORBIDDEN, None),
+        ("user", "non-existing-id", status.HTTP_200_OK, 0),
+        ("staff", "non-existing-id", status.HTTP_200_OK, 0),
     )
     @ddt.unpack
-    def test_retrieve_object_tags(self, user_attr, object_id, expected_status, expected_count, expected_results):
+    def test_retrieve_object_tags(self, user_attr, object_id, expected_status, expected_count):
         """
         Test retrieving object tags
         """
@@ -484,18 +495,16 @@ def test_retrieve_object_tags(self, user_attr, object_id, expected_status, expec
         assert response.status_code == expected_status
 
         if status.is_success(expected_status):
-            assert response.data.get("count") == expected_count
-            assert response.data.get("results") is not None
-            assert len(response.data.get("results")) == expected_results
+            assert len(response.data) == expected_count
 
     @ddt.data(
-        (None, "abc", status.HTTP_403_FORBIDDEN, None, None),
-        ("user", "abc", status.HTTP_200_OK, 20, 10),
-        ("staff", "abc", status.HTTP_200_OK, 20, 10),
+        (None, "abc", status.HTTP_403_FORBIDDEN, None),
+        ("user", "abc", status.HTTP_200_OK, 20),
+        ("staff", "abc", status.HTTP_200_OK, 20),
     )
     @ddt.unpack
     def test_retrieve_object_tags_taxonomy_queryparam(
-        self, user_attr, object_id, expected_status, expected_count, expected_results
+        self, user_attr, object_id, expected_status, expected_count
     ):
         """
         Test retrieving object tags for specific taxonomies provided
@@ -509,11 +518,8 @@ def test_retrieve_object_tags_taxonomy_queryparam(
         response = self.client.get(url, {"taxonomy": self.enabled_taxonomy.pk})
         assert response.status_code == expected_status
         if status.is_success(expected_status):
-            assert response.data.get("count") == expected_count
-            assert response.data.get("results") is not None
-            assert len(response.data.get("results")) == expected_results
-            object_tags = response.data.get("results")
-            for object_tag in object_tags:
+            assert len(response.data) == expected_count
+            for object_tag in response.data:
                 assert object_tag.get("is_valid") is True
                 assert object_tag.get("taxonomy_id") == self.enabled_taxonomy.pk
 
@@ -537,51 +543,6 @@ def test_retrieve_object_tags_invalid_taxonomy_queryparam(self, user_attr, objec
         response = self.client.get(url, {"taxonomy": 123123})
         assert response.status_code == expected_status
 
-    @ddt.data(
-        # Page 1, default page size 10, total count 200, returns 10 results
-        (None, 1, None, status.HTTP_403_FORBIDDEN, None, None),
-        ("user", 1, None, status.HTTP_200_OK, 200, 10),
-        ("staff", 1, None, status.HTTP_200_OK, 200, 10),
-        # Page 2, default page size 10, total count 200, returns 10 results
-        (None, 2, None, status.HTTP_403_FORBIDDEN, None, None),
-        ("user", 2, None, status.HTTP_200_OK, 200, 10),
-        ("staff", 2, None, status.HTTP_200_OK, 200, 10),
-        # Page 21, default page size 10, total count 200, no more results
-        (None, 21, None, status.HTTP_403_FORBIDDEN, None, None),
-        ("user", 21, None, status.HTTP_404_NOT_FOUND, None, None),
-        ("staff", 21, None, status.HTTP_404_NOT_FOUND, None, None),
-        # Page 3, page size 2, total count 200, returns 2 results
-        (None, 3, 2, status.HTTP_403_FORBIDDEN, 200, 2),
-        ("user", 3, 2, status.HTTP_200_OK, 200, 2),
-        ("staff", 3, 2, status.HTTP_200_OK, 200, 2),
-    )
-    @ddt.unpack
-    def test_retrieve_object_tags_pagination(
-        self, user_attr, page, page_size, expected_status, expected_count, expected_results
-    ):
-        """
-        Test pagination for retrieve object tags
-        """
-        url = OBJECT_TAGS_RETRIEVE_URL.format(object_id="abc")
-
-        if user_attr:
-            user = getattr(self, user_attr)
-            self.client.force_authenticate(user=user)
-
-        query_params = {"taxonomy": self.open_taxonomy_enabled.pk, "page": page}
-        if page_size:
-            query_params["page_size"] = page_size
-
-        response = self.client.get(url, query_params)
-        assert response.status_code == expected_status
-        if status.is_success(expected_status):
-            assert response.data.get("count") == expected_count
-            assert response.data.get("results") is not None
-            assert len(response.data.get("results")) == expected_results
-            object_tags = response.data.get("results")
-            for object_tag in object_tags:
-                assert object_tag.get("taxonomy_id") == self.open_taxonomy_enabled.pk
-
     @ddt.data(
         (None, "POST", status.HTTP_403_FORBIDDEN),
         (None, "PATCH", status.HTTP_403_FORBIDDEN),
@@ -660,8 +621,8 @@ def test_tag_object(self, user_attr, taxonomy_attr, tag_values, expected_status)
         response = self.client.put(url, {"tags": tag_values}, format="json")
         assert response.status_code == expected_status
         if status.is_success(expected_status):
-            assert len(response.data.get("results")) == len(tag_values)
-            assert set(t["value"] for t in response.data["results"]) == set(tag_values)
+            assert len(response.data) == len(tag_values)
+            assert set(t["value"] for t in response.data) == set(tag_values)
 
     @ddt.data(
         # Can't add invalid tags to a closed taxonomy
@@ -727,8 +688,8 @@ def test_tag_object_clear(self, user_attr, taxonomy_attr, tag_values, expected_s
         response = self.client.put(url, {"tags": tag_values}, format="json")
         assert response.status_code == expected_status
         if status.is_success(expected_status):
-            assert len(response.data.get("results")) == len(tag_values)
-            assert set(t["value"] for t in response.data["results"]) == set(tag_values)
+            assert len(response.data) == len(tag_values)
+            assert set(t["value"] for t in response.data) == set(tag_values)
 
     @ddt.data(
         # Users and staff can add multiple tags to a allow_multiple=True taxonomy
@@ -764,8 +725,8 @@ def test_tag_object_multiple(self, user_attr, taxonomy_attr, tag_values, expecte
         response = self.client.put(url, {"tags": tag_values}, format="json")
         assert response.status_code == expected_status
         if status.is_success(expected_status):
-            assert len(response.data.get("results")) == len(tag_values)
-            assert set(t["value"] for t in response.data["results"]) == set(tag_values)
+            assert len(response.data) == len(tag_values)
+            assert set(t["value"] for t in response.data) == set(tag_values)
 
     @ddt.data(
         (None, status.HTTP_403_FORBIDDEN),
@@ -782,3 +743,21 @@ def test_tag_object_without_permission(self, user_attr, expected_status):
 
         response = self.client.put(url, {"tags": ["Tag 1"]}, format="json")
         assert response.status_code == expected_status
+        assert not status.is_success(expected_status)  # No success cases here
+
+    def test_tag_object_count_limit(self):
+        """
+        Checks if the limit of 100 tags per object is enforced
+        """
+        object_id = "limit_tag_count"
+        url = OBJECT_TAGS_UPDATE_URL.format(object_id=object_id, taxonomy_id=self.enabled_taxonomy.pk)
+        self.client.force_authenticate(user=self.staff)
+        response = self.client.put(url, {"tags": ["Tag 1"]}, format="json")
+        # Can't add another tag because the object already has 100 tags
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+        # The user can edit the tags that are already on the object
+        for taxonomy in self.dummy_taxonomies:
+            url = OBJECT_TAGS_UPDATE_URL.format(object_id=object_id, taxonomy_id=taxonomy.pk)
+            response = self.client.put(url, {"tags": ["New Tag"]}, format="json")
+            assert response.status_code == status.HTTP_200_OK