Refactor code and add tests #1763

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

Author: chinyeungli

scanpipe/pipelines/scan_maven_package.py

@@ -23,8 +23,9 @@
 import json
 
 from scanpipe.pipelines.scan_single_package import ScanSinglePackage
-from scanpipe.pipes.resolve import download_and_scan_pom_file
+from scanpipe.pipes.resolve import download_pom_files
 from scanpipe.pipes.resolve import get_pom_url_list
+from scanpipe.pipes.resolve import scan_pom_files
 
 
 class ScanMavenPackage(ScanSinglePackage):
@@ -51,7 +52,7 @@ def steps(cls):
         )
 
     def fetch_and_scan_remote_pom(self):
-        """Fetch the pom.xml file from from maven.org if not present in codebase."""
+        """Fetch the .pom file from from maven.org if not present in codebase."""
         with open(self.scan_output_location) as file:
             data = json.load(file)
             # Return and do nothing if data has pom.xml
@@ -61,9 +62,8 @@ def fetch_and_scan_remote_pom(self):
             packages = data.get("packages", [])
 
         pom_url_list = get_pom_url_list(self.project.input_sources[0], packages)
-        scanned_pom_packages, scanned_dependencies = download_and_scan_pom_file(
-            pom_url_list
-        )
+        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
         # Replace/Update the package and dependencies section

scanpipe/pipes/resolve.py

@@ -586,10 +586,13 @@ def get_pom_url_list(input_source, packages):
             )
             pom_url_list.append(pom_url)
     else:
+        from urllib.parse import urlparse
+
         # Check what's the input source
         input_source_url = input_source.get("download_url", "")
 
