Add unit tests #1145

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

scanpipe/pipelines/load_sbom.py

@@ -20,9 +20,6 @@
 # ScanCode.io is a free software code scanning tool from nexB Inc. and others.
 # Visit https://github.com/nexB/scancode.io for support and download.
 
-import uuid
-
-from scanpipe.models import DiscoveredDependency
 from scanpipe.pipelines.scan_codebase import ScanCodebase
 from scanpipe.pipes import resolve
 
@@ -49,7 +46,7 @@ def steps(cls):
             cls.get_sbom_inputs,
             cls.get_packages_from_sboms,
             cls.create_packages_from_sboms,
-            cls.create_dependencies,
+            cls.create_dependencies_from_sboms,
         )
 
     def get_sbom_inputs(self):
@@ -66,35 +63,12 @@ def get_packages_from_sboms(self):
         )
 
     def create_packages_from_sboms(self):
-        """Create the packages and dependencies from the SBOM, in the database."""
+        """Create the packages declared in the SBOMs."""
         resolve.create_packages_and_dependencies(
             project=self.project,
             packages=self.packages,
         )
 
-    def create_dependencies(self):
-        packages = self.project.discoveredpackages.all()
-        for package in packages.filter(extra_data__has_key="depends_on"):
-            for bom_ref in package.extra_data.get("depends_on", []):
-                try:
-                    resolved_to_package = packages.get(extra_data__bom_ref=bom_ref)
-                except Exception:
-                    self.project.add_error(
-                        description="Could not find resolved_to package entry.",
-                        model="create_dependencies",
-                    )
-                    continue
-                DiscoveredDependency.objects.create(
-                    project=self.project,
-                    dependency_uid=str(uuid.uuid4()),
-                    for_package=package,
-                    resolved_to_package=resolved_to_package,
-                    is_runtime=True,
-                    is_resolved=True,
-                )
-                # dependency_data = {
-                #     "for_package": package,
-                #     "resolved_to_package": resolved_to_package,
-                # }
-                # raise ValueError("A purl string argument is required.")
-                # update_or_create_dependency(self.project, dependency_data)
+    def create_dependencies_from_sboms(self):
+        """Create the dependency relationship declared in the SBOMs."""
+        resolve.create_dependencies_from_packages_extra_data(project=self.project)

scanpipe/pipes/__init__.py

