Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add gzip support #171

Merged
merged 9 commits into from
Jun 4, 2024
Merged
2 changes: 2 additions & 0 deletions docs/corefile.rst
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,8 @@ In most cases, you just need to provide the location of the core to use PyStack
(Python) File "/test.py", line 16, in third_func
time.sleep(1000)

Pystack can automatically extract core dumps from `.gz` files: `pystack core ./archived_core_file.gz`

To learn more about the different options you can use to customize what is reported, check the :ref:`customizing-the-reports` section.

Providing the executable
Expand Down
19 changes: 19 additions & 0 deletions src/pystack/__main__.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
import argparse
import gzip
import logging
import os
import pathlib
import signal
import sys
import tempfile
from contextlib import suppress
from textwrap import dedent
from typing import Any
Expand Down Expand Up @@ -319,6 +321,19 @@ def process_core(parser: argparse.ArgumentParser, args: argparse.Namespace) -> N
if not corefile.exists():
parser.error(f"Core {corefile} does not exist")

temp_file = None
if corefile.suffix == ".gz":
pablogsal marked this conversation as resolved.
Show resolved Hide resolved
temp_file = tempfile.NamedTemporaryFile(delete=False)
try:
with gzip.open(corefile, "rb") as fp:
while chunk := fp.read(4 * 1024 * 1024):
temp_file.write(chunk)
except gzip.BadGzipFile:
parser.error(f"Core {corefile} is not a valid gzip file")
finally:
temp_file.close()
corefile = pathlib.Path(temp_file.name)

if args.executable is None:
corefile_analyzer = CoreFileAnalyzer(corefile)
executable = pathlib.Path(corefile_analyzer.extract_executable())
Expand Down Expand Up @@ -380,6 +395,10 @@ def process_core(parser: argparse.ArgumentParser, args: argparse.Namespace) -> N
native = args.native_mode != NativeReportingMode.OFF
print_thread(thread, native)

# Delete temporary file created if corefile is a gzip.
if temp_file:
pathlib.Path(temp_file.name).unlink()


if __name__ == "__main__": # pragma: no cover
main()
51 changes: 51 additions & 0 deletions tests/unit/test_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from textwrap import dedent
from unittest.mock import Mock
from unittest.mock import call
from unittest.mock import mock_open
from unittest.mock import patch

import pytest
Expand Down Expand Up @@ -431,6 +432,56 @@ def test_process_core_default_without_executable():
assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]


def test_process_core_default_gzip_without_executable():
# GIVEN

argv = ["pystack", "core", "corefile.gz"]

threads = [Mock(), Mock(), Mock()]

# WHEN

with patch(
"pystack.__main__.get_process_threads_for_core"
) as get_process_threads_mock, patch(
"pystack.__main__.print_thread"
) as print_thread_mock, patch(
"sys.argv", argv
), patch(
"pathlib.Path.exists", return_value=True
), patch(
"pystack.__main__.is_elf", return_value=True
), patch(
"gzip.open", mock_open(read_data=b"")
) as gzip_open_mock, patch(
"tempfile.NamedTemporaryFile"
) as tempfile_mock, patch(
"pystack.__main__.os.unlink", return_value=None
), patch(
"pystack.__main__.CoreFileAnalyzer"
) as core_file_analizer_mock:
core_file_analizer_mock().extract_executable.return_value = (
"extracted_executable"
)
get_process_threads_mock.return_value = threads
tempfile_mock.return_value.name = Path("corefile.gz")

main()

# THEN

get_process_threads_mock.assert_called_with(
Path("corefile.gz"),
Path("extracted_executable"),
library_search_path="",
native_mode=NativeReportingMode.OFF,
locals=False,
method=StackMethod.AUTO,
)
assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
gzip_open_mock.assert_called_with(Path("corefile.gz"), "rb")


def test_process_core_default_without_executable_and_executable_does_not_exist(capsys):
# GIVEN

Expand Down
Loading