Create --native-last arg

src/pystack/__main__.py

@@ -160,6 +160,15 @@ def generate_cli_parser() -> argparse.ArgumentParser:
         help="Include native (C) frames from threads not registered with "
         "the interpreter (implies --native)",
     )
+    remote_parser.add_argument(
+        "--native-last",
+        action="store_const",
+        dest="native_mode",
+        const=NativeReportingMode.LAST,
+        default=NativeReportingMode.OFF,
+        help="Include native (C) frames only after the last python frame "
+        "in the resulting stack trace",
+    )
     remote_parser.add_argument(
         "--locals",
         action="store_true",
@@ -206,6 +215,15 @@ def generate_cli_parser() -> argparse.ArgumentParser:
         help="Include native (C) frames from threads not registered with "
         "the interpreter (implies --native)",
     )
+    core_parser.add_argument(
+        "--native-last",
+        action="store_const",
+        dest="native_mode",
+        const=NativeReportingMode.LAST,
+        default=NativeReportingMode.OFF,
+        help="Include native (C) frames only after the last python frame "
+        "in the resulting stack trace",
+    )
     core_parser.add_argument(
         "--locals",
         action="store_true",
@@ -269,7 +287,8 @@ def process_remote(parser: argparse.ArgumentParser, args: argparse.Namespace) ->
         method=StackMethod.ALL if args.exhaustive else StackMethod.AUTO,
     ):
         native = args.native_mode != NativeReportingMode.OFF
-        print_thread(thread, native)
+        native_last = args.native_mode == NativeReportingMode.LAST
+        print_thread(thread, native, native_last)
 
 
 def format_psinfo_information(psinfo: Dict[str, Any]) -> str:
@@ -378,7 +397,8 @@ def process_core(parser: argparse.ArgumentParser, args: argparse.Namespace) -> N
         method=StackMethod.ALL if args.exhaustive else StackMethod.AUTO,
     ):
         native = args.native_mode != NativeReportingMode.OFF
-        print_thread(thread, native)
+        native_last = args.native_mode == NativeReportingMode.LAST
+        print_thread(thread, native, native_last)
 
 
 if __name__ == "__main__":  # pragma: no cover

src/pystack/_pystack.pyi

@@ -26,6 +26,7 @@ class NativeReportingMode(enum.Enum):
     ALL: int
     OFF: int
     PYTHON: int
+    LAST: int
 
 class StackMethod(enum.Enum):
     ALL: int

src/pystack/_pystack.pyx

@@ -85,6 +85,7 @@ class NativeReportingMode(enum.Enum):
     OFF = 0
     PYTHON = 1
     ALL = 1000
+    LAST = 2000
 
 
 cdef api void log_with_python(const cppstring *message, int level) noexcept:

src/pystack/traceback_formatter.py

@@ -11,8 +11,8 @@
 from .types import frame_type
 
 
-def print_thread(thread: PyThread, native: bool) -> None:
-    for line in format_thread(thread, native):
+def print_thread(thread: PyThread, native: bool, native_last: bool = False) -> None:
+    for line in format_thread(thread, native, native_last):
         print(line, file=sys.stdout, flush=True)
 
 
@@ -62,7 +62,9 @@ def _are_the_stacks_mergeable(thread: PyThread) -> bool:
     return n_eval_frames == n_entry_frames
 
 
-def format_thread(thread: PyThread, native: bool) -> Iterable[str]:
+def format_thread(
+    thread: PyThread, native: bool, native_last: bool = False
+) -> Iterable[str]:
     current_frame: Optional[PyFrame] = thread.frame
     if current_frame is None and not native:
         yield f"The frame stack for thread {thread.tid} is empty"
@@ -81,16 +83,20 @@ def format_thread(thread: PyThread, native: bool) -> Iterable[str]:
             yield from format_frame(current_frame)
             current_frame = current_frame.next
     else:
-        yield from _format_merged_stacks(thread, current_frame)
+        yield from _format_merged_stacks(thread, current_frame, native_last)
     yield ""
 
 
 def _format_merged_stacks(
-    thread: PyThread, current_frame: Optional[PyFrame]
+    thread: PyThread,
+    current_frame: Optional[PyFrame],
+    native_last: bool = False,
 ) -> Iterable[str]:
+    c_frames_list: list[str] = []
     for frame in thread.native_frames:
         if frame_type(frame, thread.python_version) == NativeFrame.FrameType.EVAL:
             assert current_frame is not None
+            c_frames_list = []
             yield from format_frame(current_frame)
             current_frame = current_frame.next
             while current_frame and not current_frame.is_entry:
@@ -101,12 +107,18 @@ def _format_merged_stacks(
             continue
         elif frame_type(frame, thread.python_version) == NativeFrame.FrameType.OTHER:
             function = colored(frame.symbol, "yellow")
-            yield (
+            formatted_c_frame = (
                 f'    {colored("(C)", "blue")} File "{frame.path}",'
                 f" line {frame.linenumber},"
                 f" in {function} ({colored(frame.library, attrs=['faint'])})"
             )
+            if native_last:
+                c_frames_list.append(formatted_c_frame)
+            else:
+                yield formatted_c_frame
         else:  # pragma: no cover
             raise ValueError(
                 f"Invalid frame type: {frame_type(frame, thread.python_version)}"
             )
+    for c_frame in c_frames_list:
+        yield c_frame

tests/unit/test_main.py

@@ -205,7 +205,9 @@ def test_process_remote_default():
         locals=False,
         method=StackMethod.AUTO,
     )
-    assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, False, False) for thread in threads
+    ]
 
 
 def test_process_remote_no_block():
