feat: add update taxonomy orgs rest api [FC-0036]

openedx/core/djangoapps/content_tagging/api.py

@@ -58,6 +58,9 @@ def set_taxonomy_orgs(
     If not `all_orgs`, the taxonomy is associated with each org in the `orgs` list. If that list is empty, the
     taxonomy is not associated with any orgs.
     """
+    if taxonomy.system_defined:
+        raise ValueError("Cannot set orgs for a system-defined taxonomy")
+
     TaxonomyOrg.objects.filter(
         taxonomy=taxonomy,
         rel_type=relationship,

openedx/core/djangoapps/content_tagging/rest_api/v1/serializers.py

@@ -2,6 +2,8 @@
 API Serializers for content tagging org
 """
 
+from __future__ import annotations
+
 from rest_framework import serializers, fields
 
 from openedx_tagging.core.tagging.rest_api.v1.serializers import (
@@ -21,3 +23,29 @@ class TaxonomyOrgListQueryParamsSerializer(TaxonomyListQueryParamsSerializer):
         queryset=Organization.objects.all(),
         required=False,
     )
+
+
+class TaxonomyUpdateOrgBodySerializer(serializers.Serializer):
+    """
+    Serializer for the body params for the update orgs action
+    """
+
+    orgs: fields.Field = serializers.SlugRelatedField(
+        many=True,
+        slug_field="short_name",
+        queryset=Organization.objects.all(),
+        required=False,
+    )
+
+    all_orgs: fields.Field = serializers.BooleanField(required=False)
+
+    def validate(self, attrs: dict) -> dict:
+        """
+        Validate the serializer data
+        """
+        if bool(attrs.get("orgs") is not None) == bool(attrs.get("all_orgs")):
+            raise serializers.ValidationError(
+                "You must specify either orgs or all_orgs, but not both."
+            )
+
+        return attrs

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

@@ -40,6 +40,7 @@
 
 TAXONOMY_ORG_LIST_URL = "/api/content_tagging/v1/taxonomies/"
 TAXONOMY_ORG_DETAIL_URL = "/api/content_tagging/v1/taxonomies/{pk}/"
+TAXONOMY_ORG_UPDATE_ORG_URL = "/api/content_tagging/v1/taxonomies/{pk}/orgs/"
 OBJECT_TAG_UPDATE_URL = "/api/content_tagging/v1/object_tags/{object_id}/?taxonomy={taxonomy_id}"
 TAXONOMY_TEMPLATE_URL = "/api/content_tagging/v1/taxonomies/import/{filename}"
 
@@ -1044,6 +1045,121 @@ def _test_api_call(self, **kwargs) -> None:
             assert response.status_code == status.HTTP_404_NOT_FOUND
 
 
+@skip_unless_cms
+@ddt.ddt
+class TestTaxonomyUpdateOrg(TestTaxonomyObjectsMixin, APITestCase):
+    """
+    Test cases for updating orgs from taxonomies
+    """
+
+    def test_update_org(self) -> None:
+        """
+        Tests that taxonomy admin can add/remove orgs from a taxonomy
+        """
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tA1.pk)
+        self.client.force_authenticate(user=self.staff)
+
+        response = self.client.put(url, {"orgs": [self.orgB.short_name, self.orgX.short_name]}, format="json")
+        assert response.status_code == status.HTTP_200_OK
+
+        # Check that the orgs were updated
+        taxonomy_orgs = TaxonomyOrg.objects.filter(taxonomy=self.tA1)
+        assert taxonomy_orgs.count() == 2
+        assert taxonomy_orgs[0].org == self.orgB
+        assert taxonomy_orgs[1].org == self.orgX
+
+    def test_update_all_org(self) -> None:
+        """
+        Tests that taxonomy admin can associate a taxonomy to all orgs
+        """
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tA1.pk)
+        self.client.force_authenticate(user=self.staff)
+
+        response = self.client.put(url, {"all_orgs": True}, format="json")
+        assert response.status_code == status.HTTP_200_OK
+
+        # Check that the orgs were updated
+        taxonomy_orgs = TaxonomyOrg.objects.filter(taxonomy=self.tA1)
+        assert taxonomy_orgs.count() == 1
+        assert taxonomy_orgs[0].org is None
+
+    def test_update_no_org(self) -> None:
+        """
+        Tests that taxonomy admin can associate a taxonomy no orgs
+        """
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tA1.pk)
+        self.client.force_authenticate(user=self.staff)
+
+        response = self.client.put(url, {"orgs": []}, format="json")
+
+        assert response.status_code == status.HTTP_200_OK
+
+        # Check that the orgs were updated
+        taxonomy_orgs = TaxonomyOrg.objects.filter(taxonomy=self.tA1)
+        assert taxonomy_orgs.count() == 0
+
+    @ddt.data(
+        (True, ["orgX"], "Using both all_orgs and orgs parameters should throw error"),
+        (False, None, "Using neither all_orgs or orgs parameter should throw error"),
+        (None, None, "Using neither all_orgs or orgs parameter should throw error"),
+        (False, 'InvalidOrg', "Passing an invalid org should throw error"),
+    )
+    @ddt.unpack
+    def test_update_org_invalid_inputs(self, all_orgs: bool, orgs: list[str], reason: str) -> None:
+        """
+        Tests if passing both or none of all_orgs and orgs parameters throws error
+        """
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tA1.pk)
+        self.client.force_authenticate(user=self.staff)
+
+        # Set body cleaning empty values
+        body = {k: v for k, v in {"all_orgs": all_orgs, "orgs": orgs}.items() if v is not None}
+        response = self.client.put(url, body, format="json")
+        assert response.status_code == status.HTTP_400_BAD_REQUEST, reason
+
+        # Check that the orgs didn't change
+        taxonomy_orgs = TaxonomyOrg.objects.filter(taxonomy=self.tA1)
+        assert taxonomy_orgs.count() == 1
+        assert taxonomy_orgs[0].org == self.orgA
+
+    def test_update_org_system_defined(self) -> None:
+        """
+        Tests that is not possible to change the orgs associated with a system defined taxonomy
+        """
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.st1.pk)
+        self.client.force_authenticate(user=self.staff)
+
+        response = self.client.put(url, {"orgs": [self.orgA.short_name]}, format="json")
+        assert response.status_code in [status.HTTP_403_FORBIDDEN, status.HTTP_400_BAD_REQUEST]
+
+        # Check that the orgs didn't change
+        taxonomy_orgs = TaxonomyOrg.objects.filter(taxonomy=self.st1)
+        assert taxonomy_orgs.count() == 1
+        assert taxonomy_orgs[0].org is None
+
+    @ddt.data(
+        "staffA",
+        "content_creatorA",
+        "instructorA",
+        "library_staffA",
+        "course_instructorA",
+        "course_staffA",
+        "library_userA",
+    )
+    def test_update_org_no_perm(self, user_attr: str) -> None:
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tA1.pk)
+        user = getattr(self, user_attr)
+        self.client.force_authenticate(user=user)
+
+        response = self.client.put(url, {"orgs": []}, format="json")
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+        # Check that the orgs didn't change
+        taxonomy_orgs = TaxonomyOrg.objects.filter(taxonomy=self.tA1)
+        assert taxonomy_orgs.count() == 1
+        assert taxonomy_orgs[0].org == self.orgA
+
+
 class TestObjectTagMixin(TestTaxonomyObjectsMixin):
     """
     Sets up data for testing ObjectTags.

openedx/core/djangoapps/content_tagging/rest_api/v1/views.py

@@ -3,15 +3,18 @@
 """
 
 from openedx_tagging.core.tagging.rest_api.v1.views import ObjectTagView, TaxonomyView
