Refactor the package lookups into a function #295 (#300)

Signed-off-by: tdruez <[email protected]>

Author: tdruez

product_portfolio/importers.py

@@ -8,6 +8,7 @@
 
 import json
 from collections import defaultdict
+from contextlib import suppress
 
 from django import forms
 from django.core.exceptions import ValidationError
@@ -692,29 +693,8 @@ def import_package(self, package_data):
         # Vulnerabilities are fetched post import.
         package_data.pop("affected_by_vulnerabilities", None)
 
-        unique_together_lookups = {
-            field: package_data.get(field, "") for field in self.unique_together_fields
-        }
-
-        # Check if the Package already exists in the local Dataspace
-        try:
-            package = Package.objects.scope(self.user.dataspace).get(**unique_together_lookups)
-            self.existing["package"].append(str(package))
-        except ObjectDoesNotExist:
-            package = None
-
-        # Check if the Package already exists in the reference Dataspace
-        reference_dataspace = Dataspace.objects.get_reference()
-        user_dataspace = self.user.dataspace
-        if not package and user_dataspace != reference_dataspace:
-            qs = Package.objects.scope(reference_dataspace).filter(**unique_together_lookups)
-            if qs.exists():
-                reference_object = qs.first()
-                try:
-                    package = copy_object(reference_object, user_dataspace, self.user, update=False)
-                    self.created["package"].append(str(package))
-                except IntegrityError as error:
-                    self.errors["package"].append(str(error))
+        # Check if the package already exists to prevent duplication.
+        package = self.look_for_existing_package(package_data)
 
         if license_expression := package_data.get("declared_license_expression"):
             license_expression = str(self.licensing.dedup(license_expression))
@@ -786,3 +766,43 @@ def import_dependency(self, dependency_data):
             return
 
         self.created["dependency"].append(str(dependency.dependency_uid))
+
+    def look_for_existing_package(self, package_data):
+        package = None
+        package_qs = Package.objects.scope(self.user.dataspace)
+
+        # 1. Check if the Package already exists in the local Dataspace
+        # Using exact match first: purl + download_url + filename
+        exact_match_lookups = {
+            field: package_data.get(field, "") for field in self.unique_together_fields
+        }
+        with suppress(ObjectDoesNotExist):
+            package = package_qs.get(**exact_match_lookups)
+            self.existing["package"].append(str(package))
+            return package
+
+        # 2. If the package data does not include a download_url value:
+        # Attemp to find an existing package using purl-only match.
+        if not package_data.get("download_url"):
+            purl_lookups = {field: package_data.get(field, "") for field in PACKAGE_URL_FIELDS}
+            same_purl_packages = package_qs.filter(**purl_lookups)
+            if len(same_purl_packages) == 1:
+                package = same_purl_packages[0]
+                self.existing["package"].append(str(package))
+                return package
+
+        # Check if the Package already exists in the reference Dataspace
+        # Using exact match only: purl + download_url + filename
+        reference_dataspace = Dataspace.objects.get_reference()
+        user_dataspace = self.user.dataspace
+        if not package and user_dataspace != reference_dataspace:
+            qs = Package.objects.scope(reference_dataspace).filter(**exact_match_lookups)
+            if qs.exists():
+                reference_object = qs.first()
+                try:
+                    package = copy_object(reference_object, user_dataspace, self.user, update=False)
+                    self.created["package"].append(str(package))
+                except IntegrityError as error:
+                    self.errors["package"].append(str(error))
+
+        return package

product_portfolio/tests/test_importers.py

@@ -1121,6 +1121,49 @@ def test_product_portfolio_import_packages_from_scio_importer_multiple_package_o
         self.assertEqual({}, existing)
         self.assertEqual({}, errors)
 
+    @mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_dependencies")
+    @mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_packages")
+    def test_product_portfolio_import_packages_from_scio_importer_look_for_existing_package(
+        self, mock_fetch_packages, mock_fetch_dependencies
+    ):
+        purl = "pkg:maven/org.apache.activemq/[email protected]"
+        filename = "activemq-camel.zip"
+        download_url = "https://download.url/activemq-camel.zip"
+        package1 = make_package(self.dataspace, package_url=purl)
+        package2 = make_package(self.dataspace, package_url=purl, filename=filename)
+
+        package_data = {
+            "type": "maven",
+            "namespace": "org.apache.activemq",
+            "name": "activemq-camel",
+            "version": "5.11.0",
+            "purl": purl,
+        }
+        mock_fetch_packages.return_value = [package_data]
+        mock_fetch_dependencies.return_value = []
+
+        importer = ImportPackageFromScanCodeIO(
+            user=self.super_user,
+            project_uuid=uuid.uuid4(),
+            product=self.product1,
+        )
+
+        # Check if the Package already exists in the local Dataspace
+        # Using exact match first: purl + download_url + filename
+        package = importer.look_for_existing_package(package_data)
+        self.assertEqual(package1, package)
+
+        # 2 packages are matched, cannot defined the one that should be used
+        package1.update(download_url=download_url)
+        package = importer.look_for_existing_package(package_data)
+        self.assertIsNone(package)
+
+        # If the package data does not include a download_url value:
+        # Attemp to find an existing package using purl-only match.
+        package2.delete()
+        package = importer.look_for_existing_package(package_data)
+        self.assertEqual(package1, package)
+
     @mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_dependencies")
     @mock.patch("dejacode_toolkit.scancodeio.ScanCodeIO.fetch_project_packages")
     def test_product_portfolio_import_packages_from_scio_importer_duplicate_dependency(