Add the ability to read user provided debug files

Closes: SET-941

- refactoring Hotkdump run method
- Adding unit test, fixes, optimizing maybe_* methods

README.md

@@ -29,8 +29,8 @@ hotkdump [-d/--dump-file-path] <crash-dump-file> [other-options...]
 other-options:
 
 ```bash
-usage: main.py [-h] -d DUMP_FILE_PATH [-c INTERNAL_CASE_NUMBER] [-i] [-o OUTPUT_FILE_PATH] [-l LOG_FILE_PATH] [-p DDEBS_FOLDER_PATH] [--print-vmcoreinfo-fields [PRINT_VMCOREINFO_FIELDS ...]]
-               [--no-debuginfod | --no-pullpkg]
+usage: hotkdump [-h] -d DUMP_FILE_PATH [-c INTERNAL_CASE_NUMBER] [-i] [-o OUTPUT_FILE_PATH] [-l LOG_FILE_PATH] [-p DDEBS_FOLDER_PATH]
+                [--print-vmcoreinfo-fields [PRINT_VMCOREINFO_FIELDS ...]] [--debug-file DEBUG_FILE] [--no-debuginfod | --no-pullpkg]
 
 options:
   -h, --help            show this help message and exit
@@ -47,6 +47,8 @@ options:
                         Path to save the downloaded .ddeb files. Will be created if the specified path is absent. (default: None)
   --print-vmcoreinfo-fields [PRINT_VMCOREINFO_FIELDS ...]
                         Read and print the specified VMCOREINFO fields from the given kernel crash dump, then exit.
+  --debug-file DEBUG_FILE
+                        Specify the debug file to use. Only ddebs and vmlinux files are supported. (default: None)
   --no-debuginfod       Do not use debuginfod for downloads (default: False)
   --no-pullpkg          Do not use pullpkg for downloads (default: False)
 ```

hotkdump/core/hotkdump.py

@@ -69,15 +69,16 @@ class HotkdumpParameters:
         )
     )
     print_vmcoreinfo_fields: list = None
+    debug_file: str = None
     no_debuginfod: bool = False
     no_pullpkg: bool = False
 
     def validate_sanity(self):
         """Check whether option values are not contradicting and sane."""
 
-        if all([self.no_debuginfod, self.no_pullpkg]):
+        if all([self.no_debuginfod, self.no_pullpkg, not self.debug_file]):
             raise ExceptionWithLog(
-                "At least one download method must be enabled (debuginfod, pullpkg)!"
+                "At least one download method must be enabled (debuginfod, pullpkg) or a debug file must be specified!"
             )
 
         self.ddeb_retention_settings.validate_sanity()
@@ -341,6 +342,8 @@ def maybe_download_vmlinux_via_debuginfod(self):
         """Try downloading vmlinux image with debug information
         using debuginfod-find."""
         try:
+            if self.params.no_debuginfod:
+                return None
             debuginfod_find_path = self.find_debuginfod_find_executable()
             if not debuginfod_find_path:
                 logging.debug("debuginfod-find is not present in environment.")
@@ -394,6 +397,8 @@ def maybe_download_vmlinux_via_pullpkg(self):
         Returns:
             str: The path to the .ddeb file
         """
+        if self.params.no_pullpkg:
+            return None
         # Parameters are: release, release{without -generic}, normalized version, arch
         # linux-image-unsigned-5.4.0-135-generic-dbgsym_5.4.0-135.152_amd64.ddeb
         ddeb_name_format = "linux-image-unsigned-{}-dbgsym_{}.{}_{}.ddeb"
@@ -414,7 +419,11 @@ def maybe_download_vmlinux_via_pullpkg(self):
                 )
                 # Ensure that the file's last access time is updated
                 os.utime(expected_ddeb_path, (time.time(), time.time()))
-                return expected_ddeb_path
+                extracted_vmlinux = self.extract_vmlinux_ddeb(expected_ddeb_path)
+                if extracted_vmlinux:
+                    return extracted_vmlinux
+                logging.error("Failed to extract ddeb file")
+                return None
 
             logging.info(
                 "Downloading `vmlinux` image for kernel version %s, please be patient...",
@@ -440,8 +449,11 @@ def maybe_download_vmlinux_via_pullpkg(self):
             PullPkg().pull(pull_args)
             if not os.path.exists(expected_ddeb_path):
                 raise ExceptionWithLog(f"failed to download {expected_ddeb_path}")
-
-        return expected_ddeb_path
+        extracted_vmlinux = self.extract_vmlinux_ddeb(expected_ddeb_path)
+        if extracted_vmlinux:
+            return extracted_vmlinux
+        logging.error("Failed to extract ddeb file.")
+        return None
 
     def extract_vmlinux_ddeb(self, ddeb_file):
         """Extract the given vmlinux ddeb file to temp_working_dir/ddeb-root
