Implement "update_package_license_from_resource_if_missing" function #1763

- Update package's license if missing while the same package has license detected in RESOURCES

Signed-off-by: Chin Yeung Li <[email protected]>

Author: chinyeungli

scanpipe/pipelines/scan_maven_package.py

@@ -47,6 +47,7 @@ def steps(cls):
             cls.extract_archives,
             cls.run_scan,
             cls.fetch_and_scan_remote_pom,
+            cls.update_package_license_from_resource_if_missing,
             cls.load_inventory_from_toolkit_scan,
             cls.make_summary_from_scan_results,
         )
@@ -65,9 +66,9 @@ def fetch_and_scan_remote_pom(self):
         pom_file_list = download_pom_files(pom_url_list)
         scanned_pom_packages, scanned_dependencies = scan_pom_files(pom_file_list)
 
-        updated_pacakges = packages + scanned_pom_packages
+        updated_packages = packages + scanned_pom_packages
         # Replace/Update the package and dependencies section
-        data["packages"] = updated_pacakges
+        data["packages"] = updated_packages
         data["dependencies"] = scanned_dependencies
         with open(self.scan_output_location, "w") as file:
             json.dump(data, file, indent=2)

scanpipe/pipelines/scan_single_package.py

@@ -31,6 +31,7 @@
 from scanpipe.pipes import scancode
 from scanpipe.pipes.input import copy_input
 from scanpipe.pipes.input import is_archive
+from scanpipe.pipes.resolve import update_package_license_from_resource_if_missing
 
 
 class ScanSinglePackage(Pipeline):
@@ -51,6 +52,7 @@ def steps(cls):
             cls.extract_input_to_codebase_directory,
             cls.extract_archives,
             cls.run_scan,
+            cls.update_package_license_from_resource_if_missing,
             cls.load_inventory_from_toolkit_scan,
             cls.make_summary_from_scan_results,
         )
@@ -126,6 +128,23 @@ def run_scan(self):
         if not scan_output_path.exists():
             raise FileNotFoundError("ScanCode output not available.")
 
+    def update_package_license_from_resource_if_missing(self):
+        """Update PACKAGE license from the license detected in RESOURCES if missing."""
+        with open(self.scan_output_location) as file:
+            data = json.load(file)
+            packages = data.get("packages", [])
+            resources = data.get("files", [])
+            if not packages or not resources:
+                return
+
+        updated_packages = update_package_license_from_resource_if_missing(
+            packages, resources
+        )
+        # Update the package section
+        data["packages"] = updated_packages
+        with open(self.scan_output_location, "w") as file:
+            json.dump(data, file, indent=2)
+
     def load_inventory_from_toolkit_scan(self):
         """Process a JSON Scan results to populate codebase resources and packages."""
         input.load_inventory_from_toolkit_scan(self.project, self.scan_output_location)

scanpipe/pipes/resolve.py

@@ -723,3 +723,32 @@ def scan_pom_files(pom_file_list):
                     scanned_dep["datafile_path"] = ""
                     scanned_pom_deps.append(scanned_dep)
     return scanned_pom_packages, scanned_pom_deps
+
+
+def update_package_license_from_resource_if_missing(packages, resources):
+    """Populate missing licenses to packages based on resource data."""
+    from license_expression import Licensing
+
+    updated_packages = []
+    for package in packages:
+        if not package.get("declared_license_expression"):
+            package_uid = package.get("package_uid")
+            detected_lic_list = []
+            for resource in resources:
+                if (
+                    resource.get("detected_license_expression")
+                    and package_uid in resource["for_packages"]
+                ):
+                    if (
+                        resource.get("detected_license_expression")
+                        not in detected_lic_list
+                    ):
+                        detected_lic_list.append(
+                            resource.get("detected_license_expression")
+                        )
+            license_expression = " AND ".join(detected_lic_list)
+            if license_expression:
+                declared_license_expression = str(Licensing().dedup(license_expression))
+                package["declared_license_expression"] = declared_license_expression
+        updated_packages.append(package)
+    return updated_packages

scanpipe/tests/pipes/test_resolve.py

@@ -586,3 +586,56 @@ def test_scanpipe_resolve_get_pom_url_list_with_invalid_filename(self):
         input_source = {"filename": "not-a-jar.txt"}
         result = resolve.get_pom_url_list(input_source, [])
         self.assertEqual(result, [])
+
+    def test_scanpipe_resolve_update_package_license_from_resource_if_missing(self):
+        packages = [
+            {"package_uid": "pkg1", "declared_license_expression": ""},
+            {"package_uid": "pkg2", "declared_license_expression": None},
+            {"package_uid": "pkg3", "declared_license_expression": "MIT"},
+        ]
+        resources = [
+            {
+                "for_packages": ["pkg1", "pkg2"],
+                "detected_license_expression": "GPL-2.0",
+            },
+            {"for_packages": ["pkg1"], "detected_license_expression": "MIT"},
+        ]
+
+        expected_pkg1_expr = "GPL-2.0 AND MIT"
+        expected_pkg2_expr = "GPL-2.0"
+
+        updated = resolve.update_package_license_from_resource_if_missing(
+            packages, resources
+        )
+
+        self.assertEqual(updated[0]["declared_license_expression"], expected_pkg1_expr)
+        self.assertEqual(updated[1]["declared_license_expression"], expected_pkg2_expr)
+        self.assertEqual(updated[2]["declared_license_expression"], "MIT")
+
+    def test_scanpipe_resolve_update_package_license_from_resource_if_missing_no_match(
+        self,
+    ):
+        packages = [{"package_uid": "pkgX", "declared_license_expression": None}]
+        resources = [{"for_packages": ["pkgY"], "detected_license_expression": "MIT"}]
+
+        updated = resolve.update_package_license_from_resource_if_missing(
+            packages, resources
+        )
+        self.assertEqual(updated[0]["declared_license_expression"], None)
+
+    def test_scanpipe_resolve_update_package_license_from_resource_if_missing_no_change(
+        self,
+    ):
+        packages = [
+            {"package_uid": "pkg1", "declared_license_expression": "GPL-2.0"},
+            {"package_uid": "pkg2", "declared_license_expression": "Apache-2.0"},
+        ]
+        resources = [
+            {"for_packages": ["pkg1", "pkg2"], "detected_license_expression": "MIT"},
+        ]
+
+        updated = resolve.update_package_license_from_resource_if_missing(
+            packages, resources
+        )
+        self.assertEqual(updated[0]["declared_license_expression"], "GPL-2.0")
+        self.assertEqual(updated[1]["declared_license_expression"], "Apache-2.0")