feat: new catalog query read only viewset

enterprise_catalog/apps/api/v1/serializers.py

@@ -1,5 +1,5 @@
 import logging
-from re import search
+from re import findall, search
 
 from django.db import IntegrityError, models
 from rest_framework import serializers, status
@@ -85,6 +85,25 @@ def find_and_modify_catalog_query(
         return content_filter_from_hash
 
 
+class CatalogQuerySerializer(serializers.ModelSerializer):
+    """
+    Serializer for the `CatalogQuery` model
+    """
+    content_filter = serializers.JSONField(
+        read_only=True,
+        help_text="Elastic search content filter used to determine content ownership of the query."
+    )
+
+    class Meta:
+        model = CatalogQuery
+        fields = [
+            'uuid',
+            'content_filter',
+            'content_filter_hash',
+            'title',
+        ]
+
+
 class EnterpriseCatalogSerializer(serializers.ModelSerializer):
     """
     Serializer for the `EnterpriseCatalog` model
@@ -191,6 +210,18 @@ def update(self, instance, validated_data):
         """
 
 
+class CatalogQueryGetByHashRequestSerializer(ImmutableStateSerializer):
+    """
+    Request serializer to validate request data provided to the CatalogQueryViewSet's ``get_query_by_hash`` endpoint
+    """
+    hash = serializers.CharField(required=True, max_length=32)
+
+    def validate_hash(self, value):
+        if not findall(r"([a-fA-F\d]{32})", value):
+            raise serializers.ValidationError("Invalid filter hash.")
+        return value
+
+
 class ContentMetadataSerializer(ImmutableStateSerializer):
     """
     Serializer for rendering Content Metadata objects

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

@@ -44,6 +44,7 @@
 )
 from enterprise_catalog.apps.catalog.utils import (
     enterprise_proxy_login_url,
+    get_content_filter_hash,
     get_content_key,
     get_parent_content_key,
     localized_utcnow,
@@ -2499,3 +2500,140 @@ def test_filter_list_by_uuid(self):
         assert len(response_json.get('results')) == 1
         assert response_json.get('results')[0].get("key") == self.content_metadata_object.content_key
         assert response_json.get('results')[0].get("course_runs")[0].get('start') == '2024-02-12T11:00:00Z'
+
+
+@ddt.ddt
+class CatalogQueryViewTests(APITestMixin):
+    """
+    Tests for the readonly ContentMetadata viewset.
+    """
+    def setUp(self):
+        super().setUp()
+        self.set_up_catalog_learner()
+        self.catalog_query_object = CatalogQueryFactory()
+        self.catalog_object = EnterpriseCatalogFactory(catalog_query=self.catalog_query_object)
+        self.assign_catalog_admin_feature_role(enterprise_uuids=[self.catalog_object.enterprise_uuid])
+        # Factory doesn't set up a hash, so do it manually
+        self.catalog_query_object.content_filter_hash = get_content_filter_hash(
+            self.catalog_query_object.content_filter
+        )
+        self.catalog_query_object.save()
+
+    def test_get_query_by_hash(self):
+        """
+        Test that the list content_identifiers query param accepts uuids
+        """
+        query_param_string = f"?hash={self.catalog_query_object.content_filter_hash}"
+        url = reverse('api:v1:get-query-by-hash') + query_param_string
+        response = self.client.get(url)
+        response_json = response.json()
+        # The user is a part of the enterprise that has a catalog that contains this query
+        # so they can view the data
+        assert response_json.get('uuid') == str(self.catalog_query_object.uuid)
+        assert str(response_json.get('content_filter')) == str(self.catalog_query_object.content_filter)
+
+        # Permissions verification while looking up by hash
+        different_catalog = EnterpriseCatalogFactory()
+        # Factory doesn't set up a hash, so do it manually
+        different_catalog.catalog_query.content_filter_hash = get_content_filter_hash(
+            different_catalog.catalog_query.content_filter
+        )
+        different_catalog.save()
+        query_param_string = f"?hash={different_catalog.catalog_query.content_filter_hash}"
+        url = reverse('api:v1:get-query-by-hash') + query_param_string
+        response = self.client.get(url)
+        response_json = response.json()
+        assert response_json == {'detail': 'Catalog query not found.'}
+
+        # If the user is staff, they get access to everything
+        self.set_up_staff()
+        response = self.client.get(url)
+        response_json = response.json()
+        assert response_json.get('uuid') == str(different_catalog.catalog_query.uuid)
+
+        self.set_up_invalid_jwt_role()
+        self.remove_role_assignments()
+        response = self.client.get(url)
+        assert response.status_code == 404
+
+    def test_get_query_by_hash_not_found(self):
+        """
+        Test that the get query by hash endpoint returns expected not found
+        """
+        query_param_string = f"?hash={self.catalog_query_object.content_filter_hash[:-6]}aaaaaa"
+        url = reverse('api:v1:get-query-by-hash') + query_param_string
+        response = self.client.get(url)
+        response_json = response.json()
+        assert response_json == {'detail': 'Catalog query not found.'}
+
+    def test_get_query_by_illegal_hash(self):
+        """
+        Test that the get query by hash endpoint validates filter hashes
+        """
+        query_param_string = "?hash=foobar"
+        url = reverse('api:v1:get-query-by-hash') + query_param_string
+        response = self.client.get(url)
+        response_json = response.json()
+        assert response_json == {'hash': ['Invalid filter hash.']}
+
+    def test_get_query_by_hash_requires_hash(self):
+        """
+        Test that the get query by hash requires a hash query param
+        """
+        url = reverse('api:v1:get-query-by-hash')
+        response = self.client.get(url)
+        response_json = response.json()
+        assert response_json == ['You must provide at least one of the following query parameters: hash.']
+
+    def test_catalog_query_retrieve(self):
+        """
+        Test that the Catalog Query viewset supports retrieving individual queries
+        """
+        self.assign_catalog_admin_jwt_role(
+            self.enterprise_uuid,
+            self.catalog_query_object.enterprise_catalogs.first().enterprise_uuid,
+        )
+        url = reverse('api:v1:catalog-queries-detail', kwargs={'pk': self.catalog_query_object.pk})
+        response = self.client.get(url)
+        response_json = response.json()
+        assert response_json.get('uuid') == str(self.catalog_query_object.uuid)
+
+        different_customer_catalog = EnterpriseCatalogFactory()
+        # We don't have a jwt token that includes an admin role for the new enterprise so it is
+        # essentially hidden to the requester
+        url = reverse('api:v1:catalog-queries-detail', kwargs={'pk': different_customer_catalog.catalog_query.pk})
+        response = self.client.get(url)
+        assert response.status_code == 404
+
+        # If the user is staff, they get access to everything
+        self.set_up_staff()
+        response = self.client.get(url)
+        response_json = response.json()
+        assert response_json.get('uuid') == str(different_customer_catalog.catalog_query.uuid)
+
+    def test_catalog_query_list(self):
+        """
+        Test that the Catalog Query viewset supports listing queries
+        """
+        # Create another catalog associated with another enterprise and therefore hidden to the requesting user
+        EnterpriseCatalogFactory()
+        self.assign_catalog_admin_jwt_role(
+            self.enterprise_uuid,
+            self.catalog_query_object.enterprise_catalogs.first().enterprise_uuid,
+        )
+        url = reverse('api:v1:catalog-queries-list')
+        response = self.client.get(url)
+        response_json = response.json()
+        assert response_json.get('count') == 1
+        assert response_json.get('results')[0].get('uuid') == str(self.catalog_query_object.uuid)
+
+        # If the user is staff, they get access to everything
+        self.set_up_staff()
+        response = self.client.get(url)
+        response_json = response.json()
+        assert response_json.get('count') == 2
+
+        self.set_up_invalid_jwt_role()
+        self.remove_role_assignments()
+        response = self.client.get(url)
+        assert response.data == {'count': 0, 'next': None, 'previous': None, 'results': []}

enterprise_catalog/apps/api/v1/urls.py

@@ -11,6 +11,9 @@
 from enterprise_catalog.apps.api.v1.views.catalog_csv_data import (
     CatalogCsvDataView,
 )
+from enterprise_catalog.apps.api.v1.views.catalog_query import (
+    CatalogQueryViewSet,
+)
 from enterprise_catalog.apps.api.v1.views.catalog_workbook import (
     CatalogWorkbookView,
 )
@@ -61,6 +64,7 @@
 router.register(r'highlight-sets-admin', HighlightSetViewSet, basename='highlight-sets-admin')
 router.register(r'academies', AcademiesReadOnlyViewSet, basename='academies')
 router.register(r'content-metadata', ContentMetadataView, basename='content-metadata')
+router.register(r'catalog-queries', CatalogQueryViewSet, basename='catalog-queries')
 
 urlpatterns = [
     path('enterprise-catalogs/catalog_csv_data', CatalogCsvDataView.as_view(),
@@ -106,6 +110,11 @@
         EnterpriseCustomerViewSet.as_view({'get': 'content_metadata'}),
         name='customer-content-metadata-retrieve'
     ),
+    path(
+        'catalog-queries/get_query_by_hash',
+        CatalogQueryViewSet.as_view({'get': 'get_query_by_hash'}),
+        name='get-query-by-hash'
+    ),
 ]
 
 urlpatterns += router.urls

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

@@ -0,0 +1,64 @@
+from django.utils.decorators import method_decorator
+from django.utils.functional import cached_property
+from rest_framework import viewsets
+from rest_framework.decorators import action
+from rest_framework.exceptions import NotFound
+from rest_framework.renderers import JSONRenderer
+from rest_framework.response import Response
+
+from enterprise_catalog.apps.api.v1.decorators import (
+    require_at_least_one_query_parameter,
+)
+from enterprise_catalog.apps.api.v1.serializers import (
+    CatalogQueryGetByHashRequestSerializer,
+    CatalogQuerySerializer,
+)
+from enterprise_catalog.apps.catalog.models import CatalogQuery
+from enterprise_catalog.apps.catalog.rules import (
+    enterprises_with_admin_access,
+    has_access_to_all_enterprises,
+)
+
+
+class CatalogQueryViewSet(viewsets.ReadOnlyModelViewSet):
+    """Read-only viewset for Catalog Query records"""
+    renderer_classes = [JSONRenderer]
+    serializer_class = CatalogQuerySerializer
+
+    @cached_property
+    def admin_accessible_enterprises(self):
+        """
+        Cached set of enterprise identifiers the requesting user has admin access to.
+        """
+        return enterprises_with_admin_access(self.request)
+
+    def get_queryset(self):
+        """
+        Restrict the queryset to catalog queries the requesting user has access to. Iff the user is staff they have
+        access to all queries.
+        """
+        all_queries = CatalogQuery.objects.all()
+        if not self.admin_accessible_enterprises:
+            return CatalogQuery.objects.none()
+        if has_access_to_all_enterprises(self.admin_accessible_enterprises) or self.request.user.is_staff:
+            return all_queries
+        return all_queries.filter(
+            enterprise_catalogs__enterprise_uuid__in=self.admin_accessible_enterprises
+        )
+
+    @method_decorator(require_at_least_one_query_parameter('hash'))
+    @action(detail=True, methods=['get'])
+    def get_query_by_hash(self, request, **kwargs):
+        """
+        Fetch a Catalog Query by its hash. The hash values are a product of Python's ``hashlib``'s md5 algorithm
+        in hexdigest representation.
+        """
+        request_serializer = CatalogQueryGetByHashRequestSerializer(data=request.query_params)
+        request_serializer.is_valid(raise_exception=True)
+        content_filter_hash = request_serializer.validated_data.get('hash')
+        try:
+            query = self.get_queryset().get(content_filter_hash=content_filter_hash)
+        except CatalogQuery.DoesNotExist as exc:
+            raise NotFound('Catalog query not found.') from exc
+        serialized_data = self.serializer_class(query)
+        return Response(serialized_data.data)

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

@@ -45,7 +45,7 @@ def get_serializer_class(self):
 
     def get_permission_object(self):
         """
-        Retrieves the apporpriate object to use during edx-rbac's permission checks.
+        Retrieves the appropriate object to use during edx-rbac's permission checks.
 
         This object is passed to the rule predicate(s).
         """

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

@@ -58,7 +58,7 @@ def check_permissions(self, request):
 
     def get_permission_object(self):
         """
-        Retrieves the apporpriate object to use during edx-rbac's permission checks.
+        Retrieves the appropriate object to use during edx-rbac's permission checks.
 
         This object is passed to the rule predicate(s).
         """