Skip to content

Commit

Permalink
Adding unit test, fixes, optimizing maybe_* methods
Browse files Browse the repository at this point in the history
  • Loading branch information
@ghadi-rahme @xmkg
ghadi-rahme authored and xmkg committed Aug 21, 2024
1 parent 21f7a75 commit 1169a30
Show file tree
Hide file tree
Showing 3 changed files with 149 additions and 115 deletions.
206 changes: 107 additions & 99 deletions hotkdump/core/hotkdump.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ def validate_sanity(self):

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()
Expand Down Expand Up @@ -342,49 +342,50 @@ def maybe_download_vmlinux_via_debuginfod(self):
"""Try downloading vmlinux image with debug information
using debuginfod-find."""
try:
if not self.params.no_debuginfod:
debuginfod_find_path = self.find_debuginfod_find_executable()
if not debuginfod_find_path:
logging.debug("debuginfod-find is not present in environment.")
return None

build_id = self.kdump_file.vmcoreinfo.get("BUILD-ID")
if not build_id:
logging.info(
"cannot use debuginfod-find - BUILD-ID not found in vmcoreinfo!"
)
return None

debuginfod_find_args = f"-vvv debuginfo {build_id}"
logging.info("Invoking debuginfod-find with %s", str(debuginfod_find_args))

line = ""
# $HOME/. cache/debuginfod_client/
with subprocess.Popen(
args=f"{debuginfod_find_path} {debuginfod_find_args}",
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
) as proc:
while proc.poll() is None:
lo = proc.stdout.readline()
if lo and not lo.isspace():
line = lo.strip()
self._digest_debuginfod_find_output(line)

result = proc.wait()

if result == 0:
# line should point to the vmcore file
logging.info("debuginfod-find: succeeded, vmcore path: `%s`", line)
return line

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.")
return None

build_id = self.kdump_file.vmcoreinfo.get("BUILD-ID")
if not build_id:
logging.info(
"debuginfod-find: download for BUILD-ID `%s` failed with `%s`",
build_id,
line
)
"cannot use debuginfod-find - BUILD-ID not found in vmcoreinfo!"
)
return None

debuginfod_find_args = f"-vvv debuginfo {build_id}"
logging.info("Invoking debuginfod-find with %s", str(debuginfod_find_args))

line = ""
# $HOME/. cache/debuginfod_client/
with subprocess.Popen(
args=f"{debuginfod_find_path} {debuginfod_find_args}",
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True,
) as proc:
while proc.poll() is None:
lo = proc.stdout.readline()
if lo and not lo.isspace():
line = lo.strip()
self._digest_debuginfod_find_output(line)

result = proc.wait()

if result == 0:
# line should point to the vmcore file
logging.info("debuginfod-find: succeeded, vmcore path: `%s`", line)
return line