-
+from rest_framework.decorators import action
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.response import Response
 
 from ...api import (
     create_taxonomy,
     get_taxonomies,
     get_taxonomies_for_org,
+    set_taxonomy_orgs,
 )
 from ...rules import get_admin_orgs
-from .serializers import TaxonomyOrgListQueryParamsSerializer
+from .serializers import TaxonomyOrgListQueryParamsSerializer, TaxonomyUpdateOrgBodySerializer
 from .filters import ObjectTagTaxonomyOrgFilterBackend, UserOrgFilterBackend
 
 
@@ -61,6 +64,26 @@ def perform_create(self, serializer):
         user_admin_orgs = get_admin_orgs(self.request.user)
         serializer.instance = create_taxonomy(**serializer.validated_data, orgs=user_admin_orgs)
 
+    @action(detail=True, methods=["put"])
+    def orgs(self, request, **_kwargs) -> Response:
+        """
+        Export a taxonomy.
+        """
+        taxonomy = self.get_object()
+        perm = "oel_tagging.update_orgs"
+        if not request.user.has_perm(perm, taxonomy):
+            raise PermissionDenied("You do not have permission to update the orgs associated with this taxonomy.")
+        body = TaxonomyUpdateOrgBodySerializer(
+            data=request.data,
+        )
+        body.is_valid(raise_exception=True)
+        orgs = body.validated_data.get("orgs")
+        all_orgs: bool = body.validated_data.get("all_orgs", False)
+
+        set_taxonomy_orgs(taxonomy=taxonomy, all_orgs=all_orgs, orgs=orgs)
+
+        return Response()
+
 
 class ObjectTagOrgView(ObjectTagView):
     """

openedx/core/djangoapps/content_tagging/rules.py

@@ -282,6 +282,7 @@ def can_change_taxonomy_tag(user: UserType, tag: oel_tagging.Tag | None = None)
 rules.set_perm("oel_tagging.delete_taxonomy", can_change_taxonomy)
 rules.set_perm("oel_tagging.view_taxonomy", can_view_taxonomy)
 rules.set_perm("oel_tagging.export_taxonomy", can_view_taxonomy)
+rules.add_perm("oel_tagging.update_orgs", oel_tagging.is_taxonomy_admin)
 
 # Tag
 rules.set_perm("oel_tagging.add_tag", can_change_taxonomy_tag)