-        if input_source_url and "maven.org/" in input_source_url:
+        parsed_url = urlparse(input_source_url)
+        if input_source_url and parsed_url.netloc.endswith("maven.org"):
             base_url = input_source_url.rsplit("/", 1)[0]
             pom_url = (
                 base_url + "/" + "-".join(base_url.rstrip("/").split("/")[-2:]) + ".pom"
@@ -672,17 +675,32 @@ def is_maven_pom_url(url):
         return False
 
 
-def download_and_scan_pom_file(pom_url_list):
+def download_pom_files(pom_url_list):
+    """Fetch the pom file from the input pom_url_list"""
+    pom_file_list = []
+    for pom_url in pom_url_list:
+        pom_file_dict = {}
+        downloaded_pom = fetch.fetch_http(pom_url)
+        print("download_pom.path", str(downloaded_pom.path))
+        pom_file_dict["pom_file_path"] = str(downloaded_pom.path)
+        pom_file_dict["output_path"] = str(downloaded_pom.path) + "-output.json"
+        pom_file_dict["pom_url"] = pom_url
+        pom_file_list.append(pom_file_dict)
+    return pom_file_list
+
+
+def scan_pom_files(pom_file_list):
     """Fetch and scan the pom file from the input pom_url_list"""
     scanned_pom_packages = []
     scanned_pom_deps = []
-    for pom_url in pom_url_list:
-        downloaded_pom = fetch.fetch_http(pom_url)
-        scanned_pom_output_path = str(downloaded_pom.path) + "-output.json"
+    for pom_file_dict in pom_file_list:
+        pom_file_path = pom_file_dict.get("pom_file_path", "")
+        scanned_pom_output_path = pom_file_dict.get("output_path", "")
+        pom_url = pom_file_dict.get("pom_url", "")
 
         # Run a package scan on the fetched pom.xml
         _scanning_errors = scancode.run_scan(
-            location=str(downloaded_pom.path),
+            location=pom_file_path,
             output_file=scanned_pom_output_path,
             run_scan_args={
                 "package": True,

scanpipe/tests/pipes/test_resolve.py

@@ -400,3 +400,189 @@ def test_scanpipe_resolve_parse_maven_filename(self):
         self.assertEqual(result3_version, expected3_version)
         self.assertEqual(result4_name, expected2_name)
         self.assertEqual(result4_version, expected2_version)
+
+    @mock.patch("requests.get")
+    def test_scanpipe_resolve_is_maven_pom_url_valid(self, mock_get):
+        mock_response = mock.Mock()
+        mock_response.status_code = 200
+        mock_response.headers = {"content-type": "application/xml"}
+        mock_response.text = '<?xml version="1.0"?><project></project>'
+        mock_get.return_value = mock_response
+
+        result = resolve.is_maven_pom_url(
+            "https://repo1.maven.org/maven2/example/example.pom"
+        )
+        self.assertTrue(result)
+
+    @mock.patch("requests.get")
+    def test_scanpipe_resolve_is_maven_pom_url_404(self, mock_get):
+        mock_response = mock.Mock()
+        mock_response.status_code = 404
+        mock_get.return_value = mock_response
+
+        result = resolve.is_maven_pom_url(
+            "https://repo.maven.apache.org/maven2/example/404.pom"
+        )
+        self.assertFalse(result)
+
+    @mock.patch("requests.get")
+    def test_scanpipe_resolve_is_maven_pom_url_error(self, mock_get):
+        mock_response = mock.Mock()
+        mock_response.status_code = 200
+        mock_response.headers = {"content-type": "text/html"}
+        mock_response.text = "<html>Error page</html>"
+        mock_get.return_value = mock_response
+
+        result = resolve.is_maven_pom_url(
+            "https://repo.maven.apache.org/maven2/example/error.pom"
+        )
+        self.assertFalse(result)
+
+    @mock.patch("scanpipe.pipes.resolve.fetch.fetch_http")
+    def test_scanpipe_resolve_download_pom_files(self, mock_fetch_http):
+        mock_response = mock.Mock()
+        mock_response.path = "/safe/example1.pom"
+        mock_fetch_http.return_value = mock_response
+
+        pom_urls = ["https://repo1.maven.org/maven2/example/example1.pom"]
+
+        expected = [
+            {
+                "pom_file_path": "/safe/example1.pom",
+                "output_path": "/safe/example1.pom-output.json",
+                "pom_url": "https://repo1.maven.org/maven2/example/example1.pom",
+            }
+        ]
+
+        result = resolve.download_pom_files(pom_urls)
+        self.assertEqual(result, expected)
+
+    @mock.patch("scanpipe.pipes.resolve.scancode.run_scan")
+    @mock.patch("builtins.open", new_callable=mock.mock_open)
+    @mock.patch("json.load")
+    def test_scanpipe_resolve_scan_pom_files(
+        self, mock_json_load, mock_open, mock_run_scan
+    ):
+        mock_json_load.return_value = {
+            "packages": [
+                {
+                    "name": "example-package",
+                    "version": "1.0.0",
+                    "datafile_paths": ["/safe/mock_pom.xml"],
+                }
+            ],
+            "dependencies": [
+                {
+                    "name": "example-dep",
+                    "version": "2.0.0",
+                    "datafile_path": "/safe/mock_pom.xml",
+                }
+            ],
+        }
+
+        pom_file_list = [
+            {
+                "pom_file_path": "/safe/mock.pom",
+                "output_path": "/safe/mock.pom-output.json",
+                "pom_url": "https://repo1.maven.org/maven2/example/example.pom",
+            }
+        ]
+
+        expected_packages = [
+            {
+                "name": "example-package",
+                "version": "1.0.0",
+                "datafile_paths": [
+                    "https://repo1.maven.org/maven2/example/example.pom"
+                ],
+            }
+        ]
+        expected_deps = [
+            {"name": "example-dep", "version": "2.0.0", "datafile_path": ""}
+        ]
+
+        packages, deps = resolve.scan_pom_files(pom_file_list)
+
+        self.assertEqual(packages, expected_packages)
+        self.assertEqual(deps, expected_deps)
+
+        mock_run_scan.assert_called_once_with(
+            location="/safe/mock.pom",
+            output_file="/safe/mock.pom-output.json",
+            run_scan_args={"package": True},
+        )
+        mock_open.assert_called_once_with("/safe/mock.pom-output.json")
+        mock_json_load.assert_called_once()
+
+    @mock.patch("scanpipe.pipes.resolve.is_maven_pom_url")
+    @mock.patch("scanpipe.pipes.resolve.requests.get")
+    def test_scanpipe_resolve_construct_pom_url_from_filename(
+        self, mock_get, mock_is_maven_pom_url
+    ):
+        # Setup mock response from Maven Central
+        mock_response = mock.Mock()
+        mock_response.raise_for_status.return_value = None
+        mock_response.json.return_value = {
+            "response": {"docs": [{"g": "org.apache.commons"}]}
+        }
+        mock_get.return_value = mock_response
+        mock_is_maven_pom_url.return_value = True
+
+        # Inputs
+        artifact_id = "commons-lang3"
+        version = "3.12.0"
+
+        expected_url = [
+            "https://repo1.maven.org/maven2/org/apache/commons/commons-lang3/3.12.0/commons-lang3-3.12.0.pom"
+        ]
+
+        result = resolve.construct_pom_url_from_filename(artifact_id, version)
+
+        self.assertEqual(result, expected_url)
+        mock_get.assert_called_once_with(
+            "https://search.maven.org/solrsearch/select?q=a:commons-lang3&wt=json",
+            timeout=5,
+        )
+        mock_is_maven_pom_url.assert_called_once_with(expected_url[0])
+
+    def test_scanpipe_resolve_get_pom_url_list_with_packages(self):
+        packages = [
+            {
+                "namespace": "org.apache.commons",
+                "name": "commons-lang3",
+                "version": "3.12.0",
+            }
+        ]
+        result = resolve.get_pom_url_list({}, packages)
+        expected = [
+            "https://repo1.maven.org/maven2/org/apache/commons/commons-lang3/3.12.0/commons-lang3-3.12.0.pom"
+        ]
+        self.assertEqual(result, expected)
+
+    def test_scanpipe_resolve_get_pom_url_list_with_maven_download_url(self):
+        input_source = {
+            "download_url": "https://repo1.maven.org/maven2/org/apache/commons/commons-lang3/3.12.0/commons-lang3-3.12.0.jar"
+        }
+        result = resolve.get_pom_url_list(input_source, [])
+        expected = [
+            "https://repo1.maven.org/maven2/org/apache/commons/commons-lang3/3.12.0/commons-lang3-3.12.0.pom"
+        ]
+        self.assertEqual(result, expected)
+
+    @mock.patch("scanpipe.pipes.resolve.construct_pom_url_from_filename")
+    @mock.patch("scanpipe.pipes.resolve.parse_maven_filename")
+    def test_scanpipe_resolve_get_pom_url_list_with_jar_filename(
+        self, mock_parse, mock_construct
+    ):
+        input_source = {"filename": "commons-lang3-3.12.0.jar"}
+        mock_parse.return_value = ("commons-lang3", "3.12.0")
+        mock_construct.return_value = [
+            "https://repo1.maven.org/maven2/org/apache/commons/commons-lang3/3.12.0/commons-lang3-3.12.0.pom"
+        ]
+        result = resolve.get_pom_url_list(input_source, [])
+        self.assertEqual(result, mock_construct.return_value)
+
+    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, [])