feat: v2 customer content-metadata endpoint

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

@@ -0,0 +1,64 @@
+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 BaseEnterpriseCustomerViewSetTests(APITestMixin):
+    """
+    Tests for the EnterpriseCustomerViewSet
+    """
+    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, enterprise_uuid=None):
+        """
+        Helper to construct the base url for the contains_content_items endpoint
+        """
+        return reverse(
+            f'api:{self.VERSION}:enterprise-customer-contains-content-items',
+            kwargs={'enterprise_uuid': enterprise_uuid or self.enterprise_uuid},
+        )
+
+    def _get_filter_content_base_url(self, enterprise_uuid=None):
+        """
+        Helper to construct the base url for the filter_content_items endpoint
+        """
+        return reverse(
+            f'api:{self.VERSION}:enterprise-customer-filter-content-items',
+            kwargs={'enterprise_uuid': enterprise_uuid or self.enterprise_uuid},
+        )
+
+    def _get_generate_diff_base_url(self, enterprise_catalog_uuid=None):
+        """
+        Helper to construct the base url for the catalog `generate_diff` endpoint
+        """
+        return reverse(
+            f'api:{self.VERSION}:generate-catalog-diff',
+            kwargs={'uuid': enterprise_catalog_uuid or self.enterprise_catalog.uuid},
+        )

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

@@ -5,74 +5,21 @@
 import pytest
 import pytz
 from rest_framework import status
-from rest_framework.reverse import reverse
 
-from enterprise_catalog.apps.api.v1.tests.mixins import APITestMixin
+from enterprise_catalog.apps.api.base.tests.enterprise_customer_views import BaseEnterpriseCustomerViewSetTests
 from enterprise_catalog.apps.catalog.constants import (
     RESTRICTED_RUNS_ALLOWED_KEY,
 )
-from enterprise_catalog.apps.catalog.models import (
-    CatalogQuery,
-    ContentMetadata,
-    EnterpriseCatalog,
-)
 from enterprise_catalog.apps.catalog.tests.factories import (
     ContentMetadataFactory,
     EnterpriseCatalogFactory,
 )
 
 
-class EnterpriseCustomerViewSetTests(APITestMixin):
+class EnterpriseCustomerViewSetTests(BaseEnterpriseCustomerViewSetTests):
     """
     Tests for the EnterpriseCustomerViewSet
     """
