feat: Verify collection belongs to library

openedx/core/djangoapps/content_libraries/api.py

@@ -331,7 +331,7 @@ def get_metadata(queryset, text_search=None):
     return libraries
 
 
-def require_permission_for_library_key(library_key, user, permission):
+def require_permission_for_library_key(library_key, user, permission) -> ContentLibrary:
     """
     Given any of the content library permission strings defined in
     openedx.core.djangoapps.content_libraries.permissions,
@@ -341,10 +341,12 @@ def require_permission_for_library_key(library_key, user, permission):
     Raises django.core.exceptions.PermissionDenied if the user doesn't have
     permission.
     """
-    library_obj = ContentLibrary.objects.get_by_key(library_key)
+    library_obj = ContentLibrary.objects.get_by_key(library_key)  # type: ignore[attr-defined]
     if not user.has_perm(permission, obj=library_obj):
         raise PermissionDenied
 
+    return library_obj
+
 
 def get_library(library_key):
     """

openedx/core/djangoapps/content_libraries/collections/rest_api/v1/tests/test_views.py

@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 from openedx_learning.api.authoring_models import Collection
+from opaque_keys.edx.locator import LibraryLocatorV2
 
 from openedx.core.djangolib.testing.utils import skip_unless_cms
 from openedx.core.djangoapps.content_libraries.tests.base import ContentLibrariesRestApiTest
@@ -31,9 +32,6 @@ def setUp(self):
         self.lib1 = ContentLibrary.objects.get(slug="test-lib-col-1")
         self.lib2 = ContentLibrary.objects.get(slug="test-lib-col-2")
 
-        print("self.lib1", self.lib1)
-        print("self.lib2", self.lib2)
-
         # Create Content Library Collections
         self.col1 = Collection.objects.create(
             learning_package_id=self.lib1.learning_package.id,
@@ -78,6 +76,24 @@ def test_get_library_collection(self):
             )
             assert resp.status_code == 403
 
+    def test_get_invalid_library_collection(self):
+        """
+        Test retrieving a an invalid Content Library Collection or one that does not exist
+        """
+        # Fetch collection that belongs to a different library, it should fail
+        resp = self.client.get(
+            URL_LIB_COLLECTION.format(lib_key=self.lib1.library_key, collection_id=self.col3.id)
+        )
+
+        assert resp.status_code == 404
+
+        # Fetch collection with invalid ID provided, it should fail
+        resp = self.client.get(
+            URL_LIB_COLLECTION.format(lib_key=self.lib1.library_key, collection_id=123)
+        )
+
+        assert resp.status_code == 404
+
     def test_list_library_collections(self):
         """
         Test listing Content Library Collections
@@ -100,6 +116,15 @@ def test_list_library_collections(self):
             resp = self.client.get(URL_LIB_COLLECTIONS.format(lib_key=self.lib1.library_key))
             assert resp.status_code == 403
 
+    def test_list_invalid_library_collections(self):
+        """
+        Test listing invalid Content Library Collections
+        """
+        invalid_key = LibraryLocatorV2.from_string("lib:DoesNotExist:NE1")
+        resp = self.client.get(URL_LIB_COLLECTIONS.format(lib_key=invalid_key))
+
+        assert resp.status_code == 404
+
     def test_create_library_collection(self):
         """
         Test creating a Content Library Collection
@@ -134,6 +159,28 @@ def test_create_library_collection(self):
 
             assert resp.status_code == 403
 
+    def test_create_invalid_library_collection(self):
+        """
+        Test creating an invalid Content Library Collection
+        """
+        post_data_missing_title = {
+            "description": "Description for Collection 4",
+        }
+        resp = self.client.post(
+            URL_LIB_COLLECTIONS.format(lib_key=self.lib1.library_key), post_data_missing_title, format="json"
+        )
+
+        assert resp.status_code == 400
+
+        post_data_missing_desc = {
+            "title": "Collection 4",
+        }
+        resp = self.client.post(
+            URL_LIB_COLLECTIONS.format(lib_key=self.lib1.library_key), post_data_missing_desc, format="json"
+        )
+
+        assert resp.status_code == 400
+
     def test_update_library_collection(self):
         """
         Test updating a Content Library Collection
@@ -171,6 +218,31 @@ def test_update_library_collection(self):
 
             assert resp.status_code == 403
 