@@ -498,6 +510,36 @@ def launch_crash(self, vmlinux_path: str):
             self.crash_executable, f"-x {self.params.dump_file_path} {vmlinux_path}"
         )
 
+    DBG_DDEB = ".ddeb"
+    DBG_VMLINUX = "vmlinux"
+
+    def debug_file_type(self):
+        """Return the current file type by checking the extension"""
+        if self.params.debug_file:
+            filename = os.path.basename(self.params.debug_file).split("/")[-1]
+            if filename.endswith(self.DBG_DDEB):
+                return self.DBG_DDEB
+            if filename.startswith(self.DBG_VMLINUX):
+                return self.DBG_VMLINUX
+        return None
+
+    def maybe_get_user_specified_vmlinux(self):
+        """return the vmlinux file path from the user specified debug file"""
+        if not self.params.debug_file:
+            return None
+
+        dbg_type = self.debug_file_type()
+        vmlinux_ddeb = self.params.debug_file if dbg_type == self.DBG_DDEB else None
+        extracted_vmlinux = (
+            self.params.debug_file if dbg_type == self.DBG_VMLINUX else None
+        )
+        if vmlinux_ddeb:
+            extracted_vmlinux = self.extract_vmlinux_ddeb(vmlinux_ddeb)
+        if extracted_vmlinux:
+            return extracted_vmlinux
+        logging.error("Failed to retrieve vmlinux file")
+        return None
+
     def run(self):
         """Run hotkdump main routine."""
         try:
@@ -506,19 +548,11 @@ def run(self):
                     print(f"{key}={self.kdump_file.vmcoreinfo.get(key)}")
                 return
 
-            extracted_vmlinux = None
-
-            if not self.params.no_debuginfod:
-                extracted_vmlinux = self.maybe_download_vmlinux_via_debuginfod()
-
-            if not extracted_vmlinux and not self.params.no_pullpkg:
-                vmlinux_ddeb = self.maybe_download_vmlinux_via_pullpkg()
-                if vmlinux_ddeb == "":
-                    logging.error("vmlinux ddeb dowload failed.")
-                    return
-
-                extracted_vmlinux = self.extract_vmlinux_ddeb(vmlinux_ddeb)
-
+            extracted_vmlinux = (
+                self.maybe_get_user_specified_vmlinux()
+                or self.maybe_download_vmlinux_via_debuginfod()
+                or self.maybe_download_vmlinux_via_pullpkg()
+            )
             if not extracted_vmlinux:
                 logging.error("vmlinux image with debug symbols not found, aborting")
                 return

hotkdump/main.py

@@ -69,6 +69,15 @@ def main():
         nargs="*",
         default=argparse.SUPPRESS,
     )
+    ap.add_argument(
+        "--debug-file",
+        required=False,
+        help=(
+            "Specify the debug file to use. Only ddebs and vmlinux files are supported. "
+            "Disables downloads (--no-debuginfod and --no-pullpkg)"
+        ),
+        default=None,
+    )
     download_methods_group = ap.add_mutually_exclusive_group()
     download_methods_group.add_argument(
         "--no-debuginfod",
@@ -88,6 +97,9 @@ def main():
     args = {k: v for k, v in vars(ap.parse_args()).items() if v is not None}
     params = HotkdumpParameters(**args)
 
+    if params.debug_file:
+        params.no_debuginfod = params.no_pullpkg = True
+
     try:
         hotkdump = Hotkdump(parameters=params)
         hotkdump.run()

tests/test_hotkdump.py

@@ -304,23 +304,23 @@ def test_strip_tags(self):
                     ExceptionWithLog, uut.strip_release_variant_tags, input_str
                 )
 
+    @mock.patch("hotkdump.core.hotkdump.Hotkdump.extract_vmlinux_ddeb")
     @mock.patch("os.utime")
     @mock.patch("hotkdump.core.hotkdump.PullPkg")
     @mock.patch("hotkdump.core.hotkdump.switch_cwd")
-    def test_maybe_download_vmlinux_ddeb(
-        self, mock_switch_cwd, mock_pullpkg, mock_utime
-    ):
+    @mock.patch("subprocess.Popen")
+    def test_maybe_download_vmlinux_ddeb(self, *args):
         """Verify that the hotkdump:
         - calls the PullPkg when the ddeb is absent
         - does not call the PullPkg when the ddeb is present
         - raises an ExceptionWithLog when PullPkg fails
         """
         # Set up mock return values
         mock_pull = mock.MagicMock()
