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,10 +2,13 @@
 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 (
     TaxonomyListQueryParamsSerializer,
+    TaxonomySerializer,
 )
 
 from organizations.models import Organization
@@ -21,3 +24,55 @@ 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
+
+
+class TaxonomyOrgSerializer(TaxonomySerializer):
+    """
+    Serializer for Taxonomy objects inclusing the associated orgs
+    """
+
+    orgs = serializers.SerializerMethodField()
+    all_orgs = serializers.SerializerMethodField()
+
+    def get_orgs(self, obj) -> list[str]:
+        """
+        Return the list of orgs for the taxonomy.
+         """
+        return [taxonomy_org.org.short_name for taxonomy_org in obj.taxonomyorg_set.all() if taxonomy_org.org]
+
+    def get_all_orgs(self, obj) -> bool:
+        """
+        Return True if the taxonomy is associated with all orgs.
+        """
+        return obj.taxonomyorg_set.filter(org__isnull=True).exists()
+
+    class Meta:
+        model = TaxonomySerializer.Meta.model
+        fields = TaxonomySerializer.Meta.fields + ["orgs", "all_orgs"]
+        read_only_fields = ["orgs", "all_orgs"]

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}"
 
@@ -454,7 +455,7 @@ def test_create_taxonomy(self, user_attr: str, expected_status: int) -> None:
 
             # Also checks if the taxonomy was associated with the org
             if user_attr == "staffA":
-                assert TaxonomyOrg.objects.filter(taxonomy=response.data["id"], org=self.orgA).exists()
+                assert response.data["orgs"] == [self.orgA.short_name]
 
 
 @ddt.ddt
@@ -1044,6 +1045,208 @@ 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
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
+        response = self.client.get(url)
+        assert response.data["orgs"] == [self.orgB.short_name, self.orgX.short_name]
+        assert not response.data["all_orgs"]
+
+    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
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
+        response = self.client.get(url)
+        assert response.data["orgs"] == []
+        assert response.data["all_orgs"]
+
+    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
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
+        response = self.client.get(url)
+        assert response.data["orgs"] == []
+        assert not response.data["all_orgs"]
+
+    @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
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
+        response = self.client.get(url)
+        assert response.data["orgs"] == [self.orgA.short_name]
+
+    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
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.st1.pk)
+        response = self.client.get(url)
+        assert response.data["orgs"] == []
+        assert response.data["all_orgs"]
+
+    @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:
+        """
+        Tests that only taxonomy admins can associate orgs to taxonomies
+        """
+        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
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
+        response = self.client.get(url)
+        assert response.data["orgs"] == [self.orgA.short_name]
+
+    def test_update_org_check_permissions_orgA(self) -> None:
+        """
+        Tests that adding an org to a taxonomy allow org level admins to edit it
+        """
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tB1.pk)
+        self.client.force_authenticate(user=self.staffA)
+
+        response = self.client.put(url, {"name": "new name"}, format="json")
+
+        # User staffA can't update metadata from a taxonomy from orgB
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tB1.pk)
+        self.client.force_authenticate(user=self.staff)
+
+        # Add the taxonomy tB1 to orgA
+        response = self.client.put(url, {"orgs": [self.orgA.short_name]}, format="json")
+
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tB1.pk)
+        self.client.force_authenticate(user=self.staffA)
+
+        response = self.client.put(url, {"name": "new name"}, format="json")
+
+        # Now staffA can change the metadata from a tB1 because it's associated with orgA
+        assert response.status_code == status.HTTP_200_OK
+
+    def test_update_org_check_permissions_all_orgs(self) -> None:
+        """
+        Tests that adding an org to all orgs only let taxonomy global admins to edit it
+        """
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
+        self.client.force_authenticate(user=self.staffA)
+
+        response = self.client.put(url, {"name": "new name"}, format="json")
+
+        # User staffA can update metadata from a taxonomy from orgA
+        assert response.status_code == status.HTTP_200_OK
+
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tB1.pk)
+        self.client.force_authenticate(user=self.staff)
+
+        # Add the taxonomy tA1 to all orgs
+        response = self.client.put(url, {"all_orgs": True}, format="json")
+
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tB1.pk)
+        self.client.force_authenticate(user=self.staffA)
+
+        response = self.client.put(url, {"name": "new name"}, format="json")
+
+        # Now staffA can't change the metadata from a tA1 because only global taxonomy admins can edit all orgs
+        # taxonomies
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_update_org_check_permissions_no_orgs(self) -> None:
+        """
+        Tests that remove all orgs from a taxonomy only let taxonomy global admins to edit it
+        """
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
+        self.client.force_authenticate(user=self.staffA)
+
+        response = self.client.put(url, {"name": "new name"}, format="json")
+
+        # User staffA can update metadata from a taxonomy from orgA
+        assert response.status_code == status.HTTP_200_OK
+
+        url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tB1.pk)
+        self.client.force_authenticate(user=self.staff)
+
+        # Remove all orgs from tA1
+        response = self.client.put(url, {"orgs": []}, format="json")
+
+        url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tB1.pk)
+        self.client.force_authenticate(user=self.staffA)
+
+        response = self.client.put(url, {"name": "new name"}, format="json")
+
+        # Now staffA can't change the metadata from a tA1 because only global taxonomy admins can edit no orgs
+        # taxonomies
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+
 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, TaxonomyOrgSerializer, TaxonomyUpdateOrgBodySerializer
 from .filters import ObjectTagTaxonomyOrgFilterBackend, UserOrgFilterBackend
 
 
@@ -36,6 +39,7 @@ class TaxonomyOrgView(TaxonomyView):
     """
 
     filter_backends = [UserOrgFilterBackend]
+    serializer_class = TaxonomyOrgSerializer
 
     def get_queryset(self):
         """
@@ -50,9 +54,11 @@ def get_queryset(self):
         enabled = query_params.validated_data.get("enabled", None)
         org = query_params.validated_data.get("org", None)
         if org:
-            return get_taxonomies_for_org(enabled, org)
+            queryset = get_taxonomies_for_org(enabled, org)
         else:
-            return get_taxonomies(enabled)
+            queryset = get_taxonomies(enabled)
+
+        return queryset.prefetch_related("taxonomyorg_set")
 
     def perform_create(self, serializer):
         """
@@ -61,6 +67,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:
+        """
+        Update the orgs associated with taxonomies.
+        """
+        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)