Enhance resolve_about_packages to handle filename and checksum values (#656)

- Split the pipes unit tests into their own related submodule

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

Author: tdruez

CHANGELOG.rst

@@ -11,6 +11,10 @@ v33.0.0 (unreleased)
 - Remove the admin app and views.
   https://github.com/nexB/scancode.io/issues/645
 
+- Enhance the ``resolve_about_packages`` pipe to handle filename and checksum values.
+
+- Split the pipes unit tests into their own related submodule.
+
 v32.1.0 (2023-03-23)
 --------------------
 

scanpipe/pipes/resolve.py

@@ -38,7 +38,7 @@
 from scanpipe.models import DiscoveredPackage
 
 """
-Utilities to resolve packages from manifest, lockfile, and SBOM.
+Resolve packages from manifest, lockfile, and SBOM.
 """
 
 
@@ -61,14 +61,22 @@ def resolve_about_packages(input_location):
     """Resolve the packages from the `input_location` .ABOUT file."""
     about = About(location=input_location)
     about_data = about.as_dict()
+    package_data = about_data.copy()
 
     if package_url := about_data.get("package_url"):
         package_url_data = PackageURL.from_string(package_url).to_dict(encode=True)
         for field_name, value in package_url_data.items():
             if value:
-                about_data[field_name] = value
+                package_data[field_name] = value
 
-    package_data = DiscoveredPackage.clean_data(about_data)
+    if about_resource := about_data.get("about_resource"):
+        package_data["filename"] = list(about_resource.keys())[0]
+
+    for field_name, value in about_data.items():
+        if field_name.startswith("checksum_"):
+            package_data[field_name.replace("checksum_", "")] = value
+
+    package_data = DiscoveredPackage.clean_data(package_data)
     return [package_data]
 
 
@@ -177,11 +185,15 @@ def get_default_package_type(input_location):
     Return the package type associated with the provided `input_location`.
     This type is used to get the related handler that knows how process the input.
     """
+    input_location = str(input_location)
+
     for handler in APPLICATION_PACKAGE_DATAFILE_HANDLERS:
         if handler.is_datafile(input_location):
             return handler.default_package_type
+
         if input_location.endswith((".spdx", ".spdx.json")):
             return "spdx"
+
         if input_location.endswith((".bom.json", ".cdx.json")):
             return "cyclonedx"
 

scanpipe/tests/data/Django-4.0.8-py3-none-any.whl.ABOUT

@@ -1,7 +1,7 @@
 about_resource: Django-4.0.8-py3-none-any.whl
 name: django
 version: 4.0.8
-download_url: https://files.pythonhosted.org/packages/e1/d0/d90528978da16288d470bb423abad307ed7ae724090132ff6bf67d6a5579/Django-4.0.8-py3-none-any.whl
+download_url: https://python.org/Django-4.0.8-py3-none-any.whl
 package_url: pkg:pypi/django@4.0.8
 license_expression: bsd-new
 attribute: yes

scanpipe/tests/data/poor_values.ABOUT

@@ -0,0 +1,3 @@
+name: project
+version:
+license_expression:

scanpipe/tests/pipes/__init__.py

@@ -0,0 +1,162 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/nexB/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# 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 json
+import sys
+from pathlib import Path
+from unittest import skipIf
+
+from django.core.management import call_command
+from django.test import TestCase
+
+from scanpipe.models import Project
+from scanpipe.pipes import codebase
+from scanpipe.pipes import scancode
+
+
+class ScanPipeCodebasePipesTest(TestCase):
+    data_location = Path(__file__).parent.parent / "data"
+
+    def test_scanpipe_pipes_codebase_get_codebase_tree(self):
+        def _replace_path(virtual_tree_children):
+            """
+            Given a list `virtual_tree_children` of mappings, remove instances
+            of "virtual_root/" from the paths of mappings and their children,
+            recursively.
+            """
+            for res in virtual_tree_children:
+                path = res["path"]
+                path = path.replace("virtual_root/", "")
+                res["path"] = path
+                _replace_path(res.get("children", []))
+
+        fixtures = self.data_location / "asgiref-3.3.0_fixtures.json"
+        call_command("loaddata", fixtures, **{"verbosity": 0})
+        project = Project.objects.get(name="asgiref")
+
+        scan_results = self.data_location / "asgiref-3.3.0_scanpipe_output.json"
+        virtual_codebase = scancode.get_virtual_codebase(project, scan_results)
+        project_codebase = codebase.ProjectCodebase(project)
+
+        fields = ["name", "path"]
+
+        virtual_tree = codebase.get_codebase_tree(virtual_codebase, fields)
+        project_tree = codebase.get_codebase_tree(project_codebase, fields)
+
+        with open(self.data_location / "asgiref-3.3.0_tree.json") as f:
+            expected = json.loads(f.read())
+
+        self.assertEqual(expected, project_tree)
+
+        virtual_tree_children = virtual_tree["children"][0]["children"]
+        _replace_path(virtual_tree_children)
+
+        self.assertEqual(expected["children"], virtual_tree_children)
+
+    def test_scanpipe_pipes_codebase_project_codebase_class_no_resources(self):
+        project = Project.objects.create(name="project")
+        project_codebase = codebase.ProjectCodebase(project)
+
+        self.assertEqual([], list(project_codebase.root_resources))
+        self.assertEqual([], list(project_codebase.resources))
+        self.assertEqual([], list(project_codebase.walk()))
+        self.assertEqual(dict(children=[]), project_codebase.get_tree())
+
+    def test_scanpipe_pipes_codebase_project_codebase_class_with_resources(self):
+        fixtures = self.data_location / "asgiref-3.3.0_fixtures.json"
+        call_command("loaddata", fixtures, **{"verbosity": 0})
+
+        project = Project.objects.get(name="asgiref")
+        project_codebase = codebase.ProjectCodebase(project)
+
+        expected_root_resources = project.codebaseresources.exclude(path__contains="/")
+        expected_root_resources = list(expected_root_resources)
+        self.assertEqual(expected_root_resources, list(project_codebase.root_resources))
+
+        self.assertEqual(18, len(project_codebase.resources))
+
+        walk_gen = project_codebase.walk()
+        self.assertEqual(next(iter(expected_root_resources)), next(walk_gen))
+        expected = "asgiref-3.3.0-py3-none-any.whl-extract"
+        self.assertEqual(expected, next(walk_gen).path)
+
+        tree = project_codebase.get_tree()
+        with open(self.data_location / "asgiref-3.3.0_tree.json") as f:
+            expected = json.loads(f.read())
+
+        self.assertEqual(expected, tree)
+
+    @skipIf(sys.platform != "linux", "Ordering differs on macOS.")
+    def test_scanpipe_pipes_codebase_project_codebase_class_walk(self):
+        fixtures = self.data_location / "asgiref-3.3.0_fixtures.json"
+        call_command("loaddata", fixtures, **{"verbosity": 0})
+
+        project = Project.objects.get(name="asgiref")
+        project_codebase = codebase.ProjectCodebase(project)
+
+        topdown_paths = list(r.path for r in project_codebase.walk(topdown=True))
+        expected_topdown_paths = [
+            "asgiref-3.3.0-py3-none-any.whl",
+            "asgiref-3.3.0-py3-none-any.whl-extract",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/compatibility.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/current_thread_executor.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/__init__.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/local.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/server.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/sync.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/testing.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/timeout.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/wsgi.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info/LICENSE",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info/METADATA",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info/RECORD",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info/"
+            "top_level.txt",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info/WHEEL",
+        ]
+        self.assertEqual(expected_topdown_paths, topdown_paths)
+
+        bottom_up_paths = list(r.path for r in project_codebase.walk(topdown=False))
+        expected_bottom_up_paths = [
+            "asgiref-3.3.0-py3-none-any.whl",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/compatibility.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/current_thread_executor.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/__init__.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/local.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/server.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/sync.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/testing.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/timeout.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref/wsgi.py",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info/LICENSE",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info/METADATA",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info/RECORD",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info/"
+            "top_level.txt",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info/WHEEL",
+            "asgiref-3.3.0-py3-none-any.whl-extract/asgiref-3.3.0.dist-info",
+            "asgiref-3.3.0-py3-none-any.whl-extract",
+        ]
+        self.assertEqual(expected_bottom_up_paths, bottom_up_paths)

scanpipe/tests/pipes/test_codebase.py

@@ -37,7 +37,7 @@
 
 
 class ScanPipeDockerPipesTest(TestCase):
-    data_path = Path(__file__).parent / "data"
+    data_path = Path(__file__).parent.parent / "data"
     maxDiff = None
 
     def assertResultsEqual(self, expected_file, results, regen=FIXTURES_REGEN):

scanpipe/tests/test_pipes_docker.py scanpipe/tests/pipes/test_docker.pyscanpipe/tests/test_pipes_docker.py renamed to scanpipe/tests/pipes/test_docker.py

@@ -0,0 +1,102 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# http://nexb.com and https://github.com/nexB/scancode.io
+# The ScanCode.io software is licensed under the Apache License version 2.0.
+# Data generated with ScanCode.io is provided as-is without warranties.
+# ScanCode is a trademark of nexB Inc.
+#
+# You may not use this software except in compliance with the License.
+# You may obtain a copy of the License at: http://apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software distributed
+# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+# CONDITIONS OF ANY KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations under the License.
+#
+# Data Generated with ScanCode.io is provided on an "AS IS" BASIS, WITHOUT WARRANTIES
+# OR CONDITIONS OF ANY KIND, either express or implied. No content created from
+# ScanCode.io should be considered or used as legal advice. Consult an Attorney
+# for any legal advice.
+#
+# 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.
+
+from pathlib import Path
+from unittest import mock
+
+from django.test import TestCase
+
+from scanpipe.pipes import fetch
+
+
+class ScanPipeFetchPipesTest(TestCase):
+    data_location = Path(__file__).parent.parent / "data"
+
+    @mock.patch("requests.get")
+    def test_scanpipe_pipes_fetch_http(self, mock_get):
+        url = "https://example.com/filename.zip"
+
+        mock_get.return_value = mock.Mock(
+            content=b"\x00", headers={}, status_code=200, url=url
+        )
+        downloaded_file = fetch.fetch_http(url)
+        self.assertTrue(Path(downloaded_file.directory, "filename.zip").exists())
+
+        redirect_url = "https://example.com/redirect.zip"
+        mock_get.return_value = mock.Mock(
+            content=b"\x00", headers={}, status_code=200, url=redirect_url
+        )
+        downloaded_file = fetch.fetch_http(url)
+        self.assertTrue(Path(downloaded_file.directory, "redirect.zip").exists())
+
+        headers = {
+            "content-disposition": 'attachment; filename="another_name.zip"',
+        }
+        mock_get.return_value = mock.Mock(
+            content=b"\x00", headers=headers, status_code=200, url=url
+        )
+        downloaded_file = fetch.fetch_http(url)
+        self.assertTrue(Path(downloaded_file.directory, "another_name.zip").exists())
+
+    @mock.patch("scanpipe.pipes.fetch.get_docker_image_platform")
+    @mock.patch("scanpipe.pipes.fetch._get_skopeo_location")
+    @mock.patch("scanpipe.pipes.run_command")
+    def test_scanpipe_pipes_fetch_docker_image(
+        self, mock_run_command, mock_skopeo, mock_platform
+    ):
+        url = "docker://debian:10.9"
+
+        mock_platform.return_value = "linux", "amd64", ""
+        mock_skopeo.return_value = "skopeo"
+        mock_run_command.return_value = 1, "error"
+
+        with self.assertRaises(fetch.FetchDockerImageError):
+            fetch.fetch_docker_image(url)
+
+        mock_run_command.assert_called_once()
+        cmd = mock_run_command.call_args[0][0]
+        self.assertTrue(cmd.startswith("skopeo copy --insecure-policy"))
+        self.assertIn("docker://debian:10.9 docker-archive:/", cmd)
+        self.assertIn("--override-os=linux --override-arch=amd64", cmd)
+        self.assertTrue(cmd.endswith("debian_10_9.tar"))
+
+    @mock.patch("requests.get")
+    def test_scanpipe_pipes_fetch_fetch_urls(self, mock_get):
+        urls = [
+            "https://example.com/filename.zip",
+            "https://example.com/archive.tar.gz",
+        ]
+
+        mock_get.return_value = mock.Mock(
+            content=b"\x00", headers={}, status_code=200, url="mocked_url"
+        )
+        downloads, errors = fetch.fetch_urls(urls)
+        self.assertEqual(2, len(downloads))
+        self.assertEqual(urls[0], downloads[0].uri)
+        self.assertEqual(urls[1], downloads[1].uri)
+        self.assertEqual(0, len(errors))
+
+        mock_get.side_effect = Exception
+        downloads, errors = fetch.fetch_urls(urls)
+        self.assertEqual(0, len(downloads))
+        self.assertEqual(2, len(errors))
+        self.assertEqual(urls, errors)