Fix swagger API docs generation

vulnerabilities/api.py

@@ -11,8 +11,11 @@
 
 from django.db.models import Prefetch
 from django_filters import rest_framework as filters
+from drf_spectacular.utils import extend_schema
+from drf_spectacular.utils import inline_serializer
 from packageurl import PackageURL
 from rest_framework import serializers
+from rest_framework import status
 from rest_framework import viewsets
 from rest_framework.decorators import action
 from rest_framework.response import Response
@@ -272,6 +275,26 @@ def filter_purl(self, queryset, name, value):
         return self.queryset.filter(**lookups)
 
 
+class PackageurlListSerializer(serializers.Serializer):
+    purls = serializers.ListField(
+        child=serializers.CharField(),
+        allow_empty=False,
+        help_text="List of PackageURL strings in canonical form.",
+    )
+
+
+class PackageBulkSearchRequestSerializer(PackageurlListSerializer):
+    purl_only = serializers.BooleanField(required=False, default=False)
+    plain_purl = serializers.BooleanField(required=False, default=False)
+
+
+class LookupRequestSerializer(serializers.Serializer):
+    purl = serializers.CharField(
+        required=True,
+        help_text="PackageURL strings in canonical form.",
+    )
+
+
 class PackageViewSet(viewsets.ReadOnlyModelViewSet):
     """
     Lookup for vulnerable packages by Package URL.
@@ -283,21 +306,34 @@ class PackageViewSet(viewsets.ReadOnlyModelViewSet):
     filterset_class = PackageFilterSet
     throttle_classes = [StaffUserRateThrottle, AnonRateThrottle]
 
-    # TODO: Fix the swagger documentation for this endpoint
-    @action(detail=False, methods=["post"])
+    @extend_schema(
+        request=PackageBulkSearchRequestSerializer,
+        responses={200: PackageSerializer(many=True)},
+    )
+    @action(
+        detail=False,
+        methods=["post"],
+        serializer_class=PackageBulkSearchRequestSerializer,
+        filter_backends=[],
+        pagination_class=None,
+    )
     def bulk_search(self, request):
         """
         Lookup for vulnerable packages using many Package URLs at once.
         """
-
-        purls = request.data.get("purls", []) or []
-        purl_only = request.data.get("purl_only", False)
-        plain_purl = request.data.get("plain_purl", False)
-        if not purls or not isinstance(purls, list):
+        serializer = self.serializer_class(data=request.data)
+        if not serializer.is_valid():
             return Response(
-                status=400,
-                data={"Error": "A non-empty 'purls' list of PURLs is required."},
+                status=status.HTTP_400_BAD_REQUEST,
+                data={
+                    "error": serializer.errors,
+                    "message": "A non-empty 'purls' list of PURLs is required.",
+                },
             )
+        validated_data = serializer.validated_data
+        purls = validated_data.get("purls")
+        purl_only = validated_data.get("purl_only", False)
+        plain_purl = validated_data.get("plain_purl", False)
 
         if plain_purl:
             purl_objects = [PackageURL.from_string(purl) for purl in purls]
@@ -347,34 +383,66 @@ def all(self, request):
         vulnerable_purls = [str(package.package_url) for package in vulnerable_packages]
         return Response(vulnerable_purls)
 
-    @action(detail=False, methods=["post"])
+    @extend_schema(
+        request=LookupRequestSerializer,
+        responses={200: PackageSerializer(many=True)},
+    )
+    @action(
+        detail=False,
+        methods=["post"],
+        serializer_class=LookupRequestSerializer,
+        filter_backends=[],
+        pagination_class=None,
+    )
     def lookup(self, request):
         """
         Return the response for exact PackageURL requested for.
         """
-        purl = request.data.get("purl")
-        if not purl:
+        serializer = self.serializer_class(data=request.data)
+        if not serializer.is_valid():
             return Response(
-                status=400,
-                data={"Error": "A 'purl' is required."},
+                status=status.HTTP_400_BAD_REQUEST,
+                data={
+                    "error": serializer.errors,
+                    "message": "A 'purl' is required.",
+                },
             )
+        validated_data = serializer.validated_data
+        purl = validated_data.get("purl")
+
         return Response(
             PackageSerializer(
                 Package.objects.for_purls([purl]), many=True, context={"request": request}
             ).data
         )
 
-    @action(detail=False, methods=["post"])
+    @extend_schema(
+        request=PackageurlListSerializer,
+        responses={200: PackageSerializer(many=True)},
+    )
+    @action(
+        detail=False,
+        methods=["post"],
+        serializer_class=PackageurlListSerializer,
+        filter_backends=[],
+        pagination_class=None,
+    )
     def bulk_lookup(self, request):
         """
         Return the response for exact PackageURLs requested for.
         """
-        purls = request.data.get("purls") or []
-        if not purls:
+        serializer = self.serializer_class(data=request.data)
+        if not serializer.is_valid():
             return Response(
-                status=400,
-                data={"Error": "A non-empty 'purls' list of PURLs is required."},
+                status=status.HTTP_400_BAD_REQUEST,
+                data={
+                    "error": serializer.errors,
+                    "message": "A non-empty 'purls' list of PURLs is required.",
+                },
             )
+        validated_data = serializer.validated_data
+        purls = validated_data.get("purls")
+
         return Response(
             PackageSerializer(
                 Package.objects.for_purls(purls),

vulnerabilities/tests/test_api.py

@@ -647,6 +647,55 @@ def test_bulk_api_with_purl_only_option(self):
         assert len(response) == 1
         assert response[0] == "pkg:nginx/[email protected]"
 
+    def test_bulk_api_without_purls_list(self):
+        request_body = {
+            "purls": None,
+        }
+        response = self.csrf_client.post(
+            "/api/packages/bulk_search",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+
+        expected = {
+            "error": {"purls": ["This field may not be null."]},
+            "message": "A non-empty 'purls' list of PURLs is required.",
+        }
+
+        self.assertEqual(response, expected)
+
+    def test_bulk_api_without_purls_empty_list(self):
+        request_body = {
+            "purls": [],
+        }
+        response = self.csrf_client.post(
+            "/api/packages/bulk_search",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+
+        expected = {
+            "error": {"purls": ["This list may not be empty."]},
+            "message": "A non-empty 'purls' list of PURLs is required.",
+        }
+
+        self.assertEqual(response, expected)
+
+    def test_bulk_api_with_empty_request_body(self):
+        request_body = {}
+        response = self.csrf_client.post(
+            "/api/packages/bulk_search",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+
+        expected = {
+            "error": {"purls": ["This field is required."]},
+            "message": "A non-empty 'purls' list of PURLs is required.",
+        }
+
+        self.assertEqual(response, expected)
+
 
 class BulkSearchAPICPE(TestCase):
     def setUp(self):
@@ -768,7 +817,13 @@ def test_lookup_endpoint_failure(self):
             data=json.dumps(request_body),
             content_type="application/json",
         ).json()
-        assert response == {"Error": "A 'purl' is required."}
+
+        expected = {
+            "error": {"purl": ["This field may not be null."]},
+            "message": "A 'purl' is required.",
+        }
+
+        self.assertEqual(response, expected)
 
     def test_lookup_endpoint(self):
         request_body = {"purl": "pkg:pypi/microweber/[email protected]"}
@@ -844,3 +899,18 @@ def test_bulk_lookup_endpoint(self):
             content_type="application/json",
         ).json()
         assert len(response) == 1
+
+    def test_bulk_lookup_endpoint_failure(self):
+        request_body = {"purls": None}
+        response = self.csrf_client.post(
+            "/api/packages/bulk_lookup",
+            data=json.dumps(request_body),
+            content_type="application/json",
+        ).json()
+
+        expected = {
+            "error": {"purls": ["This field may not be null."]},
+            "message": "A non-empty 'purls' list of PURLs is required.",
+        }
+
+        self.assertEqual(response, expected)