-
-    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, enterprise_uuid=None):
-        """
-        Helper to construct the base url for the contains_content_items endpoint
-        """
-        return reverse(
-            'api:v1:enterprise-customer-contains-content-items',
-            kwargs={'enterprise_uuid': enterprise_uuid or self.enterprise_uuid},
-        )
-
-    def _get_filter_content_base_url(self, enterprise_uuid=None):
-        """
-        Helper to construct the base url for the filter_content_items endpoint
-        """
-        return reverse(
-            'api:v1:enterprise-customer-filter-content-items',
-            kwargs={'enterprise_uuid': enterprise_uuid or self.enterprise_uuid},
-        )
-
-    def _get_generate_diff_base_url(self, enterprise_catalog_uuid=None):
-        """
-        Helper to construct the base url for the catalog `generate_diff` endpoint
-        """
-        return reverse(
-            'api:v1:generate-catalog-diff',
-            kwargs={'uuid': enterprise_catalog_uuid or self.enterprise_catalog.uuid},
-        )
-
     def test_generate_diff_unauthorized_non_catalog_learner(self):
         """
         Verify the generate_diff endpoint rejects users that are not catalog learners
@@ -357,152 +304,6 @@ def test_contains_catalog_list_with_catalog_list_param(self):
         catalog_list = response.json()['catalog_list']
         assert set(catalog_list) == {str(second_catalog.uuid)}
 
-    @pytest.mark.skip(reason="We need a version of this test for the v2 API.")
-    def test_contains_catalog_list_with_content_ids_param(self):
-        """
-        Verify the contains_content_items endpoint returns a list of catalogs the course is in if the correct
-        parameter is passed
-        """
-        content_metadata = ContentMetadataFactory()
-        self.add_metadata_to_catalog(self.enterprise_catalog, [content_metadata])
-
-        # Create a two catalogs that have the content we're looking for
-        content_key = 'fake-key+101x'
-        second_catalog = EnterpriseCatalogFactory(enterprise_uuid=self.enterprise_uuid)
-
-        relevant_content = ContentMetadataFactory(content_key=content_key)
-        self.add_metadata_to_catalog(second_catalog, [relevant_content])
-        url = self._get_contains_content_base_url() + '?course_run_ids=' + content_key + \
-            '&get_catalogs_containing_specified_content_ids=True'
-        self.assert_correct_contains_response(url, True)
-
-        response = self.client.get(url)
-        response_payload = response.json()
-        catalog_list = response_payload['catalog_list']
-        assert set(catalog_list) == {str(second_catalog.uuid)}
-        self.assertIsNone(response_payload['restricted_runs_allowed'])
-
-    @pytest.mark.skip(reason="We need a version of this test for the v2 API.")
-    def test_contains_catalog_key_restricted_runs_allowed(self):
-        """
-        Tests that, when a course key is requested, we also get a response
-        describing any child restricted runs that are allowed under that course key
-        for the customer.
-        """
-        catalog = EnterpriseCatalogFactory(enterprise_uuid=self.enterprise_uuid)
-        catalog.catalog_query.content_filter[RESTRICTED_RUNS_ALLOWED_KEY] = {
-            'org+key1': ['course-v1:org+key1+restrictedrun']
-        }
-        catalog.catalog_query.save()
-        content_one = ContentMetadataFactory(content_key='org+key1')
-        content_two = ContentMetadataFactory(content_key='org+key2')
-        self.add_metadata_to_catalog(catalog, [content_one, content_two])
-
-        other_catalog = EnterpriseCatalogFactory(enterprise_uuid=self.enterprise_uuid)
-        other_catalog.catalog_query.content_filter[RESTRICTED_RUNS_ALLOWED_KEY] = {
-            'course:org+key3': ['course-v1:org+key3+restrictedrun'],
-            'course:org+key4': ['course-v1:org+key4+restrictedrun']
-        }
-        other_catalog.catalog_query.save()
-        content_three = ContentMetadataFactory(content_key='org+key3')
-        # created a content record that has a restricted run,
-        # but which we won't make a request for.
-        content_four = ContentMetadataFactory(content_key='org+key4')
-        content_five = ContentMetadataFactory(content_key='org+key5')
-        self.add_metadata_to_catalog(other_catalog, [content_three, content_four, content_five])
-
-        # make sure to also request a course key that has no restricted runs,
-        # and then assert that it is *not* included in the response payload.
-        url = self._get_contains_content_base_url() + \
-            '?course_run_ids=org+key1&course_run_ids=org+key2&course_run_ids=org+key3&get_catalog_list=true'
-
-        response = self.client.get(url)
-        response_payload = response.json()
-
-        self.assertTrue(response_payload.get('contains_content_items'))
-        self.assertEqual(
-            set(response_payload['catalog_list']),
-            set([str(catalog.uuid), str(other_catalog.uuid)])
-        )
-        self.assertEqual(
-            response_payload['restricted_runs_allowed'],
-            {
-                'org+key1': {
-                    'course-v1:org+key1+restrictedrun': {
-                        'catalog_uuids': [str(catalog.uuid)]
-                    },
-                },
-                'org+key3': {
-                    'course-v1:org+key3+restrictedrun': {
-                        'catalog_uuids': [str(other_catalog.uuid)]
-                    },
-                },
-            }
-        )
-
-    @pytest.mark.skip(reason="We need a version of this test for the v2 API.")
-    def test_restricted_course_disallowed_if_course_not_in_catalog(self):
-        """
-        Tests that a requested course with restricted runs is "disallowed"
-        if the course is not part of a customer's catalog.
-        """
-        catalog = EnterpriseCatalogFactory(enterprise_uuid=self.enterprise_uuid)
-        catalog.catalog_query.content_filter[RESTRICTED_RUNS_ALLOWED_KEY] = {
-            'org+key1': ['course-v1:org+key1+restrictedrun']
-        }
-        catalog.catalog_query.save()
-        ContentMetadataFactory(content_key='org+key1')
-        # don't add this content to the catalog
-
-        url = self._get_contains_content_base_url() + '?course_run_ids=org+key1'
-
-        response = self.client.get(url)
-        response_payload = response.json()
-
-        self.assertFalse(response_payload.get('contains_content_items'))
-        self.assertIsNone(response_payload['restricted_runs_allowed'])
-
-    @pytest.mark.skip(reason="We need a version of this test for the v2 API.")
-    def test_restricted_course_run_allowed_even_if_course_not_in_catalog(self):
-        """
-        Tests that a requested restricted course run is "allowed"
-        even if the course is not part of a customer's catalog. This is necessary
-        because typically restricted runs will not have corresponding
-        `ContentMetadata` records present in the DB, so a lookup via only
-        `EnterpriseCatalog.contains_content_keys` will fail. We rely
-        on the restricted run mapping to ascertain the *implicit* inclusion
-        of a restricted course run in a catalog.
-        """
-        catalog = EnterpriseCatalogFactory(enterprise_uuid=self.enterprise_uuid)
-        catalog.catalog_query.content_filter[RESTRICTED_RUNS_ALLOWED_KEY] = {
-            'org+key1': ['course-v1:org+key1+restrictedrun']
-        }
-        catalog.catalog_query.save()
-        ContentMetadataFactory(content_key='org+key1')
-        # don't add this content to the catalog
-
-        url = self._get_contains_content_base_url() + \
-            '?course_run_ids=course-v1:org+key1+restrictedrun&get_catalog_list=true'
-
-        response = self.client.get(url)
-        response_payload = response.json()
-
-        self.assertTrue(response_payload.get('contains_content_items'))
-        self.assertEqual(
-            response_payload['catalog_list'],
-            [str(catalog.uuid)],
-        )
-        self.assertEqual(
-            response_payload['restricted_runs_allowed'],
-            {
-                'org+key1': {
-                    'course-v1:org+key1+restrictedrun': {
-                        'catalog_uuids': [str(catalog.uuid)]
-                    },
-                },
-            }
-        )
-
     def test_contains_catalog_list_parent_key(self):
         """
         Verify the contains_content_items endpoint returns a list of catalogs the course is in

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

