Refine the extraction errors reporting include the resource path #1273 (#1276)

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

Author: tdruez

CHANGELOG.rst

@@ -7,6 +7,10 @@ v34.6.4 (unreleased)
 - Add all "classify" plugin fields from scancode-toolkit on the CodebaseResource model.
   https://github.com/nexB/scancode.io/issues/1275
 
+- Refine the extraction errors reporting to include the resource path for rendering
+  link to the related resources in the UI.
+  https://github.com/nexB/scancode.io/issues/1273
+
 - Add a ``flush-projects`` management command, to Delete all project data and their
   related work directories created more than a specified number of days ago.
   https://github.com/nexB/scancode.io/issues/1289

scanpipe/pipelines/__init__.py

@@ -25,6 +25,7 @@
 import traceback
 from contextlib import contextmanager
 from functools import wraps
+from pathlib import Path
 from pydoc import getdoc
 from pydoc import splitdoc
 from timeit import default_timer as timer
@@ -301,17 +302,45 @@ def flag_ignored_resources(self):
         if ignored_patterns := self.env.get("ignored_patterns"):
             flag.flag_ignored_patterns(self.project, patterns=ignored_patterns)
 
-    def extract_archives(self):
+    def extract_archive(self, location, target):
+        """Extract archive at `location` to `target`. Save errors as messages."""
+        from scanpipe.pipes import scancode
+
+        extract_errors = scancode.extract_archive(location, target)
+
+        for resource_location, errors in extract_errors.items():
+            resource_path = Path(resource_location)
+
+            if resource_path.is_relative_to(self.project.codebase_path):
+                resource_path = resource_path.relative_to(self.project.codebase_path)
+                details = {"resource_path": str(resource_path)}
+            elif resource_path.is_relative_to(self.project.input_path):
+                resource_path = resource_path.relative_to(self.project.input_path)
+                details = {"path": f"input/{str(resource_path)}"}
+            else:
+                details = {"filename": str(resource_path.name)}
+
+            self.project.add_error(
+                description="\n".join(errors),
+                model="extract_archive",
+                details=details,
+            )
+
+    def extract_archives(self, location=None):
         """Extract archives located in the codebase/ directory with extractcode."""
         from scanpipe.pipes import scancode
 
-        extract_errors = scancode.extract_archives(
-            location=self.project.codebase_path,
-            recurse=True,
-        )
+        if not location:
+            location = self.project.codebase_path
 
-        if extract_errors:
-            self.add_error("\n".join(extract_errors))
+        extract_errors = scancode.extract_archives(location=location, recurse=True)
+
+        for resource_path, errors in extract_errors.items():
+            self.project.add_error(
+                description="\n".join(errors),
+                model="extract_archives",
+                details={"resource_path": resource_path},
+            )
 
         # Reload the project env post-extraction as the scancode-config.yml file
         # may be located in one of the extracted archives.

scanpipe/pipelines/deploy_to_develop.py

@@ -127,13 +127,9 @@ def extract_inputs_to_codebase_directory(self):
             (self.to_files, self.project.codebase_path / d2d.TO),
         ]
 
-        errors = []
         for input_files, codebase_path in inputs_with_codebase_path_destination:
             for input_file_path in input_files:
-                errors += scancode.extract_archive(input_file_path, codebase_path)
-
-        if errors:
-            self.add_error("\n".join(errors))
+                self.extract_archive(input_file_path, codebase_path)
 
         # Reload the project env post-extraction as the scancode-config.yml file
         # may be located in one of the extracted archives.

scanpipe/pipelines/root_filesystem.py

@@ -53,15 +53,10 @@ def extract_input_files_to_codebase_directory(self):
         """Extract root filesystem input archives with extractcode."""
         input_files = self.project.inputs("*")
         target_path = self.project.codebase_path
-        errors = []
 
         for input_file in input_files:
             extract_target = target_path / f"{input_file.name}{EXTRACT_SUFFIX}"
-            extract_errors = scancode.extract_archive(input_file, extract_target)
-            errors.extend(extract_errors)
-
-        if errors:
-            self.add_error("\n".join(errors))
+            self.extract_archive(input_file, extract_target)
 
         # Reload the project env post-extraction as the scancode-config.yml file
         # may be located in one of the extracted archives.

scanpipe/pipelines/scan_single_package.py

@@ -93,13 +93,7 @@ def extract_input_to_codebase_directory(self):
             copy_input(self.input_path, self.project.codebase_path)
             return
 
-        extract_errors = scancode.extract_archive(
-            location=self.input_path,
-            target=self.project.codebase_path,
-        )
-
-        if extract_errors:
-            self.add_error("\n".join(extract_errors))
+        self.extract_archive(self.input_path, self.project.codebase_path)
 
         # Reload the project env post-extraction as the scancode-config.yml file
         # may be located in one of the extracted archives.

scanpipe/pipes/scancode.py

@@ -102,15 +102,15 @@ def extract_archive(location, target):
     Extract a single archive or compressed file at `location` to the `target`
     directory.
 