@@ -236,13 +238,55 @@ def test_process_remote_no_block():
         locals=False,
         method=StackMethod.AUTO,
     )
-    assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, False, False) for thread in threads
+    ]
+
+
+@pytest.mark.parametrize(
+    "argument, mode",
+    [
+        ["--native", NativeReportingMode.PYTHON],
+        ["--native-all", NativeReportingMode.ALL],
+    ],
+)
+def test_process_remote_native(argument, mode):
+    # GIVEN
+
+    argv = ["pystack", "remote", "31", argument]
+
+    threads = [Mock(), Mock(), Mock()]
+
+    # WHEN
+
+    with patch(
+        "pystack.__main__.get_process_threads"
+    ) as get_process_threads_mock, patch(
+        "pystack.__main__.print_thread"
+    ) as print_thread_mock, patch(
+        "sys.argv", argv
+    ):
+        get_process_threads_mock.return_value = threads
+        main()
+
+    # THEN
+
+    get_process_threads_mock.assert_called_with(
+        31,
+        stop_process=True,
+        native_mode=mode,
+        locals=False,
+        method=StackMethod.AUTO,
+    )
+    assert print_thread_mock.mock_calls == [
+        call(thread, True, False) for thread in threads
+    ]
 
 
-def test_process_remote_native():
+def test_process_remote_native_last():
     # GIVEN
 
-    argv = ["pystack", "remote", "31", "--native"]
+    argv = ["pystack", "remote", "31", "--native-last"]
 
     threads = [Mock(), Mock(), Mock()]
 
@@ -263,11 +307,13 @@ def test_process_remote_native():
     get_process_threads_mock.assert_called_with(
         31,
         stop_process=True,
-        native_mode=NativeReportingMode.PYTHON,
+        native_mode=NativeReportingMode.LAST,
         locals=False,
         method=StackMethod.AUTO,
     )
-    assert print_thread_mock.mock_calls == [call(thread, True) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, True, True) for thread in threads
+    ]
 
 
 def test_process_remote_locals():
@@ -298,7 +344,9 @@ def test_process_remote_locals():
         locals=True,
         method=StackMethod.AUTO,
     )
-    assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, False, False) for thread in threads
+    ]
 
 
 def test_process_remote_native_no_block(capsys):
@@ -355,7 +403,9 @@ def test_process_remote_exhaustive():
         locals=False,
         method=StackMethod.ALL,
     )
-    assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, False, False) for thread in threads
+    ]
 
 
 @pytest.mark.parametrize(
@@ -428,7 +478,9 @@ def test_process_core_default_without_executable():
         locals=False,
         method=StackMethod.AUTO,
     )
-    assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, False, False) for thread in threads
+    ]
 
 
 def test_process_core_default_without_executable_and_executable_does_not_exist(capsys):
@@ -517,7 +569,9 @@ def test_process_core_default_with_executable():
         locals=False,
         method=StackMethod.AUTO,
     )
-    assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, False, False) for thread in threads
+    ]
 
 
 @pytest.mark.parametrize(
@@ -562,7 +616,49 @@ def test_process_core_native(argument, mode):
         locals=False,
         method=StackMethod.AUTO,
     )
-    assert print_thread_mock.mock_calls == [call(thread, True) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, True, False) for thread in threads
+    ]
+
+
+def test_process_core_native_last():
+    # GIVEN
+
+    argv = ["pystack", "core", "corefile", "executable", "--native-last"]
+
+    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__.CoreFileAnalyzer"
+    ), patch(
+        "pystack.__main__.is_elf", return_value=True
+    ):
+        get_process_threads_mock.return_value = threads
+        main()
+
+    # THEN
+
+    get_process_threads_mock.assert_called_with(
+        Path("corefile"),
+        Path("executable"),
+        library_search_path="",
+        native_mode=NativeReportingMode.LAST,
+        locals=False,
+        method=StackMethod.AUTO,
+    )
+    assert print_thread_mock.mock_calls == [
+        call(thread, True, True) for thread in threads
+    ]
 
 
 def test_process_core_locals():
@@ -600,7 +696,9 @@ def test_process_core_locals():
         locals=True,
         method=StackMethod.AUTO,
     )
-    assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, False, False) for thread in threads
+    ]
 
 
 def test_process_core_with_search_path():
@@ -645,7 +743,9 @@ def test_process_core_with_search_path():
         locals=False,
         method=StackMethod.AUTO,
     )
-    assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, False, False) for thread in threads
+    ]
 
 
 def test_process_core_with_search_root():
@@ -691,7 +791,9 @@ def test_process_core_with_search_root():
         locals=False,
         method=StackMethod.AUTO,
     )
-    assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, False, False) for thread in threads
+    ]
 
 
 def test_process_core_with_not_readable_search_root():
@@ -870,7 +972,9 @@ def test_process_core_exhaustive():
         locals=False,
         method=StackMethod.ALL,
     )
-    assert print_thread_mock.mock_calls == [call(thread, False) for thread in threads]
+    assert print_thread_mock.mock_calls == [
+        call(thread, False, False) for thread in threads
+    ]
 
 
 def test_default_colored_output():