logging.info(
"debuginfod-find: download for BUILD-ID `%s` failed with `%s`",
build_id,
line,
)
return None
finally:
self.debuginfod_find_progress = None
Expand All @@ -396,42 +397,43 @@ def maybe_download_vmlinux_via_pullpkg(self):
Returns:
str: The path to the .ddeb file
"""
if not self.params.no_pullpkg:
# 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"
expected_ddeb_path = ddeb_name_format.format(
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"
expected_ddeb_path = ddeb_name_format.format(
self.kdump_file.ddhdr.utsname.release,
self.strip_release_variant_tags(self.kdump_file.ddhdr.utsname.release),
self.kdump_file.ddhdr.utsname.normalized_version,
self.get_architecture(),
)

with switch_cwd(self.params.ddebs_folder_path):
# Check if we already have the .ddeb
if os.path.exists(expected_ddeb_path):
# Already exists, do not download again
# TODO(mkg): Verify SHA checksum?
logging.info(
"The .ddeb file %s already exists, re-using it", expected_ddeb_path
)
# Ensure that the file's last access time is updated
os.utime(expected_ddeb_path, (time.time(), time.time()))
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...",
self.kdump_file.ddhdr.utsname.release,
self.strip_release_variant_tags(self.kdump_file.ddhdr.utsname.release),
self.kdump_file.ddhdr.utsname.normalized_version,
self.get_architecture(),
)

with switch_cwd(self.params.ddebs_folder_path):
# Check if we already have the .ddeb
if os.path.exists(expected_ddeb_path):
# Already exists, do not download again
# TODO(mkg): Verify SHA checksum?
logging.info(
"The .ddeb file %s already exists, re-using it", expected_ddeb_path
)
# Ensure that the file's last access time is updated
os.utime(expected_ddeb_path, (time.time(), time.time()))
extracted_vmlinux = self.extract_vmlinux_ddeb(expected_ddeb_path)
if extracted_vmlinux:
return expected_ddeb_path
logging.error("Failed to extract ddeb file")
return None

logging.info(
"Downloading `vmlinux` image for kernel version %s, please be patient...",
self.kdump_file.ddhdr.utsname.release
)

# (mkg): To force pull-lp-ddebs to use launchpadlibrarian.net for download
# pass an empty mirror list env variable to the hotkdump, e.g.:
# UBUNTUTOOLS_UBUNTU_DDEBS_MIRROR= python3 hotkdump.py -c 123 -d dump.dump
pull_args = [
# (mkg): To force pull-lp-ddebs to use launchpadlibrarian.net for download
# pass an empty mirror list env variable to the hotkdump, e.g.:
# UBUNTUTOOLS_UBUNTU_DDEBS_MIRROR= python3 hotkdump.py -c 123 -d dump.dump
pull_args = [
"--distro",
"ubuntu",
"--arch",
Expand All @@ -441,16 +443,16 @@ def maybe_download_vmlinux_via_pullpkg(self):
f"linux-image-unsigned-{self.kdump_file.ddhdr.utsname.release}",
f"{self.strip_release_variant_tags(self.kdump_file.ddhdr.utsname.release)}"
f".{self.kdump_file.ddhdr.utsname.normalized_version}",
]
logging.info("Invoking PullPkg().pull with %s", str(pull_args))

PullPkg().pull(pull_args)
if not os.path.exists(expected_ddeb_path):
raise ExceptionWithLog(f"failed to download {expected_ddeb_path}")
extracted_vmlinux = self.extract_vmlinux_ddeb(expected_ddeb_path)
if extracted_vmlinux:
return expected_ddeb_path
logging.error("Failed to extract ddeb file.")
]
logging.info("Invoking PullPkg().pull with %s", str(pull_args))

PullPkg().pull(pull_args)
if not os.path.exists(expected_ddeb_path):
raise ExceptionWithLog(f"failed to download {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):
Expand Down Expand Up @@ -514,7 +516,7 @@ def launch_crash(self, vmlinux_path: str):
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]
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):
Expand All @@ -523,15 +525,19 @@ def debug_file_type(self):

def maybe_get_user_specified_vmlinux(self):
"""return the vmlinux file path from the user specified debug file"""
if self.params.debug_file:
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")
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):
Expand All @@ -542,9 +548,11 @@ def run(self):
print(f"{key}={self.kdump_file.vmcoreinfo.get(key)}")
return

extracted_vmlinux = (self.maybe_get_user_specified_vmlinux() or
self.maybe_download_vmlinux_via_debuginfod() or
self.maybe_download_vmlinux_via_pullpkg())
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
Expand Down
5 changes: 4 additions & 1 deletion hotkdump/main.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,10 @@ def main():
ap.add_argument(
"--debug-file",
required=False,
help="Specify the debug file to use. Only ddebs and vmlinux files are supported.",
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()
Expand Down
53 changes: 38 additions & 15 deletions tests/test_hotkdump.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -304,24 +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")
@mock.patch("subprocess.Popen")
def test_maybe_download_vmlinux_ddeb(
self,mock_popen, mock_switch_cwd, mock_pullpkg, mock_utime
):
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")
Expand All @@ -338,10 +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]
mock_popen.return_value.__enter__.return_value.returncode = 0
args[0].return_value.__enter__.return_value.returncode = 0
result = uut.maybe_download_vmlinux_via_pullpkg()

mock_exists.assert_called()
Expand All @@ -361,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
Expand All @@ -376,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")
Expand Down Expand Up @@ -619,11 +620,10 @@ def test_post_run_ddeb_retention_disabled(
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")
"""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")

Expand All @@ -635,4 +635,27 @@ def test_debug_file_type(self):

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)

0 comments on commit 1169a30

Please sign in to comment.