@@ -64,6 +64,18 @@ def get_permission_object(self):
         """
         return self.kwargs.get('enterprise_uuid')
 
+    def filter_content_keys(self, catalog, content_keys):
+        return catalog.filter_content_keys(content_keys)
+
+    def contains_content_keys(self, catalog, content_keys):
+        return catalog.contains_content_keys(content_keys)
+
+    def get_metadata_by_uuid(self, catalog, content_uuid):
+        return catalog.content_metadata.filter(content_uuid=content_uuid)
+
+    def get_metadata_by_content_key(self, catalog, content_key):
+        return catalog.get_matching_content(content_keys=[content_key])
+
     @method_decorator(require_at_least_one_query_parameter('course_run_ids', 'program_uuids'))
     @action(detail=True)
     def contains_content_items(self, request, enterprise_uuid, course_run_ids, program_uuids, **kwargs):
@@ -105,9 +117,9 @@ def contains_content_items(self, request, enterprise_uuid, course_run_ids, progr
 
         any_catalog_contains_content_items = False
         catalogs_that_contain_course = []
+        content_keys = requested_course_or_run_keys + program_uuids
         for catalog in customer_catalogs:
-            contains_content_items = catalog.contains_content_keys(requested_course_or_run_keys + program_uuids)
-            if contains_content_items:
+            if contains_content_items := self.contains_content_keys(catalog, content_keys):
                 any_catalog_contains_content_items = True
                 if not (get_catalogs_containing_specified_content_ids or get_catalog_list):
                     # Break as soon as we find a catalog that contains the specified content
@@ -136,8 +148,7 @@ def filter_content_items(self, request, enterprise_uuid, **kwargs):
 
         filtered_content_keys = set()
         for catalog in customer_catalogs:
-            items_included = catalog.filter_content_keys(content_keys)
-            if items_included:
+            if items_included := self.filter_content_keys(catalog, content_keys):
                 filtered_content_keys = filtered_content_keys.union(items_included)
 
         response_data = {
@@ -164,17 +175,15 @@ def get_metadata_item_serializer(self):
             # identifier is a valid UUID.
             content_uuid = uuid.UUID(content_identifier)
             for catalog in enterprise_catalogs:
-                content_with_uuid = catalog.content_metadata.filter(content_uuid=content_uuid)
-                if content_with_uuid:
+                if content_with_uuid := self.get_metadata_by_uuid(catalog, content_uuid):
                     return ContentMetadataSerializer(
                         content_with_uuid.first(),
                         context={'enterprise_catalog': catalog, **serializer_context},
                     )
         except ValueError:
             # Otherwise, search for matching metadata as a content key
             for catalog in enterprise_catalogs:
-                content_with_key = catalog.get_matching_content(content_keys=[content_identifier])
-                if content_with_key:
+                if content_with_key := self.get_metadata_by_content_key(catalog, content_identifier):
                     return ContentMetadataSerializer(
                         content_with_key.first(),
                         context={'enterprise_catalog': catalog, **serializer_context},