+    def test_update_invalid_library_collection(self):
+        """
+        Test updating an invalid Content Library Collection or one that does not exist
+        """
+        patch_data = {
+            "title": "Collection 3 Updated",
+        }
+        # Update collection that belongs to a different library, it should fail
+        resp = self.client.patch(
+            URL_LIB_COLLECTION.format(lib_key=self.lib1.library_key, collection_id=self.col3.id),
+            patch_data,
+            format="json"
+        )
+
+        assert resp.status_code == 404
+
+        # Update collection with invalid ID provided, it should fail
+        resp = self.client.patch(
+            URL_LIB_COLLECTION.format(lib_key=self.lib1.library_key, collection_id=123),
+            patch_data,
+            format="json"
+        )
+
+        assert resp.status_code == 404
+
     def test_delete_library_collection(self):
         """
         Test deleting a Content Library Collection

openedx/core/djangoapps/content_libraries/collections/rest_api/v1/views.py

@@ -6,7 +6,6 @@
 
 from django.http import Http404
 
-# from rest_framework.generics import GenericAPIView
 from rest_framework.response import Response
 from rest_framework.viewsets import ModelViewSet
 from rest_framework.status import HTTP_405_METHOD_NOT_ALLOWED
@@ -30,19 +29,37 @@ class LibraryCollectionsView(ModelViewSet):
 
     serializer_class = ContentLibraryCollectionSerializer
 
+    def _verify_and_fetch_library_collection(self, library_key, collection_id, user, permission) -> Collection | None:
+        """
+        Verify that the collection belongs to the library and the user has the correct permissions
+        """
+        try:
+            library_obj = api.require_permission_for_library_key(library_key, user, permission)
+        except api.ContentLibraryNotFound:
+            raise Http404
+
+        collection = None
+        if library_obj.learning_package_id:
+            collection = authoring_api.get_learning_package_collections(
+                library_obj.learning_package_id
+            ).filter(id=collection_id).first()
+        return collection
+
     def retrieve(self, request, lib_key_str, pk=None):
         """
         Retrieve the Content Library Collection
         """
-        try:
-            collection = authoring_api.get_collection(pk)
-        except Collection.DoesNotExist as exc:
-            raise Http404 from exc
+        library_key = LibraryLocatorV2.from_string(lib_key_str)
 
         # Check if user has permissions to view this collection by checking if
         # user has permission to view the Content Library it belongs to
-        library_key = LibraryLocatorV2.from_string(lib_key_str)
-        api.require_permission_for_library_key(library_key, request.user, permissions.CAN_VIEW_THIS_CONTENT_LIBRARY)
+        collection = self._verify_and_fetch_library_collection(
+            library_key, pk, request.user, permissions.CAN_VIEW_THIS_CONTENT_LIBRARY
+        )
+
+        if not collection:
+            raise Http404
+
         serializer = self.get_serializer(collection)
         return Response(serializer.data)
 
@@ -53,8 +70,13 @@ def list(self, request, lib_key_str):
         # Check if user has permissions to view collections by checking if user
         # has permission to view the Content Library they belong to
         library_key = LibraryLocatorV2.from_string(lib_key_str)
-        api.require_permission_for_library_key(library_key, request.user, permissions.CAN_VIEW_THIS_CONTENT_LIBRARY)
-        content_library = api.get_library(library_key)
+        try:
+            content_library = api.require_permission_for_library_key(
+                library_key, request.user, permissions.CAN_VIEW_THIS_CONTENT_LIBRARY
+            )
+        except api.ContentLibraryNotFound:
+            raise Http404
+
         collections = authoring_api.get_learning_package_collections(content_library.learning_package.id)
         serializer = self.get_serializer(collections, many=True)
         return Response(serializer.data)
@@ -66,10 +88,15 @@ def create(self, request, lib_key_str):
         # Check if user has permissions to create a collection in the Content Library
         # by checking if user has permission to edit the Content Library
         library_key = LibraryLocatorV2.from_string(lib_key_str)
-        api.require_permission_for_library_key(library_key, request.user, permissions.CAN_EDIT_THIS_CONTENT_LIBRARY)
+        try:
+            content_library = api.require_permission_for_library_key(
+                library_key, request.user, permissions.CAN_EDIT_THIS_CONTENT_LIBRARY
+            )
+        except api.ContentLibraryNotFound:
+            raise Http404
+
         create_serializer = ContentLibraryCollectionCreateOrUpdateSerializer(data=request.data)
         create_serializer.is_valid(raise_exception=True)
-        content_library = api.get_library(library_key)
         collection = authoring_api.create_collection(
             content_library.learning_package.id,
             create_serializer.validated_data["title"],
@@ -83,15 +110,15 @@ def partial_update(self, request, lib_key_str, pk=None):
         """
         Update a Collection that belongs to a Content Library
         """
+        library_key = LibraryLocatorV2.from_string(lib_key_str)
         # Check if user has permissions to update a collection in the Content Library
         # by checking if user has permission to edit the Content Library
-        library_key = LibraryLocatorV2.from_string(lib_key_str)
-        api.require_permission_for_library_key(library_key, request.user, permissions.CAN_EDIT_THIS_CONTENT_LIBRARY)
+        collection = self._verify_and_fetch_library_collection(
+            library_key, pk, request.user, permissions.CAN_EDIT_THIS_CONTENT_LIBRARY
+        )
 
-        try:
-            collection = authoring_api.get_collection(pk)
-        except Collection.DoesNotExist as exc:
-            raise Http404 from exc
+        if not collection:
+            raise Http404
 
         update_serializer = ContentLibraryCollectionCreateOrUpdateSerializer(
             collection, data=request.data, partial=True