-    Return a list of extraction errors.
+    Return a dict of extraction errors, keyed by the resource location.
 
     Wrapper of the `extractcode.api.extract_archive` function.
     """
-    errors = []
+    errors = {}
 
     for event in extractcode_api.extract_archive(location, target):
-        if event.done:
-            errors.extend(event.errors)
+        if event.done and event.errors:
+            errors[str(event.source)] = event.errors
 
     return errors
 
@@ -125,7 +125,7 @@ def extract_archives(location, recurse=False):
 
     If `recurse` is True, extract nested archives-in-archives recursively.
 
-    Return a list of extraction errors.
+    Return a dict of extraction errors, keyed by the resource location.
 
     Wrapper of the `extractcode.api.extract_archives` function.
     """
@@ -135,10 +135,10 @@ def extract_archives(location, recurse=False):
         "all_formats": True,
     }
 
-    errors = []
+    errors = {}
     for event in extractcode_api.extract_archives(location, **options):
-        if event.done:
-            errors.extend(event.errors)
+        if event.done and event.errors:
+            errors[str(event.source)] = event.errors
 
     return errors
 

scanpipe/tests/data/scancode/corrupted.tar.gz

@@ -59,7 +59,7 @@ def test_scanpipe_pipes_scancode_extract_archive(self):
         input_location = str(self.data_location / "archive.zip")
 
         errors = scancode.extract_archive(input_location, target)
-        self.assertEqual([], errors)
+        self.assertEqual({}, errors)
 
         results = [path.name for path in list(Path(target).glob("**/*"))]
         expected = [
@@ -72,13 +72,25 @@ def test_scanpipe_pipes_scancode_extract_archive(self):
         for path in expected:
             self.assertIn(path, results)
 
+    def test_scanpipe_pipes_scancode_extract_archive_errors(self):
+        target = tempfile.mkdtemp()
+        input_location = str(self.data_location / "scancode" / "corrupted.tar.gz")
+        errors = scancode.extract_archive(input_location, target)
+
+        error_message = "gzip decompression failed"
+        if sys.platform == "darwin":
+            error_message += " (zlib returned error -3, msg invalid code lengths set)"
+
+        expected = {input_location: [error_message]}
+        self.assertEqual(expected, errors)
+
     def test_scanpipe_pipes_scancode_extract_archives(self):
         tempdir = Path(tempfile.mkdtemp())
         input_location = str(self.data_location / "archive.zip")
         copy_input(input_location, tempdir)
 
         errors = scancode.extract_archives(tempdir)
-        self.assertEqual([], errors)
+        self.assertEqual({}, errors)
 
         results = [path.name for path in list(tempdir.glob("**/*"))]
         self.assertEqual(9, len(results))
@@ -93,6 +105,19 @@ def test_scanpipe_pipes_scancode_extract_archives(self):
         for path in expected:
             self.assertIn(path, results)
 
+    def test_scanpipe_pipes_scancode_extract_archives_errors(self):
+        tempdir = Path(tempfile.mkdtemp())
+        input_location = str(self.data_location / "scancode" / "corrupted.tar.gz")
+        target = copy_input(input_location, tempdir)
+        errors = scancode.extract_archives(tempdir)
+
+        error_message = "gzip decompression failed"
+        if sys.platform == "darwin":
+            error_message += " (zlib returned error -3, msg invalid code lengths set)"
+
+        expected = {str(target): [error_message]}
+        self.assertEqual(expected, errors)
+
     @skipIf(sys.platform != "linux", "QCOW2 extraction is not available on macOS.")
     def test_scanpipe_pipes_scancode_extract_archive_vmimage_qcow2(self):
         target = tempfile.mkdtemp()
@@ -104,7 +129,7 @@ def test_scanpipe_pipes_scancode_extract_archive_vmimage_qcow2(self):
 
         # The VM image extraction features are available in the Docker image context.
         if from_docker_image:
-            self.assertEqual([], errors)
+            self.assertEqual({}, errors)
             results = [path.name for path in list(Path(target).glob("**/*"))]
             expected = [
                 "bin",
@@ -118,15 +143,17 @@ def test_scanpipe_pipes_scancode_extract_archive_vmimage_qcow2(self):
             self.assertEqual(sorted(expected), sorted(results))
 
         else:
-            error = errors[0]
-            self.assertTrue(
-                any(
-                    [
-                        "Unable to read kernel" in error,
-                        "VM Image extraction only supported on Linux." in error,
-                    ]
-                )
-            )
+            expected = {
+                str(input_location): [
+                    "Unable to read kernel at: /boot/vmlinuz-6.5.0-1022-azure.\n"
+                    "libguestfs requires the kernel executable to be readable.\n"
+                    "This is the case by default on most Linux distributions except on "
+                    "Ubuntu.\nPlease follow the ExtractCode installation instructions "
+                    "in the README.rst at:\n"
+                    "https://github.com/nexB/extractcode/blob/main/README.rst '\n"
+                ]
+            }
+            self.assertEqual(expected, errors)
 
     def test_scanpipe_pipes_scancode_get_resource_info(self):
         input_location = str(self.data_location / "notice.NOTICE")