Add tests for static dependency resolution

Reference: #1262
Reference: #1263
Reference: #1278
Reference: #1279
Signed-off-by: Ayan Sinha Mahapatra <[email protected]>

scanpipe/tests/pipes/test_scancode.py

@@ -574,7 +574,7 @@ def test_scanpipe_scancode_create_packages_and_dependencies_from_mapping(self):
         pipeline_name = "inspect_packages"
         project1 = Project.objects.create(name="Analysis")
 
-        input_location = self.data / "dependencies" / "resolved_dependencies.zip"
+        input_location = self.data / "dependencies" / "resolved_dependencies_npm.zip"
         project1.copy_input_from(input_location)
 
         run = project1.add_pipeline(
@@ -589,7 +589,7 @@ def test_scanpipe_scancode_create_packages_and_dependencies_from_mapping(self):
         self.assertEqual(7, project1.discovereddependencies.count())
 
         yarn_resource = project1.codebaseresources.get(
-            path="resolved_dependencies.zip-extract/yarn.lock"
+            path="resolved_dependencies_npm.zip-extract/yarn.lock"
         )
         lockfile_package_data = yarn_resource.package_data[0]
         scancode.create_packages_and_dependencies_from_mapping(

scanpipe/tests/test_pipelines.py

@@ -590,15 +590,34 @@ def _normalize_package_uids(self, data):
 
         return data
 
+    def _sort_dependencies(self, data):
+        """
+        Sort dependencies by their "for_package_uid".
+
+        After dependency resolution in some cases we have multiple
+        dependency requirements resolved to a same package, and they
+        are not sorted the same way every time.
+        """
+        mappings = data.get("dependencies")
+        if mappings:
+            mappings_by_uid = {}
+            for mapping in mappings:
+                uid = mapping.get("for_package_uid") or ""
+                mappings_by_uid[uid] = mapping
+            data["dependencies"] = list(dict(sorted(mappings_by_uid.items())).values())
+        return data
+
     def assertPipelineResultEqual(
-        self, expected_file, result_file, regen=FIXTURES_REGEN
+        self, expected_file, result_file, sort_dependencies=False, regen=FIXTURES_REGEN
     ):
         """
         Set `regen` to True to regenerate the expected results.
         """
         result_json = json.loads(Path(result_file).read_text())
         result_json = self._normalize_package_uids(result_json)
         result_data = self._without_keys(result_json, self.exclude_from_diff)
+        if sort_dependencies:
+            result_data = self._sort_dependencies(result_data)
         result_data = sort_for_os_compatibility(result_data)
 
         if regen:
@@ -607,6 +626,8 @@ def assertPipelineResultEqual(
         expected_json = json.loads(expected_file.read_text())
         expected_json = self._normalize_package_uids(expected_json)
         expected_data = self._without_keys(expected_json, self.exclude_from_diff)
+        if sort_dependencies:
+            result_data = self._sort_dependencies(result_data)
         expected_data = sort_for_os_compatibility(expected_data)
 
         self.assertEqual(expected_data, result_data)
@@ -797,11 +818,11 @@ def test_scanpipe_inspect_packages_creates_packages_pypi(self):
         self.assertEqual(0, project1.discoveredpackages.count())
         self.assertEqual(26, project1.discovereddependencies.count())
 
-    def test_scanpipe_inspect_packages_with_resolved_dependencies(self):
+    def test_scanpipe_inspect_packages_with_resolved_dependencies_npm(self):
         pipeline_name = "inspect_packages"
         project1 = Project.objects.create(name="Analysis")
 
-        input_location = self.data / "dependencies" / "resolved_dependencies.zip"
+        input_location = self.data / "dependencies" / "resolved_dependencies_npm.zip"
         project1.copy_input_from(input_location)
 
         run = project1.add_pipeline(
@@ -817,9 +838,122 @@ def test_scanpipe_inspect_packages_with_resolved_dependencies(self):
         self.assertEqual(6, project1.discovereddependencies.count())
 
         result_file = output.to_json(project1)
-        expected_file = self.data / "resolved_dependencies_inspect_packages.json"
+        expected_file = (
+            self.data
+            / "dependencies"
+            / "resolved_dependencies_npm_inspect_packages.json"
+        )
         self.assertPipelineResultEqual(expected_file, result_file)
 
+    def test_scanpipe_inspect_packages_with_resolved_dependencies_poetry(self):
+        pipeline_name = "inspect_packages"
+        project1 = Project.objects.create(name="Analysis")
+
+        input_location = self.data / "dependencies" / "resolved_dependencies_poetry.zip"
+        project1.copy_input_from(input_location)
+
+        run = project1.add_pipeline(
+            pipeline_name=pipeline_name,
+            selected_groups=["Static Resolver"],
+        )
+        pipeline = run.make_pipeline_instance()
+
+        exitcode, out = pipeline.execute()
+        self.assertEqual(0, exitcode, msg=out)
+        self.assertEqual(5, project1.codebaseresources.count())
+        self.assertEqual(6, project1.discoveredpackages.count())
+        self.assertEqual(10, project1.discovereddependencies.count())
+
+        result_file = output.to_json(project1)
+        expected_file = (
+            self.data
+            / "dependencies"
+            / "resolved_dependencies_poetry_inspect_packages.json"
+        )
+        self.assertPipelineResultEqual(expected_file, result_file)
+
+    def test_scanpipe_resolved_dependencies_cocoapods(self):
+        pipeline_name = "resolve_dependencies"
+        project1 = Project.objects.create(name="Analysis")
+
+        input_location = (
+            self.data / "dependencies" / "resolved_dependencies_cocoapods.zip"
+        )
+        project1.copy_input_from(input_location)
+
+        run = project1.add_pipeline(
+            pipeline_name=pipeline_name,
+            selected_groups=["Static Resolver"],
+        )
+        pipeline = run.make_pipeline_instance()
+
+        exitcode, out = pipeline.execute()
+        self.assertEqual(0, exitcode, msg=out)
+        self.assertEqual(3, project1.codebaseresources.count())
+        self.assertEqual(25, project1.discoveredpackages.count())
+        self.assertEqual(30, project1.discovereddependencies.count())
+
+        result_file = output.to_json(project1)
+        expected_file = (
+            self.data / "dependencies" / "resolved_dependencies_cocoapods.json"
+        )
+        self.assertPipelineResultEqual(
+            expected_file, result_file, sort_dependencies=True
+        )
+
+    def test_scanpipe_resolved_dependencies_pip_inspect(self):
+        pipeline_name = "resolve_dependencies"
+        project1 = Project.objects.create(name="Analysis")
+
+        input_location = self.data / "dependencies" / "resolved_dependencies_pip.zip"
+        project1.copy_input_from(input_location)
+
+        run = project1.add_pipeline(
+            pipeline_name=pipeline_name,
+            selected_groups=["Static Resolver"],
+        )
+        pipeline = run.make_pipeline_instance()
+
+        exitcode, out = pipeline.execute()
+        self.assertEqual(0, exitcode, msg=out)
+        self.assertEqual(3, project1.codebaseresources.count())
+        self.assertEqual(4, project1.discoveredpackages.count())
+        self.assertEqual(17, project1.discovereddependencies.count())
+
+        result_file = output.to_json(project1)
+        expected_file = self.data / "dependencies" / "resolved_dependencies_pip.json"
+        self.assertPipelineResultEqual(
+            expected_file,
+            result_file,
+        )
+
+    def test_scanpipe_resolved_dependencies_nuget(self):
+        pipeline_name = "resolve_dependencies"
+        project1 = Project.objects.create(name="Analysis")
+
+        input_location = self.data / "dependencies" / "resolved_dependencies_nuget.zip"
+        project1.copy_input_from(input_location)
+
+        run = project1.add_pipeline(
+            pipeline_name=pipeline_name,
+            selected_groups=["Static Resolver"],
+        )
+        pipeline = run.make_pipeline_instance()
+
+        exitcode, out = pipeline.execute()
+        self.assertEqual(0, exitcode, msg=out)
+        self.assertEqual(3, project1.codebaseresources.count())
+        self.assertEqual(34, project1.discoveredpackages.count())
+        self.assertEqual(108, project1.discovereddependencies.count())
+
+        result_file = output.to_json(project1)
+        expected_file = self.data / "dependencies" / "resolved_dependencies_nuget.json"
+        self.assertPipelineResultEqual(
+            expected_file,
+            result_file,
+            sort_dependencies=True,
+        )
+
     def test_scanpipe_scan_codebase_can_process_wheel(self):
         pipeline_name = "scan_codebase"
         project1 = Project.objects.create(name="Analysis")