feat: v2 catalog contains_content_items view

ENT-9408

enterprise_catalog/apps/api/base/tests/enterprise_catalog_views.py

@@ -0,0 +1,46 @@
+from rest_framework.reverse import reverse
+
+from enterprise_catalog.apps.api.v1.tests.mixins import APITestMixin
+from enterprise_catalog.apps.catalog.models import (
+    CatalogQuery,
+    ContentMetadata,
+    EnterpriseCatalog,
+)
+from enterprise_catalog.apps.catalog.tests.factories import (
+    EnterpriseCatalogFactory,
+)
+
+
+class BaseEnterpriseCatalogViewSetTests(APITestMixin):
+    """
+    Base tests for EnterpriseCatalog view sets.
+    """
+    VERSION = 'v1'
+
+    def setUp(self):
+        super().setUp()
+        # clean up any stale test objects
+        CatalogQuery.objects.all().delete()
+        ContentMetadata.objects.all().delete()
+        EnterpriseCatalog.objects.all().delete()
+
+        self.enterprise_catalog = EnterpriseCatalogFactory(enterprise_uuid=self.enterprise_uuid)
+
+        # Set up catalog.has_learner_access permissions
+        self.set_up_catalog_learner()
+
+    def tearDown(self):
+        super().tearDown()
+        # clean up any stale test objects
+        CatalogQuery.objects.all().delete()
+        ContentMetadata.objects.all().delete()
+        EnterpriseCatalog.objects.all().delete()
+
+    def _get_contains_content_base_url(self, catalog_uuid=None):
+        """
+        Helper to construct the base url for the catalog contains_content_items endpoint
+        """
+        return reverse(
+            f'api:{self.VERSION}:enterprise-catalog-content-contains-content-items',
+            kwargs={'uuid': catalog_uuid or self.enterprise_catalog.uuid},
+        )

enterprise_catalog/apps/api/v1/views/enterprise_catalog_contains_content_items.py

@@ -41,6 +41,14 @@ def get_permission_object(self):
             return str(enterprise_catalog.enterprise_uuid)
         return None
 
+    def catalog_contains_content_items(self, content_keys):
+        """
+        Returns a boolean indicating whether all of the provided content_keys
+        are contained by the catalog record associated with the current request.
+        """
+        enterprise_catalog = self.get_object()
+        return enterprise_catalog.contains_content_keys(content_keys)
+
     # Becuase the edx-rbac perms are built around a part of the URL
     # path, here (the uuid of the catalog), we can utilize per-view caching,
     # rather than per-user caching.
