Clean up code and improve formatting #1763

Signed-off-by: Chin Yeung Li <tli@nexb.com>

Author: chinyeungli

scanpipe/pipelines/scan_maven_package.py

@@ -37,8 +37,8 @@ class ScanMavenPackage(ScanSinglePackage, DeployToDevelop):
     This pipeline takes a Maven PURL as input, fetches the binary and
     source archives (if they exist), and then performs scans for package
     metadata, declared dependencies, licenses, license clarity scores, and
-    copyrights. It also performs a D2D scan if both the source and binary
-    archives are available.
+    copyrights. It also performs a deployment to development relation scan
+    if both the source and binary archives are available.
 
     The output is a summary of the scan results in JSON format.
     """
@@ -76,7 +76,10 @@ def fetch_inputs(self):
         self.to_files = [to_file]
 
     def d2d_check(self):
-        """Set D2D enable if both from and to files are present."""
+        """
+        Enable the deployment to development relation scan if both from
+        and to files are present.
+        """
         self.d2d_enabled = False
         if self.from_files and self.to_files:
             self.d2d_enabled = True
@@ -98,7 +101,7 @@ def extract_input(self):
             self.extract_inputs_to_codebase_directory()
 
     def maven_d2d_steps(self):
-        """Run D2D steps for Maven projects."""
+        """Run deployment to development relation scan steps for Maven projects."""
         if self.d2d_enabled:
             self.collect_and_create_codebase_resources()
             self.fingerprint_codebase_directories()
@@ -116,11 +119,6 @@ def maven_d2d_steps(self):
             self.d2d_process()
 
     def d2d_ecosystem_config(self):
-        # The following langages will be included:
-        # - Java
-        # - Kotlin
-        # - Scala
-        # - JavaScript
         options = ["Java", "Kotlin", "Scala", "JavaScript"]
         d2d_config.load_ecosystem_config(pipeline=self, options=options)
 
@@ -178,5 +176,8 @@ def fetch_and_scan_remote_pom(self):
                 )
 
     def update_package_license_from_resource_if_missing(self):
-        """Update PACKAGE license from the license detected in RESOURCES if missing."""
+        """
+        Fill in missing package licenses using licenses detected in
+        their resources.
+        """
         update_package_license_from_resource_if_missing(self.project)

scanpipe/pipes/maven.py

@@ -25,6 +25,7 @@
 from urllib.parse import urlparse
 
 import requests
+from license_expression import Licensing
 from packageurl import PackageURL
 from packageurl.contrib import purl2url
 
@@ -46,19 +47,11 @@ def check_input_and_return_purl(project):
     if input_purl.type != "maven":
         error_msg = "Only maven purl is supported."
         raise ValueError(error_msg)
-    # Version is required
     if not input_purl.version:
         error_msg = "Version is required."
         raise ValueError(error_msg)
 
-    purl = PackageURL(
-        type=input_purl.type,
-        namespace=input_purl.namespace,
-        name=input_purl.name,
-        version=input_purl.version,
-    )
-
-    return purl
+    return input_purl
 
 
 def fetch_inputs(purl):
@@ -88,11 +81,10 @@ def fetch_and_scan_remote_pom(input_purl, scan_output_location):
     """Fetch the .pom file from from maven.org if not present in codebase."""
     with open(scan_output_location) as file:
         data = json.load(file)
-        # Return and do nothing if data has pom.xml
-        for file in data["files"]:
-            if "pom.xml" in file["path"]:
-                return []
-    packages = data.get("packages", [])
+    # Return and do nothing if data has pom.xml
+    for file in data["files"]:
+        if "pom.xml" in file["path"]:
+            return []
 
     pom_url = get_pom_url(input_purl)
     if not pom_url:
@@ -103,14 +95,18 @@ def fetch_and_scan_remote_pom(input_purl, scan_output_location):
     scanning_errors = scan_pom_file(pom_file)
 
     scanned_pom_packages, scanned_dependencies = update_datafile_paths(pom_file)
+    updated_data = update_scan_data(data, scanned_pom_packages, scanned_dependencies)
 
-    updated_packages = packages + scanned_pom_packages
-    # Replace/Update the package and dependencies section
-    data["packages"] = updated_packages
-    data["dependencies"] = scanned_dependencies
     with open(scan_output_location, "w") as file:
-        json.dump(data, file, indent=2)
-    return scanning_errors
+        json.dump(updated_data, file, indent=2)
+    return [scanning_errors]
+
+
+def update_scan_data(data, scanned_packages, scanned_dependencies):
+    """Update packages and dependencies data"""
+    data["packages"] = data.get("packages", []) + scanned_packages
+    data["dependencies"] = scanned_dependencies
+    return data
 
 
 def get_pom_url(input_purl):
@@ -135,25 +131,9 @@ def get_pom_url(input_purl):
 
 def download_pom_file(pom_url):
     """Fetch the pom file from the input pom_url"""
-    # PO: Could we use fetchcode to fetch instead? Yes, we could, but
-    # the issue is do we want to. Following is the code if we switch to
-    # fetchcode which seems making things complicated, OR we can move
-    # the "fetch_http" to fetchcode and use it.
-    """
-    import os
-    import fetchcode
-    downloaded_pom = fetchcode.fetch(pom_url)
-    location = str(downloaded_pom.location)
-    path = location + ".pom"
-    # The fetch function from fetchcode save the file as /tmp/name
-    # without an extension. We need to add the ".pom" extension so that
-    # the package scan can work properly for this file.
-    os.rename(location, path)
-    """
     try:
         downloaded_pom = fetch.fetch_http(pom_url)
     except requests.RequestException:
-        # Return an empty dictionary
         return {}
     path = str(downloaded_pom.path)
     pom_file_dict = {}
@@ -165,21 +145,18 @@ def download_pom_file(pom_url):
 
 def scan_pom_file(pom_file_dict):
     """Fetch and scan the pom file from the input pom_urls"""
-    scan_errors = []
     pom_file_path = pom_file_dict.get("pom_file_path", "")
     scanned_pom_output_path = pom_file_dict.get("output_path", "")
 
     # Run a package scan on the fetched pom.xml
-    scanning_errors = scancode.run_scan(
+    # Return scanning errors, if present
+    return scancode.run_scan(
         location=pom_file_path,
         output_file=scanned_pom_output_path,
         run_scan_args={
             "package": True,
         },
     )
-    if scanning_errors:
-        scan_errors.append(scanning_errors)
-    return scan_errors
 
 
 def update_datafile_paths(pom_file_dict):
@@ -197,11 +174,9 @@ def update_datafile_paths(pom_file_dict):
     scanned_dependencies = scanned_pom_data.get("dependencies", [])
 
     for scanned_package in scanned_packages:
-        # Replace the 'datafile_path' with the pom_url
         scanned_package["datafile_paths"] = [pom_url]
         scanned_pom_packages.append(scanned_package)
     for scanned_dep in scanned_dependencies:
-        # Replace the 'datafile_path' with empty string
         # See https://github.com/aboutcode-org/scancode.io/issues/1763#issuecomment-3525165830
         scanned_dep["datafile_path"] = ""
         scanned_pom_deps.append(scanned_dep)
@@ -210,20 +185,20 @@ def update_datafile_paths(pom_file_dict):
 
 def update_package_license_from_resource_if_missing(project):
     """Populate missing licenses to packages based on resource data."""
-    from license_expression import Licensing
-
     for package in project.discoveredpackages.all():
         if not package.get_declared_license_expression():
             package_uid = package.package_uid
-            detected_lics = []
+            detected_licenses = []
             for resource in project.codebaseresources.has_license_expression():
                 for for_package in resource.for_packages:
                     if for_package == package_uid:
-                        detected_lic_exp = resource.detected_license_expression
-                        if detected_lic_exp not in detected_lics:
-                            detected_lics.append(detected_lic_exp)
-            if detected_lics:
-                lic_exp = " AND ".join(detected_lics)
-                declared_lic_exp = str(Licensing().dedup(lic_exp))
-                package.declared_license_expression = declared_lic_exp
+                        detected_license_expression = (
+                            resource.detected_license_expression
+                        )
+                        if detected_license_expression not in detected_licenses:
+                            detected_licenses.append(detected_license_expression)
+            if detected_licenses:
+                license_expression = " AND ".join(detected_licenses)
+                declared_license_expression = str(Licensing().dedup(license_expression))
+                package.declared_license_expression = declared_license_expression
                 package.save()

scanpipe/tests/pipes/test_maven.py

@@ -241,66 +241,27 @@ def test_scanpipe_maven_fetch_and_scan_remote_pom_no_pom_file(
         )
         self.assertEqual(result, ["Failed to download the POM file."])
 
-    @mock.patch("builtins.open", new_callable=mock.mock_open)
-    @mock.patch("json.load")
-    @mock.patch("json.dump")
-    @mock.patch("scanpipe.pipes.maven.get_pom_url")
-    @mock.patch("scanpipe.pipes.maven.download_pom_file")
-    @mock.patch("scanpipe.pipes.maven.scan_pom_file")
-    @mock.patch("scanpipe.pipes.maven.update_datafile_paths")
-    def test_scanpipe_maven_fetch_and_scan_remote_pom_success(
-        self,
-        mock_update_datafile_paths,
-        mock_scan_pom_file,
-        mock_download_pom_file,
-        mock_get_pom_url,
-        mock_json_dump,
-        mock_json_load,
-        mock_open,
-    ):
-        mock_json_load.return_value = {
-            "files": [],
-            "packages": [{"name": "sample-package"}],
-        }
-        mock_get_pom_url.return_value = "https://example.com/test.pom"
-        mock_download_pom_file.return_value = {
-            "pom_file_path": "/project/path.pom",
-            "output_path": "/project/out.json",
-        }
-        mock_scan_pom_file.return_value = []
+    def test_update_scan_data(self):
+        original_data = {"packages": [{"name": "package1"}], "dependencies": []}
+        new_package = [{"name": "package2"}]
+        new_dependency = [{"name": "dep1"}]
 
-        mock_update_datafile_paths.return_value = (
-            [{"name": "new-package"}],
-            [{"name": "new-dependency"}],
-        )
+        result = maven.update_scan_data(original_data, new_package, new_dependency)
 
-        result = maven.fetch_and_scan_remote_pom(
-            "pkg:maven/org/test@1.0", "/path/to/output.json"
+        self.assertEqual(
+            result["packages"], [{"name": "package1"}, {"name": "package2"}]
         )
-        self.assertEqual(result, [])
-
-        expected_saved_data = {
-            "files": [],
-            "packages": [{"name": "sample-package"}, {"name": "new-package"}],
-            "dependencies": [{"name": "new-dependency"}],
-        }
-
-        self.assertTrue(mock_json_dump.called)
-
-        args, _kwargs = mock_json_dump.call_args
-        # Get the data as in the first argument: json.dump(data, file, indent=2)
-        data = args[0]
-        self.assertEqual(data, expected_saved_data)
+        self.assertEqual(result["dependencies"], [{"name": "dep1"}])
 
     @mock.patch("scanpipe.pipes.maven.scancode.run_scan")
     def test_scanpipe_maven_scan_pom_file(self, mock_run_scan):
         pom_file_dict = {
             "pom_file_path": "/main/mock.pom",
             "output_path": "/main/mock.pom-output.json",
         }
-        mock_run_scan.return_value = None
+        mock_run_scan.return_value = {}
         result = maven.scan_pom_file(pom_file_dict)
-        self.assertEqual(result, [])
+        self.assertEqual(result, {})
 
         mock_run_scan.assert_called_once_with(
             location="/main/mock.pom",