Add unit test for InspectManifest #284

Signed-off-by: Thomas Druez <[email protected]>

Author: tdruez

scanpipe/pipelines/inspect_manifest.py

@@ -29,15 +29,16 @@
 
 def resolve_pypi_packages(input_location):
     """
-    https://github.com/nexB/scancode-toolkit/blob/develop/requirements.txt
-    https://raw.githubusercontent.com/nexB/python-inspector/main/requirements.txt
+    Resolve the PyPI packages from the `input_location` requirements file.
     """
-    inspector_output = resolver_api(requirement_files=[input_location])
-    resolved_packages = inspector_output.packages
-    return resolved_packages
+    inspector_output = resolver_api(
+        requirement_files=[input_location],
+        prefer_source=True,
+    )
+    return inspector_output.packages
 
 
-# `default_package_type`: resolver callable
+# Mapping between the `default_package_type` its related resolver function
 resolver_registry = {
     "pypi": resolve_pypi_packages,
 }
@@ -52,6 +53,8 @@ def get_default_package_type(input_location):
 class InspectManifest(Pipeline):
     """
     A pipeline to inspect one or more manifest files and resolve its packages.
+
+    Only PyPI requirements file are supported.
     """
 
     @classmethod
@@ -75,6 +78,8 @@ def create_packages_from_manifest(self):
         """
         for input_location in self.input_locations:
             default_package_type = get_default_package_type(input_location)
+            if not default_package_type:
+                raise Exception(f"No package type found for {input_location}")
 
             resolver = resolver_registry.get(default_package_type)
             if not resolver:

scanpipe/tests/test_pipelines.py

@@ -624,3 +624,38 @@ def test_scanpipe_check_vulnerabilities_pipeline_integration_test(
         package1.refresh_from_db()
         expected = {"discovered_vulnerabilities": vulnerability_data}
         self.assertEqual(expected, package1.extra_data)
+
+    @mock.patch("scanpipe.pipelines.inspect_manifest.resolver_api")
+    def test_scanpipe_inspect_manifest_pipeline_integration_test(self, resolver_api):
+        resolver_api.return_value = mock.Mock(packages=[])
+
+        pipeline_name = "inspect_manifest"
+        project1 = Project.objects.create(name="Analysis")
+
+        run = project1.add_pipeline(pipeline_name)
+        pipeline = run.make_pipeline_instance()
+
+        project1.move_input_from(tempfile.mkstemp()[1])
+        exitcode, out = pipeline.execute()
+        self.assertEqual(1, exitcode, msg=out)
+        self.assertIn("No package type found for", out)
+
+        project1.reset(keep_input=False)
+        run = project1.add_pipeline(pipeline_name)
+        pipeline = run.make_pipeline_instance()
+
+        project1.move_input_from(tempfile.mkstemp(suffix="requirements.txt")[1])
+        exitcode, out = pipeline.execute()
+        self.assertEqual(1, exitcode, msg=out)
+        self.assertIn("No packages could be resolved.", out)
+
+        resolver_api.return_value = mock.Mock(packages=[package_data1])
+        exitcode, out = pipeline.execute()
+        self.assertEqual(0, exitcode, msg=out)
+
+        self.assertEqual(1, project1.discoveredpackages.count())
+        discoveredpackage = project1.discoveredpackages.get()
+        exclude_fields = ["qualifiers", "release_date", "size"]
+        for field_name, value in package_data1.items():
+            if value and field_name not in exclude_fields:
+                self.assertEqual(value, getattr(discoveredpackage, field_name))

setup.cfg

@@ -76,7 +76,7 @@ install_requires =
     # FetchCode
     fetchcode-container==1.2.3.210512; sys_platform == "linux"
     # Python-inspector
-    python-inspector==0.9.0
+    python-inspector==0.9.1
     # Utilities
     XlsxWriter==3.0.3
     requests==2.28.1