@@ -56,6 +64,6 @@ def contains_content_items(self, request, uuid, course_run_ids, program_uuids, *
         """
         course_run_ids = unquote_course_keys(course_run_ids)
 
-        enterprise_catalog = self.get_object()
-        contains_content_items = enterprise_catalog.contains_content_keys(course_run_ids + program_uuids)
-        return Response({'contains_content_items': contains_content_items})
+        return Response({
+            'contains_content_items': self.catalog_contains_content_items(course_run_ids + program_uuids),
+        })

enterprise_catalog/apps/api/v2/tests/test_enterprise_catalog_views.py

@@ -0,0 +1,159 @@
+import uuid
+from datetime import datetime, timedelta
+from unittest import mock
+
+import ddt
+import pytest
+import pytz
+from rest_framework import status
+
+from enterprise_catalog.apps.api.base.tests.enterprise_catalog_views import (
+    BaseEnterpriseCatalogViewSetTests,
+)
+from enterprise_catalog.apps.catalog.constants import (
+    COURSE,
+    COURSE_RUN,
+    RESTRICTED_RUNS_ALLOWED_KEY,
+)
+from enterprise_catalog.apps.catalog.tests.factories import (
+    ContentMetadataFactory,
+    EnterpriseCatalogFactory,
+    RestrictedCourseMetadataFactory,
+    RestrictedRunAllowedForRestrictedCourseFactory,
+)
+from enterprise_catalog.apps.catalog.utils import localized_utcnow
+
+
+@ddt.ddt
+class EnterpriseCatalogContainsContentItemsTests(BaseEnterpriseCatalogViewSetTests):
+    """
+    Tests for the EnterpriseCatalogViewSetV2, which is permissive of restricted course/run metadata.
+    """
+    VERSION = 'v2'
+
+    def setUp(self):
+        super().setUp()
+
+        self.customer_details_patcher = mock.patch(
+            'enterprise_catalog.apps.catalog.models.EnterpriseCustomerDetails'
+        )
+        self.mock_customer_details = self.customer_details_patcher.start()
+        self.NOW = localized_utcnow()
+        self.mock_customer_details.return_value.last_modified_date = self.NOW
+
+        self.addCleanup(self.customer_details_patcher.stop)
+
+    def test_contains_content_items_unauthorized_non_catalog_learner(self):
+        """
+        Verify the contains_content_items endpoint rejects users that are not catalog learners
+        """
+        self.set_up_invalid_jwt_role()
+        self.remove_role_assignments()
+        url = self._get_contains_content_base_url() + '?course_run_ids=fakeX'
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_contains_content_items_unauthorized_incorrect_jwt_context(self):
+        """
+        Verify the contains_content_items endpoint rejects users that are catalog learners
+        with an incorrect JWT context (i.e., enterprise uuid)
+        """
+        other_customer_catalog = EnterpriseCatalogFactory(enterprise_uuid=uuid.uuid4())
+
+        base_url = self._get_contains_content_base_url(other_customer_catalog.uuid)
+        url = base_url + '?course_run_ids=fakeX'
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+    def test_contains_content_items_implicit_access(self):
+        """
+        Verify the contains_content_items endpoint responds with 200 OK for
+        user with implicit JWT access
+        """
+        self.remove_role_assignments()
+        url = self._get_contains_content_base_url() + '?program_uuids=fakeX'
+        self.assert_correct_contains_response(url, False)
+
+    def test_contains_content_items_no_params(self):
+        """
+        Verify the contains_content_items endpoint errors if no parameters are provided
+        """
+        response = self.client.get(self._get_contains_content_base_url())
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_contains_content_items_not_in_catalogs(self):
+        """
+        Verify the contains_content_items endpoint returns False if the content is not in any associated catalog
+        """
+        self.add_metadata_to_catalog(self.enterprise_catalog, [ContentMetadataFactory()])
+
+        url = self._get_contains_content_base_url() + '?program_uuids=this-is-not-the-uuid-youre-looking-for'
+        self.assert_correct_contains_response(url, False)
+
+    def test_contains_content_items_in_catalogs(self):
+        """
+        Verify the contains_content_items endpoint returns True if the content is in any associated catalog
+        """
+        content_key = 'fake-key+101x'
+        relevant_content = ContentMetadataFactory(content_key=content_key)
+        self.add_metadata_to_catalog(self.enterprise_catalog, [relevant_content])
+
+        url = self._get_contains_content_base_url() + '?course_run_ids=' + content_key
+        self.assert_correct_contains_response(url, True)
+
+    def _create_restricted_course_and_run(self, catalog):
+        """
+        Helper to setup restricted course and run.
+        """
+        content_one = ContentMetadataFactory(content_key='org+key1', content_type=COURSE)
+        restricted_course = RestrictedCourseMetadataFactory.create(
+            content_key='org+key1',
+            content_type=COURSE,
+            unrestricted_parent=content_one,
+            catalog_query=catalog.catalog_query,
+        )
+        restricted_run = ContentMetadataFactory.create(
+            content_key='course-v1:org+key1+restrictedrun',
+            content_type=COURSE_RUN,
+        )
+        restricted_course.restricted_run_allowed_for_restricted_course.set(
+            [restricted_run], clear=True,
+        )
+        return content_one, restricted_course, restricted_run
+
+    def test_contains_catalog_key_restricted_runs_allowed(self):
+        """
+        Tests that a catalog is considered to contain a restricted run.
+        """
+        content_one, _, restricted_run = self._create_restricted_course_and_run(self.enterprise_catalog)
+
+        self.add_metadata_to_catalog(self.enterprise_catalog, [content_one, restricted_run])
+
+        url = self._get_contains_content_base_url() + \
+            f'?course_run_ids={restricted_run.content_key}&get_catalogs_containing_specified_content_ids=true'
+
+        response = self.client.get(url)
+        response_payload = response.json()
+
+        self.assertTrue(response_payload.get('contains_content_items'))
+
+    def test_contains_catalog_key_restricted_run_present_but_not_associated_with_catalog(self):
+        """
+        Tests that a catalog is not considered to contain a restricted run if the
+        run exists in the database but is not explicitly linked to the requested catalog
+        (and even if the parent course *is* linked to the catalog).
+        """
+        other_catalog = EnterpriseCatalogFactory(enterprise_uuid=self.enterprise_uuid)
+
+        content_one, _, restricted_run = self._create_restricted_course_and_run(other_catalog)
+
+        self.add_metadata_to_catalog(self.enterprise_catalog, [content_one])
+        self.add_metadata_to_catalog(other_catalog, [content_one, restricted_run])
+
+        url = self._get_contains_content_base_url() + \
+            f'?course_run_ids={restricted_run.content_key}&get_catalogs_containing_specified_content_ids=true'
+
+        response = self.client.get(url)
+        response_payload = response.json()
+
+        self.assertFalse(response_payload.get('contains_content_items'))

enterprise_catalog/apps/api/v2/urls.py

@@ -4,6 +4,9 @@
 from django.urls import path, re_path
 from rest_framework.routers import DefaultRouter
 
+from enterprise_catalog.apps.api.v2.views.enterprise_catalog_contains_content_items import (
+    EnterpriseCatalogContainsContentItemsV2,
+)
 from enterprise_catalog.apps.api.v2.views.enterprise_catalog_get_content_metadata import (
     EnterpriseCatalogGetContentMetadataV2,
 )
@@ -17,6 +20,7 @@
 router = DefaultRouter()
 
 router.register(r'enterprise-customer', EnterpriseCustomerViewSetV2, basename='enterprise-customer')
+router.register(r'enterprise-catalogs', EnterpriseCatalogContainsContentItemsV2, basename='enterprise-catalog-content')
 
 urlpatterns = [
     re_path(

enterprise_catalog/apps/api/v2/views/enterprise_catalog_contains_content_items.py

@@ -0,0 +1,23 @@
+import logging
+
+from enterprise_catalog.apps.api.v1.views.enterprise_catalog_contains_content_items import (
+    EnterpriseCatalogContainsContentItems,
+)
+
+
+logger = logging.getLogger(__name__)
+
+
+class EnterpriseCatalogContainsContentItemsV2(EnterpriseCatalogContainsContentItems):
+    """
+    Viewset to indicate if given content keys are contained by a catalog, with
+    restricted content taken into account.
+    """
+    def catalog_contains_content_items(self, content_keys):
+        """
+        Returns a boolean indicating whether all of the provided content_keys
+        are contained by the catalog record associated with the current request.
+        Takes restricted content into account.
+        """
+        enterprise_catalog = self.get_object()
+        return enterprise_catalog.contains_content_keys(content_keys, include_restricted=True)