feat: v2 catalog contains_content_items view

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,157 @@
+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,
+            _json_metadata=content_one.json_metadata,
+        )
+        restricted_run = ContentMetadataFactory.create(
+            content_key='course-v1:org+key1+restrictedrun',
+            parent_content_key=restricted_course.content_key,
+            content_type=COURSE_RUN,
+        )
+        restricted_course.restricted_run_allowed_for_restricted_course.set(
+            [restricted_run], clear=True,
+        )
+        catalog.catalog_query.content_filter[RESTRICTED_RUNS_ALLOWED_KEY] = {
+            content_one.content_key: [restricted_run.content_key],
+        }
+        catalog.catalog_query.save()
+        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,
+        and that a different catalog that does *not* allow the restricted run
+        is not considered to contain it.
+        """
+        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])
+
+        url = self._get_contains_content_base_url(other_catalog.uuid) + \
+            f'?course_run_ids={restricted_run.content_key}'
+
+        response = self.client.get(url)
+        response_payload = response.json()
+
+        self.assertTrue(response_payload.get('contains_content_items'))
+
+        # self.enterprise_catalog does not contain the restricted run.
+        url = self._get_contains_content_base_url(self.enterprise_catalog) + \
+            f'?course_run_ids={restricted_run.content_key}'
+
+        response = self.client.get(url)
+        response_payload = response.json()
+
+        self.assertFalse(response_payload.get('contains_content_items'))

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

@@ -115,14 +115,20 @@ def _create_restricted_course_and_run(self, catalog):
             content_type=COURSE,
             unrestricted_parent=content_one,
             catalog_query=catalog.catalog_query,
+            _json_metadata=content_one.json_metadata,
         )
         restricted_run = ContentMetadataFactory.create(
             content_key='course-v1:org+key1+restrictedrun',
             content_type=COURSE_RUN,
+            parent_content_key=restricted_course.content_key,
         )
         restricted_course.restricted_run_allowed_for_restricted_course.set(
             [restricted_run], clear=True,
         )
+        catalog.catalog_query.content_filter[RESTRICTED_RUNS_ALLOWED_KEY] = {
+            content_one.content_key: [restricted_run.content_key],
+        }
+        catalog.catalog_query.save()
         return content_one, restricted_course, restricted_run
 
     def test_contains_catalog_key_restricted_runs_allowed(self):
@@ -134,8 +140,9 @@ def test_contains_catalog_key_restricted_runs_allowed(self):
 
         content_one, _, restricted_run = self._create_restricted_course_and_run(catalog)
 
-        self.add_metadata_to_catalog(catalog, [content_one, restricted_run])
-        self.add_metadata_to_catalog(catalog_b, [content_one])
+        self.add_metadata_to_catalog(catalog, [content_one])
+        # add the top-level course to catalog_b, too
+        self.add_metadata_to_catalog(catalog, [content_one])
 
         url = self._get_contains_content_base_url() + \
             f'?course_run_ids={restricted_run.content_key}&get_catalogs_containing_specified_content_ids=true'
@@ -210,14 +217,10 @@ def test_get_content_metadata_restricted_runs(self):
         Tests that we can retrieve restricted content metadata for a customer.
         """
         catalog = EnterpriseCatalogFactory(enterprise_uuid=self.enterprise_uuid)
-        catalog_b = EnterpriseCatalogFactory(enterprise_uuid=self.enterprise_uuid)
 
         content_one, _, restricted_run = self._create_restricted_course_and_run(catalog)
 
-        self.add_metadata_to_catalog(catalog, [content_one, restricted_run])
-
-        # add only the top-level course to catalog B
-        self.add_metadata_to_catalog(catalog_b, [content_one])
+        self.add_metadata_to_catalog(catalog, [content_one])
 
         # Test that we can retrieve the course record
         url = self._get_content_metadata_base_url(self.enterprise_uuid, content_one.content_key)
@@ -230,27 +233,30 @@ def test_get_content_metadata_restricted_runs(self):
         url = self._get_content_metadata_base_url(self.enterprise_uuid, restricted_run.content_key)
 
         response_payload = self.client.get(url).json()
-        self.assertEqual(response_payload['key'], restricted_run.content_key)
-        self.assertEqual(response_payload['content_type'], COURSE_RUN)
+        # this will be a top-level course, with course_runs nested within it
+        self.assertEqual(response_payload['key'], content_one.content_key)
+        self.assertEqual(response_payload['content_type'], COURSE)
 
         # Test that we can retrieve the restricted run by uuid
         url = self._get_content_metadata_base_url(self.enterprise_uuid, restricted_run.content_uuid)
 
         response_payload = self.client.get(url).json()
-        self.assertEqual(response_payload['uuid'], str(restricted_run.content_uuid))
-        self.assertEqual(response_payload['content_type'], COURSE_RUN)
+        # this will be a top-level course, with course_runs nested within it
+        self.assertEqual(response_payload['key'], content_one.content_key)
+        self.assertEqual(response_payload['content_type'], COURSE)
 
     def test_get_content_metadata_restricted_runs_not_found(self):
         """
         Tests that when restricted runs are not explicitly linked to a customer's catalog,
         they cannot be retrieved.
         """
         catalog = EnterpriseCatalogFactory(enterprise_uuid=self.enterprise_uuid)
+        another_customers_catalog = EnterpriseCatalogFactory(enterprise_uuid=str(uuid.uuid4()))
 
-        content_one, _, restricted_run = self._create_restricted_course_and_run(catalog)
+        content_one, _, restricted_run = self._create_restricted_course_and_run(another_customers_catalog)
 
-        # don't add the restricted run to the catalog, just the plain, top-level course
         self.add_metadata_to_catalog(catalog, [content_one])
+        self.add_metadata_to_catalog(another_customers_catalog, [content_one])
 
         # Test that we can retrieve the course record
         url = self._get_content_metadata_base_url(self.enterprise_uuid, content_one.content_key)

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)

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

@@ -3,6 +3,7 @@
 from enterprise_catalog.apps.api.v1.views.enterprise_customer import (
     EnterpriseCustomerViewSet,
 )
+from enterprise_catalog.apps.catalog.models import ContentMetadata
 
 
 logger = logging.getLogger(__name__)
@@ -13,10 +14,24 @@ class EnterpriseCustomerViewSetV2(EnterpriseCustomerViewSet):
     V2 views for content metadata and catalog-content inclusion for retrieving.
     """
     def get_metadata_by_uuid(self, catalog, content_uuid):
-        return catalog.content_metadata_with_restricted.filter(content_uuid=content_uuid).first()
+        """
+        Slightly more complicated - we have to find the content metadata
+        record, regardless of catalog, with this uuid, then use `get_matching_content`
+        on that record's content key.
+        """
+        record = ContentMetadata.objects.filter(content_uuid=content_uuid).first()
+        if not record:
+            return
+        return catalog.get_matching_content(
+            content_keys=[record.content_key],
+            include_restricted=True,
+        ).first()
 
     def get_metadata_by_content_key(self, catalog, content_key):
-        return catalog.get_matching_content(content_keys=[content_key], include_restricted=True).first()
+        return catalog.get_matching_content(
+            content_keys=[content_key],
+            include_restricted=True,
+        ).first()
 
     def filter_content_keys(self, catalog, content_keys):
         return catalog.filter_content_keys(content_keys, include_restricted=True)