@@ -261,7 +261,7 @@ def update_or_create_dependency(
     where Dependency data is imported from a scancode-toolkit scan, where the
     root path segments are not stripped for `datafile_path`.
     If the dependency is resolved and a resolved package is created, we have the
-    corresponsing package_uid at `resolved_to`.
+    corresponding package_uid at `resolved_to`.
     """
     if ignore_dependency_scope(project, dependency_data):
         return  # Do not create the DiscoveredDependency record.

scanpipe/pipes/resolve.py

@@ -22,14 +22,19 @@
 
 import json
 import sys
+import uuid
 from pathlib import Path
 
+from django.core.exceptions import MultipleObjectsReturned
+from django.core.exceptions import ObjectDoesNotExist
+
 from attributecode.model import About
 from packagedcode import APPLICATION_PACKAGE_DATAFILE_HANDLERS
 from packagedcode.licensing import get_license_detections_and_expression
 from packageurl import PackageURL
 from python_inspector.api import resolve_dependencies
 
+from scanpipe.models import DiscoveredDependency
 from scanpipe.models import DiscoveredPackage
 from scanpipe.pipes import cyclonedx
 from scanpipe.pipes import flag
@@ -108,6 +113,52 @@ def create_packages_and_dependencies(project, packages, resolved=False):
                 update_or_create_dependency(project, dependency_data)
 
 
+def create_dependencies_from_packages_extra_data(project):
+    """
+    Create Dependency objects from the Package extra_data values.
+    The Package instances need to be saved first in the database before creating the
+    Dependency objects.
+    The dependencies declared in the SBOM are stored on the Package.extra_data field
+    and resolved as Dependency objects in this function.
+    """
+    project_packages = project.discoveredpackages.all()
+    created_count = 0
+
+    packages_with_depends_on = project_packages.filter(
+        extra_data__has_key="depends_on"
+    ).prefetch_related("codebase_resources")
+
+    for for_package in packages_with_depends_on:
+        datafile_resource = None
+        codebase_resources = for_package.codebase_resources.all()
+        if len(codebase_resources) == 1:
+            datafile_resource = codebase_resources[0]
+
+        for bom_ref in for_package.extra_data.get("depends_on", []):
+            try:
+                resolved_to_package = project_packages.get(extra_data__bom_ref=bom_ref)
+            except (ObjectDoesNotExist, MultipleObjectsReturned):
+                project.add_error(
+                    description=f"Could not find resolved_to package entry: {bom_ref}.",
+                    model="create_dependencies",
+                )
+                continue
+
+            DiscoveredDependency.objects.create(
+                project=project,
+                dependency_uid=str(uuid.uuid4()),
+                for_package=for_package,
+                resolved_to_package=resolved_to_package,
+                datafile_resource=datafile_resource,
+                is_runtime=True,
+                is_resolved=True,
+                is_direct=True,
+            )
+            created_count += 1
+
+    return created_count
+
+
 def get_packages_from_manifest(input_location, package_registry=None):
     """
     Resolve packages or get packages data from a package manifest file/

scanpipe/tests/pipes/test_resolve.py

@@ -30,6 +30,7 @@
 from scanpipe.pipes import resolve
 from scanpipe.pipes.input import copy_inputs
 from scanpipe.pipes.scancode import extract_archives
+from scanpipe.tests import make_package
 from scanpipe.tests import package_data1
 
 
@@ -235,6 +236,32 @@ def test_scanpipe_resolve_create_packages_and_dependencies(self):
         package = project1.discoveredpackages.get()
         self.assertEqual(resource1, package.codebase_resources.get())
 
+    def test_scanpipe_resolve_create_dependencies_from_packages_extra_data(self):
+        p1 = Project.objects.create(name="Analysis")
+        parent_extra_data = {"depends_on": ["child"]}
+        parent = make_package(p1, "pkg:type/parent", extra_data=parent_extra_data)
+        child = make_package(p1, "pkg:type/child", extra_data={"bom_ref": "child"})
+
+        created_count = resolve.create_dependencies_from_packages_extra_data(p1)
+        self.assertEqual(1, created_count)
+        dependency = p1.discovereddependencies.get()
+        self.assertEqual(parent, dependency.for_package)
+        self.assertEqual(child, dependency.resolved_to_package)
+        self.assertTrue(dependency.is_runtime)
+        self.assertTrue(dependency.is_resolved)
+        self.assertTrue(dependency.is_direct)
+        self.assertFalse(dependency.is_optional)
+
+        parent.update_extra_data({"depends_on": ["unknown"]})
+        created_count = resolve.create_dependencies_from_packages_extra_data(p1)
+        self.assertEqual(0, created_count)
+        message = p1.projectmessages.get()
+        self.assertEqual("error", message.severity)
+        self.assertEqual(
+            "Could not find resolved_to package entry: unknown.", message.description
+        )
+        self.assertEqual("create_dependencies", message.model)
+
     def test_scanpipe_resolve_get_manifest_headers(self):
         input_location = self.data / "manifests" / "toml.spdx.json"
         resource = mock.Mock(location=input_location)

scanpipe/tests/test_pipelines.py

@@ -1416,6 +1416,24 @@ def test_scanpipe_load_sbom_pipeline_cyclonedx_integration(self):
             )
             self.assertEqual(expected["filename"], package.filename)
 
+    def test_scanpipe_load_sbom_pipeline_cyclonedx_with_dependencies_integration(self):
+        pipeline_name = "load_sbom"
+        project1 = Project.objects.create(name="Analysis")
+
+        input_location = self.data / "cyclonedx" / "laravel-7.12.0" / "bom.1.4.json"
+        project1.copy_input_from(input_location)
+
+        run = project1.add_pipeline(pipeline_name)
+        pipeline = run.make_pipeline_instance()
+
+        exitcode, out = pipeline.execute()
+        self.assertEqual(0, exitcode, msg=out)
+
+        self.assertEqual(62, project1.discoveredpackages.count())
+        self.assertEqual(112, project1.discovereddependencies.count())
+        dependency = project1.discovereddependencies.all()[0]
+        self.assertEqual("bom.1.4.json", str(dependency.datafile_resource))
+
     @mock.patch("scanpipe.pipes.purldb.request_post")
     @mock.patch("uuid.uuid4")
     def test_scanpipe_deploy_to_develop_pipeline_integration(