-        mock_pullpkg.return_value.pull = mock_pull
+        args[2].return_value.pull = mock_pull
 
         switch_cwd = mock.MagicMock()
-        mock_switch_cwd.return_value = switch_cwd
+        args[1].return_value = switch_cwd
 
         # mock_pull.return_value.pull
         params = HotkdumpParameters(dump_file_path="empty")
@@ -337,9 +337,12 @@ def test_maybe_download_vmlinux_ddeb(
         expected_ddeb_path = (
             "linux-image-unsigned-5.15.0-1030-gcp-dbgsym_5.15.0-1030.37_amd64.ddeb"
         )
+        expected_vmlinux_path = "/tmp/path/to/vmlinux/file/vmlinux-yy.xx"
+        args[4].return_value = expected_vmlinux_path
 
         with mock.patch("os.path.exists") as mock_exists:
             mock_exists.side_effect = [False, True]
+            args[0].return_value.__enter__.return_value.returncode = 0
             result = uut.maybe_download_vmlinux_via_pullpkg()
 
             mock_exists.assert_called()
@@ -359,7 +362,7 @@ def test_maybe_download_vmlinux_ddeb(
             )
 
             # Assert that the expected ddeb file path was returned
-            self.assertEqual(result, expected_ddeb_path)
+            self.assertEqual(result, expected_vmlinux_path)
 
         mock_pull.reset_mock()
         # Test reusing an existing ddeb file
@@ -374,12 +377,12 @@ def test_maybe_download_vmlinux_ddeb(
                 mock_pull.assert_not_called()
 
                 # # Assert that the file's last access time was updated
-                mock_utime.assert_called_once_with(
+                args[3].assert_called_once_with(
                     expected_ddeb_path, (1234567890.0, 1234567890.0)
                 )
 
                 # Assert that the expected ddeb file path was returned
-                self.assertEqual(result, expected_ddeb_path)
+                self.assertEqual(result, expected_vmlinux_path)
 
         # Test failing to download a new ddeb file
         mock_pull.return_value = Exception("Error")
@@ -615,3 +618,44 @@ def test_post_run_ddeb_retention_disabled(
                 mock.call("/path/to/ddebs/file4.ddeb"),
             ]
         mock_remove.assert_has_calls(expected_calls, any_order=True)
+
+    def test_debug_file_type(self):
+        """Verify that the file type is correctly inferred"""
+        params = HotkdumpParameters(
+            debug_file="/path/to/a/ddeb/linux-yy.xx.ddeb", dump_file_path="empty"
+        )
+        hkdump = Hotkdump(params)
+        self.assertEqual(hkdump.debug_file_type(), ".ddeb")
+
+        hkdump.params.debug_file = "/path/to/a/vmlinux/vmlinux-yy.xx"
+        self.assertEqual(hkdump.debug_file_type(), "vmlinux")
+
+        hkdump.params.debug_file = None
+        self.assertEqual(hkdump.debug_file_type(), None)
+
+        hkdump.params.debug_file = ""
+        self.assertEqual(hkdump.debug_file_type(), None)
+
+    @mock.patch("hotkdump.core.hotkdump.Hotkdump.extract_vmlinux_ddeb")
+    def test_maybe_get_user_specified_vmlinux(self, mock_extract_ddeb):
+        """Verify that the correct user psecified vmlinux file path is returned"""
+        ddeb_path = "/tmp/home/user/path/to/ddeb/linux-yy.xx.ddeb"
+        vmlinux_path = "/tmp/home/user/path/to/vmlinux/vmlinux-yy.xx"
+        ex_ddeb_path = "/tmp/home/user/path/to/ddeb/tmp/ddeb-root/vmlinux-yy.xx"
+
+        # Mocks a failed extraction
+        mock_extract_ddeb.return_value = None
+
+        params = HotkdumpParameters(debug_file=ddeb_path, dump_file_path="empty")
+        hkdump = Hotkdump(params)
+        self.assertEqual(hkdump.maybe_get_user_specified_vmlinux(), None)
+
+        # Mocks a successfull extraction
+        mock_extract_ddeb.return_value = ex_ddeb_path
+        self.assertEqual(hkdump.maybe_get_user_specified_vmlinux(), ex_ddeb_path)
+
+        hkdump.params.debug_file = vmlinux_path
+        self.assertEqual(hkdump.maybe_get_user_specified_vmlinux(), vmlinux_path)
+
+        hkdump.params.debug_file = None
+        self.assertEqual(hkdump.maybe_get_user_specified_vmlinux(), None)