Add a REST API "action" endpoint to track product imports #273

CHANGELOG.rst

@@ -35,6 +35,10 @@ Release notes
   code_view_url, vcs_url, api_data_url, size, md5, sha1, sha256, sha512.
   https://github.com/aboutcode-org/dejacode/issues/255
 
+- Add a Product REST API "action" endpoint to track product imports and their status:
+  * `/products/{uuid}/imports/`
+  https://github.com/aboutcode-org/dejacode/issues/273
+
 ### Version 5.3.0
 
 - Rename ProductDependency is_resolved to is_pinned.

product_portfolio/api.py

@@ -47,6 +47,7 @@
 from product_portfolio.models import ProductComponent
 from product_portfolio.models import ProductDependency
 from product_portfolio.models import ProductPackage
+from product_portfolio.models import ScanCodeProject
 from vulnerabilities.api import VulnerabilityAnalysisSerializer
 
 base_extra_kwargs = {
@@ -289,6 +290,26 @@ class PullProjectDataSerializer(serializers.Serializer):
     )
 
 
+class ScanCodeProjectSerializer(DataspacedSerializer):
+    input_filename = serializers.ReadOnlyField(source="input_file_filename")
+
+    class Meta:
+        model = ScanCodeProject
+        fields = (
+            "uuid",
+            "type",
+            "project_uuid",
+            "input_filename",
+            "update_existing_packages",
+            "scan_all_packages",
+            "status",
+            "import_log",
+            "results",
+            "created_date",
+            "last_modified_date",
+        )
+
+
 class ProductViewSet(
     SendAboutFilesMixin,
     AboutCodeFilesActionMixin,
@@ -346,6 +367,18 @@ def perform_create(self, serializer):
         super().perform_create(serializer)
         assign_all_object_permissions(self.request.user, serializer.instance)
 
+    @action(detail=True)
+    def imports(self, request, uuid):
+        """
+        List of Product Imports, including their current status, log, and results.
+
+        Statuses: "submitted", "importing", "success", "failure", "warning".
+        """
+        product = self.get_object()
+        scancode_projects = product.scancodeprojects.all()
+        projects_data = ScanCodeProjectSerializer(scancode_projects, many=True).data
+        return Response(projects_data)
+
     @action(detail=True, methods=["post"], serializer_class=LoadSBOMsFormSerializer)
     def load_sboms(self, request, *args, **kwargs):
         """

product_portfolio/tests/test_api.py

@@ -352,6 +352,41 @@ def test_api_product_endpoint_update_permissions(self):
         product1 = Product.unsecured_objects.get(pk=self.product1.pk)
         self.assertEqual("Updated Name", product1.name)
 
+    def test_api_product_endpoint_imports_action(self):
+        url = reverse("api_v2:product-imports", args=[self.product1.uuid])
+
+        self.client.login(username=self.base_user.username, password="secret")
+        response = self.client.get(url)
+        self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)
+
+        # Required permissions
+        add_perm(self.base_user, "add_product")
+        assign_perm("view_product", self.base_user, self.product1)
+
+        response = self.client.get(url)
+        self.assertEqual(status.HTTP_200_OK, response.status_code)
+        self.assertEqual(0, self.product1.scancodeprojects.count())
+        self.assertEqual([], response.data)
+
+        ScanCodeProject.objects.create(
+            product=self.product1,
+            dataspace=self.product1.dataspace,
+            type=ScanCodeProject.ProjectType.LOAD_SBOMS,
+            status=ScanCodeProject.Status.SUCCESS,
+            input_file=ContentFile("Data", name="data.json"),
+        )
+
+        response = self.client.get(url)
+        self.assertEqual(status.HTTP_200_OK, response.status_code)
+        self.assertEqual(1, self.product1.scancodeprojects.count())
+        self.assertEqual(1, len(response.data))
+        entry = response.data[0]
+        self.assertEqual(ScanCodeProject.ProjectType.LOAD_SBOMS, entry["type"])
+        self.assertEqual("data.json", entry["input_filename"])
+        self.assertEqual(ScanCodeProject.Status.SUCCESS, entry["status"])
+        self.assertEqual([], entry["import_log"])
+        self.assertEqual({}, entry["results"])
+
     def test_api_product_endpoint_load_sboms_action(self):
         url = reverse("api_v2:product-load-sboms", args=[self.product1.uuid])