Add "Pull data from a ScanCode.io" Product action in the REST API #59

Signed-off-by: tdruez <tdruez@nexb.com>

CHANGELOG.rst

@@ -24,8 +24,8 @@ Release notes
 - Add dark theme support in UI.
   https://github.com/nexB/dejacode/issues/25
 
-- Add "Load Packages from SBOMs" and "Import scan results" feature as Product action
-  in the REST API.
+- Add "Load Packages from SBOMs", "Import scan results", and
+  "Pull ScanCode.io project data" feature as Product action in the REST API.
   https://github.com/nexB/dejacode/issues/59
 
 - Refactor the "Import manifest" feature as "Load SBOMs".

product_portfolio/api.py

@@ -36,6 +36,7 @@
 from product_portfolio.filters import ComponentCompletenessAPIFilter
 from product_portfolio.forms import ImportFromScanForm
 from product_portfolio.forms import LoadSBOMsForm
+from product_portfolio.forms import PullProjectDataForm
 from product_portfolio.models import CodebaseResource
 from product_portfolio.models import Product
 from product_portfolio.models import ProductComponent
@@ -233,6 +234,20 @@ class ImportFromScanSerializer(serializers.Serializer):
     )
 
 
+class PullProjectDataSerializer(serializers.Serializer):
+    """Serializer equivalent of PullProjectDataForm, used for API documentation."""
+
+    project_name_or_uuid = serializers.CharField(
+        required=True,
+        help_text=PullProjectDataForm.base_fields["project_name_or_uuid"].label,
+    )
+    update_existing_packages = serializers.BooleanField(
+        required=False,
+        default=False,
+        help_text=PullProjectDataForm.base_fields["update_existing_packages"].help_text,
+    )
+
+
 class ProductViewSet(CreateRetrieveUpdateListViewSet):
     queryset = Product.objects.none()
     serializer_class = ProductSerializer
@@ -329,6 +344,25 @@ def import_from_scan(self, request, *args, **kwargs):
             msg += ", ".join([f"{value} {key}" for key, value in created_counts.items()])
         return Response({"status": msg})
 
+    @action(detail=True, methods=["post"], serializer_class=PullProjectDataSerializer)
+    def pull_scancodeio_project_data(self, request, *args, **kwargs):
+        """
+        Pull data from a ScanCode.io Project to import all its Discovered Packages.
+        Imported Packages will be assigned to this Product.
+        """
+        product = self.get_object()
+
+        form = PullProjectDataForm(data=request.POST)
+        if not form.is_valid():
+            return Response(form.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        try:
+            form.submit(product=product, user=request.user)
+        except ValidationError as error:
+            return Response(error.messages, status=status.HTTP_400_BAD_REQUEST)
+
+        return Response({"status": "Packages import from ScanCode.io in progress..."})
+
 
 class BaseProductRelationSerializer(ValidateLicenseExpressionMixin, DataspacedSerializer):
     product = NameVersionHyperlinkedRelatedField(

product_portfolio/forms.py

@@ -30,6 +30,7 @@
 from component_catalog.license_expression_dje import LicenseExpressionFormMixin
 from component_catalog.models import Component
 from component_catalog.programming_languages import PROGRAMMING_LANGUAGES
+from dejacode_toolkit.scancodeio import ScanCodeIO
 from dje import tasks
 from dje.fields import SmartFileField
 from dje.forms import ColorCodeFormMixin
@@ -890,3 +891,35 @@ def helper(self):
         helper.form_id = "pull-project-data-form"
         helper.attrs = {"autocomplete": "off"}
         return helper
+
+    def get_project_data(self, project_name_or_uuid, user):
+        scancodeio = ScanCodeIO(user)
+        for field_name in ["name", "uuid"]:
+            project_data = scancodeio.find_project(**{field_name: project_name_or_uuid})
+            if project_data:
+                return project_data
+
+    def submit(self, product, user):
+        project_name_or_uuid = self.cleaned_data.get("project_name_or_uuid")
+        project_data = self.get_project_data(project_name_or_uuid, user)
+
+        if not project_data:
+            msg = f'Project "{project_name_or_uuid}" not found on ScanCode.io.'
+            raise ValidationError(msg)
+
+        scancode_project = ScanCodeProject.objects.create(
+            product=product,
+            dataspace=product.dataspace,
+            type=ScanCodeProject.ProjectType.PULL_FROM_SCANCODEIO,
+            project_uuid=project_data.get("uuid"),
+            update_existing_packages=self.cleaned_data.get("update_existing_packages"),
+            scan_all_packages=False,
+            status=ScanCodeProject.Status.SUBMITTED,
+            created_by=user,
+        )
+
+        transaction.on_commit(
+            lambda: tasks.pull_project_data_from_scancodeio.delay(
+                scancodeproject_uuid=scancode_project.uuid,
+            )
+        )

product_portfolio/tests/test_api.py

@@ -7,7 +7,9 @@
 #
 
 import json
+import uuid
 from pathlib import Path
+from unittest import mock
 
 from django.core import mail
 from django.core.files.base import ContentFile
@@ -352,6 +354,8 @@ def test_api_product_endpoint_load_sboms_action(self):
         url = reverse("api_v2:product-load-sboms", args=[self.product1.uuid])
 
         self.client.login(username=self.base_user.username, password="secret")
+        response = self.client.get(url)
+        self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
         response = self.client.post(url, data={})
         self.assertEqual(status.HTTP_403_FORBIDDEN, response.status_code)
 
@@ -379,6 +383,8 @@ def test_api_product_endpoint_import_from_scan_action(self):
         url = reverse("api_v2:product-import-from-scan", args=[self.product1.uuid])
 
         self.client.login(username=self.base_user.username, password="secret")
+        response = self.client.get(url)
+        self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
         response = self.client.post(url, data={})
         self.assertEqual(status.HTTP_403_FORBIDDEN, response.status_code)
 
@@ -416,6 +422,41 @@ def test_api_product_endpoint_import_from_scan_action(self):
         self.assertEqual(1, self.product1.packages.count())
         self.assertEqual(3, self.product1.codebaseresources.count())
 
+    @mock.patch("product_portfolio.forms.PullProjectDataForm.get_project_data")
+    def test_api_product_endpoint_pull_scancodeio_project_data_action(self, mock_get_project_data):
+        url = reverse("api_v2:product-pull-scancodeio-project-data", args=[self.product1.uuid])
+
+        self.client.login(username=self.base_user.username, password="secret")
+        response = self.client.get(url)
+        self.assertEqual(status.HTTP_405_METHOD_NOT_ALLOWED, response.status_code)
+        response = self.client.post(url, data={})
+        self.assertEqual(status.HTTP_403_FORBIDDEN, response.status_code)
+
+        # Required permissions
+        add_perm(self.base_user, "add_product")
+        assign_perm("view_product", self.base_user, self.product1)
+
+        response = self.client.post(url, data={})
+        self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
+        expected = {"project_name_or_uuid": ["This field is required."]}
+        self.assertEqual(expected, response.data)
+
+        mock_get_project_data.return_value = None
+        data = {
+            "project_name_or_uuid": "project_name",
+            "update_existing_packages": False,
+        }
+        response = self.client.post(url, data)
+        expected = ['Project "project_name" not found on ScanCode.io.']
+        self.assertEqual(expected, response.data)
+
+        mock_get_project_data.return_value = {"uuid": uuid.uuid4()}
+        response = self.client.post(url, data)
+        self.assertEqual(status.HTTP_200_OK, response.status_code)
+        expected = {"status": "Packages import from ScanCode.io in progress..."}
+        self.assertEqual(expected, response.data)
+        self.assertEqual(1, ScanCodeProject.objects.count())
+
 
 class ProductRelatedAPITestCase(TestCase):
     def setUp(self):

product_portfolio/views.py

@@ -1961,40 +1961,15 @@ def form_invalid(self, form):
     def get_success_url(self):
         return f"{self.object.get_absolute_url()}#imports"
 
-    def get_project_data(self, project_name_or_uuid):
-        scancodeio = ScanCodeIO(self.request.user)
-        for field_name in ["name", "uuid"]:
-            project_data = scancodeio.find_project(**{field_name: project_name_or_uuid})
-            if project_data:
-                return project_data
-
     def form_valid(self, form):
-        project_name_or_uuid = form.cleaned_data.get("project_name_or_uuid")
-        project_data = self.get_project_data(project_name_or_uuid)
+        self.object = self.get_object()
 
-        if not project_data:
-            msg = f'Project "{project_name_or_uuid}" not found on ScanCode.io.'
-            messages.error(self.request, msg)
+        try:
+            form.submit(product=self.object, user=self.request.user)
+        except ValidationError as error:
+            messages.error(self.request, error)
             return redirect(self.object.get_absolute_url())
 
-        scancode_project = ScanCodeProject.objects.create(
-            product=self.object,
-            dataspace=self.object.dataspace,
-            type=ScanCodeProject.ProjectType.PULL_FROM_SCANCODEIO,
-            project_uuid=project_data.get("uuid"),
-            update_existing_packages=form.cleaned_data.get("update_existing_packages"),
-            scan_all_packages=False,
-            status=ScanCodeProject.Status.SUBMITTED,
-            created_by=self.request.user,
-        )
-
-        transaction.on_commit(
-            lambda: tasks.pull_project_data_from_scancodeio.delay(
-                scancodeproject_uuid=scancode_project.uuid,
-            )
-        )
-
-        project_name = project_data.get("name")
-        msg = f'Packages import from ScanCode.io "{project_name}" in progress...'
+        msg = "Packages import from ScanCode.io in progress..."
         messages.success(self.request, msg)
         return super().form_valid(form)