Sync canonical asyncio, marshal and offline profiling libraries (#9315)

Migrate nine substantial stdlib and canonical test modules from CPython
v3.15.0 through scripts/update_lib, preserving RustPython adaptations.
Include offline stack formatting without live process inspection or
changes to the held profiling frontend and test-support closure.

Assisted-by: Codex:unknown

Lib/asyncio/__main__.py

@@ -12,13 +12,16 @@ import threading
 import types
 import warnings
 
-from _colorize import get_theme
-from _pyrepl.console import InteractiveColoredConsole
+try:
+    from _colorize import get_theme
+    from _pyrepl.console import InteractiveColoredConsole as InteractiveConsole
+except ModuleNotFoundError:
+    from code import InteractiveConsole
 
 from . import futures
 
 
-class AsyncIOInteractiveConsole(InteractiveColoredConsole):
+class AsyncIOInteractiveConsole(InteractiveConsole):
 
     def __init__(self, locals, loop):
         super().__init__(locals, filename="<stdin>")
@@ -96,7 +99,7 @@ class REPLThread(threading.Thread):
 
                 console.write(banner)
 
-            if not sys.flags.isolated and (startup_path := os.getenv("PYTHONSTARTUP")):
+            if not sys.flags.ignore_environment and (startup_path := os.getenv("PYTHONSTARTUP")):
                 sys.audit("cpython.run_startup", startup_path)
                 try:
                     import tokenize
@@ -112,13 +115,17 @@ class REPLThread(threading.Thread):
             if CAN_USE_PYREPL:
                 theme = get_theme().syntax
                 ps1 = f"{theme.prompt}{ps1}{theme.reset}"
-            console.write(f"{ps1}import asyncio\n")
+                import_line = f'{theme.keyword}import{theme.reset} asyncio'
+            else:
+                import_line = "import asyncio"
+            console.write(f"{ps1}{import_line}\n")
 
             if CAN_USE_PYREPL:
                 from _pyrepl.simple_interact import (
                     run_multiline_interactive_console,
                 )
                 try:
+                    sys.ps1 = ps1
                     run_multiline_interactive_console(console)
                 except SystemExit:
                     # expected via the `exit` and `quit` commands
@@ -152,24 +159,35 @@ if __name__ == '__main__':
     parser = argparse.ArgumentParser(
         prog="python3 -m asyncio",
         description="Interactive asyncio shell and CLI tools",
-        color=True,
     )
     subparsers = parser.add_subparsers(help="sub-commands", dest="command")
     ps = subparsers.add_parser(
         "ps", help="Display a table of all pending tasks in a process"
     )
     ps.add_argument("pid", type=int, help="Process ID to inspect")
+    ps.add_argument(
+        "--retries",
+        type=int,
+        default=3,
+        help="Number of retries on transient attach errors",
+    )
     pstree = subparsers.add_parser(
         "pstree", help="Display a tree of all pending tasks in a process"
     )
     pstree.add_argument("pid", type=int, help="Process ID to inspect")
+    pstree.add_argument(
+        "--retries",
+        type=int,
+        default=3,
+        help="Number of retries on transient attach errors",
+    )
     args = parser.parse_args()
     match args.command:
         case "ps":
-            asyncio.tools.display_awaited_by_tasks_table(args.pid)
+            asyncio.tools.display_awaited_by_tasks_table(args.pid, retries=args.retries)
             sys.exit(0)
         case "pstree":
-            asyncio.tools.display_awaited_by_tasks_tree(args.pid)
+            asyncio.tools.display_awaited_by_tasks_tree(args.pid, retries=args.retries)
             sys.exit(0)
         case None:
             pass  # continue to the interactive shell
@@ -182,20 +200,26 @@ if __name__ == '__main__':
 
     sys.audit("cpython.run_stdin")
 
-    if os.getenv('PYTHON_BASIC_REPL'):
+    if not sys.flags.ignore_environment and os.getenv('PYTHON_BASIC_REPL'):
         CAN_USE_PYREPL = False
     else:
-        from _pyrepl.main import CAN_USE_PYREPL
+        try:
+            from _pyrepl.main import CAN_USE_PYREPL
+        except ModuleNotFoundError:
+            CAN_USE_PYREPL = False
 
     return_code = 0
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
 
-    repl_locals = {'asyncio': asyncio}
-    for key in {'__name__', '__package__',
-                '__loader__', '__spec__',
-                '__builtins__', '__file__'}:
-        repl_locals[key] = locals()[key]
+    repl_locals = {
+        'asyncio': asyncio,
+        '__name__': __name__,
+        '__package__': None,
+        '__loader__': __loader__,
+        '__spec__': None,
+        '__builtins__': __builtins__,
+    }
 
     console = AsyncIOInteractiveConsole(repl_locals, loop)
 

Lib/ctypes/macholib/__init__.py

@@ -6,4 +6,10 @@ See the relevant header files in /usr/include/mach-o
 And also Apple's documentation.
 """
 
-__version__ = '1.0'
+def __getattr__(name):
+    if name == "__version__":
+        from warnings import _deprecated
+
+        _deprecated("__version__", remove=(3, 20))
+        return "1.0"  # Do not change
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

Lib/profiling/sampling/dump.py

@@ -0,0 +1,295 @@
+"""Pretty printing for one-shot sampling stack dumps."""
+
+import contextlib
+import linecache
+import os
+import sys
+from traceback import _byte_offset_to_character_offset
+
+import _colorize
+
+from .collector import extract_lineno, filter_internal_frames, iter_async_frames
+from .constants import (
+    THREAD_STATUS_GIL_REQUESTED,
+    THREAD_STATUS_HAS_EXCEPTION,
+    THREAD_STATUS_HAS_GIL,
+    THREAD_STATUS_MAIN_THREAD,
+    THREAD_STATUS_ON_CPU,
+    THREAD_STATUS_UNKNOWN,
+)
+from .opcode_utils import format_opcode
+
+
+_STATUS_LABELS = (
+    (THREAD_STATUS_MAIN_THREAD, "main thread"),
+    (THREAD_STATUS_HAS_GIL, "has GIL"),
+    (THREAD_STATUS_ON_CPU, "on CPU"),
+    (THREAD_STATUS_GIL_REQUESTED, "waiting for GIL"),
+    (THREAD_STATUS_HAS_EXCEPTION, "has exception"),
+)
+
+
+def _theme_for(file, colorize):
+    if colorize is True:
+        return _colorize.get_theme(force_color=True).profiler_dump
+    if colorize is False:
+        return _colorize.get_theme(force_no_color=True).profiler_dump
+    return _colorize.get_theme(tty_file=file).profiler_dump
+
+
+def _color(text, color, theme):
+    if not color:
+        return text
+    return f"{color}{text}{theme.reset}"
+
+
+def _frame_fields(frame):
+    if isinstance(frame, tuple):
+        filename = frame[0] if len(frame) > 0 else ""
+        location = frame[1] if len(frame) > 1 else None
+        qualname = frame[2] if len(frame) > 2 else "<unknown>"
+        opcode = frame[3] if len(frame) > 3 else None
+    else:
+        filename = getattr(frame, "filename", "")
+        location = getattr(frame, "location", None)
+        qualname = getattr(frame, "qualname", None)
+        if qualname is None:
+            qualname = getattr(frame, "funcname", "<unknown>")
+        opcode = getattr(frame, "opcode", None)
+    return filename, location, qualname, opcode
+
+
+def _location_field(location, index, default=None):
+    if location is None:
+        return default
+    try:
+        value = location[index]
+    except (IndexError, TypeError):
+        return default
+    return default if value is None else value
+
+
+def _status_text(status):
+    labels = [label for flag, label in _STATUS_LABELS if status & flag]
+    has_state = status & (
+        THREAD_STATUS_HAS_GIL
+        | THREAD_STATUS_GIL_REQUESTED
+        | THREAD_STATUS_HAS_EXCEPTION
+    )
+    if not has_state and not status & (THREAD_STATUS_UNKNOWN | THREAD_STATUS_ON_CPU):
+        labels.append("idle")
+    return ", ".join(labels) if labels else None
+
+
+def _is_async_dump(stack_frames):
+    return bool(stack_frames) and hasattr(stack_frames[0], "awaited_by")
+
+
+def _iter_dump_sections(stack_frames):
+    if not stack_frames:
+        return
+
+    if _is_async_dump(stack_frames):
+        for frames, thread_id, leaf_id in iter_async_frames(stack_frames):
+            frames = filter_internal_frames(frames)
+            if frames:
+                yield None, thread_id, None, frames, f"task {leaf_id}"
+        return
+
+    for interpreter_info in stack_frames:
+        interpreter_id = getattr(interpreter_info, "interpreter_id", None)
+        for thread_info in getattr(interpreter_info, "threads", ()):
+            frames = getattr(thread_info, "frame_info", None) or []
+            frames = filter_internal_frames(frames)
+            yield (
+                interpreter_id,
+                getattr(thread_info, "thread_id", None),
+                getattr(thread_info, "status", None),
+                frames,
+                None,
+            )
+
+
+def _display_filename(filename):
+    if not filename or filename == "~":
+        return filename
+    with contextlib.suppress(ValueError):
+        relpath = os.path.relpath(filename)
+        if not relpath.startswith(".." + os.sep) and relpath != "..":
+            return relpath
+    return filename
+
+
+def _format_frame(frame, theme):
+    filename, location, qualname, opcode = _frame_fields(frame)
+    source_filename = filename
+    lineno = extract_lineno(location)
+    qualname_part = _color(qualname, theme.frame, theme)
+
+    if filename == "~" and lineno == 0:
+        line = f"  {qualname_part}"
+    else:
+        filename = _display_filename(filename)
+        if filename:
+            file_part = _color(f'"{filename}"', theme.filename, theme)
+            if lineno > 0:
+                line_part = _color(str(lineno), theme.line_no, theme)
+                line = f"  File {file_part}, line {line_part}, in {qualname_part}"
+            else:
+                line = f"  File {file_part}, in {qualname_part}"
+        else:
+            line = f"  {qualname_part}"
+
+    if opcode is not None:
+        line = f"{line}  {_color(f'opcode={format_opcode(opcode)}', theme.opcode, theme)}"
+
+    lines = [line]
+    source = _source_line(source_filename, location, lineno, theme)
+    if source:
+        lines.append(f"    {source}")
+    return lines
+
+
+def _source_offsets(line, location, lineno):
+    end_lineno = _location_field(location, 1, lineno)
+    col_offset = _location_field(location, 2, -1)
+    end_col_offset = _location_field(location, 3, -1)
+
+    if col_offset < 0 or end_col_offset < 0 or end_lineno < lineno:
+        return None
+
+    start = _byte_offset_to_character_offset(line, col_offset)
+    if end_lineno == lineno:
+        end = _byte_offset_to_character_offset(line, end_col_offset)
+    else:
+        end = len(line)
+    if start < 0 or end <= start:
+        return None
+    return start, end
+
+
+def _trim_source_line(line, offsets):
+    stripped = line.lstrip()
+    leading = len(line) - len(stripped)
+    if offsets is None:
+        return stripped, None
+
+    start, end = offsets
+    start = max(start - leading, 0)
+    end = max(end - leading, start + 1)
+    end = min(end, len(stripped))
+    return stripped, (start, end)
+
+
+def _highlight_source_line(line, offsets, theme):
+    if offsets is None or offsets[1] <= offsets[0]:
+        return _color(line, theme.source, theme)
+
+    start, end = offsets
+    parts = []
+    if line[:start]:
+        parts.append(_color(line[:start], theme.source, theme))
+    parts.append(_color(line[start:end], theme.source_highlight, theme))
+    if line[end:]:
+        parts.append(_color(line[end:], theme.source, theme))
+    return "".join(parts)
+
+
+def _source_line(filename, location, lineno, theme):
+    if not filename or filename == "~" or lineno <= 0:
+        return None
+    line = linecache.getline(filename, lineno).removesuffix("\n")
+    if not line:
+        return None
+
+    offsets = _source_offsets(line, location, lineno)
+    line, offsets = _trim_source_line(line, offsets)
+    if not line:
+        return None
+    return _highlight_source_line(line, offsets, theme)
+
+
+def _section_header(
+    *,
+    pid,
+    interpreter_id,
+    thread_id,
+    status,
+    label,
+    show_pid,
+    show_interpreter,
+    theme,
+):
+    subject = "Stack dump"
+    if show_pid and pid is not None:
+        subject = f"{subject} for PID {pid}"
+    if thread_id is not None:
+        subject = f"{subject}, thread {thread_id}"
+
+    details = []
+    if show_interpreter and interpreter_id is not None:
+        details.append(f"interpreter {interpreter_id}")
+    if label:
+        details.append(label)
+    if status is not None:
+        status_text = _status_text(status)
+        if status_text:
+            details.append(status_text)
+
+    suffix = "most recent call last"
+    if details:
+        suffix = f"{'; '.join(details)}; {suffix}"
+    return _color(f"{subject} ({suffix}):", theme.header, theme)
+
+
+def format_stack_dump(stack_frames, *, pid=None, file=None, colorize=None):
+    """Return a formatted one-shot stack dump."""
+    if file is None:
+        file = sys.stdout
+
+    theme = _theme_for(file, colorize)
+    lines = []
+    sections = list(_iter_dump_sections(stack_frames))
+    if not sections:
+        if pid is None:
+            return f"{_color('No Python stacks found', theme.warning, theme)}\n"
+        return f"{_color(f'No Python stacks found for PID {pid}', theme.warning, theme)}\n"
+
+    interpreter_ids = {
+        interpreter_id
+        for interpreter_id, _thread_id, _status, _frames, _label in sections
+        if interpreter_id is not None
+    }
+    show_interpreter = len(interpreter_ids) > 1
+
+    for section_index, (interpreter_id, thread_id, status, frames, label) in enumerate(sections):
+        if section_index:
+            lines.append("")
+        lines.append(
+            _section_header(
+                pid=pid,
+                interpreter_id=interpreter_id,
+                thread_id=thread_id,
+                status=status,
+                label=label,
+                show_pid=section_index == 0,
+                show_interpreter=show_interpreter,
+                theme=theme,
+            )
+        )
+
+        if not frames:
+            lines.append(_color("No Python frames", theme.warning, theme))
+            continue
+
+        for frame in reversed(frames):
+            lines.extend(_format_frame(frame, theme))
+
+    return "\n".join(lines) + "\n"
+
+
+def print_stack_dump(stack_frames, *, pid=None, file=None, colorize=None):
+    """Pretty-print a one-shot stack dump."""
+    if file is None:
+        file = sys.stdout
+    file.write(format_stack_dump(stack_frames, pid=pid, file=file, colorize=colorize))

Lib/test/test__colorize.py

@@ -5,6 +5,7 @@ import sys
 import unittest
 import unittest.mock
 import _colorize
+from test.support import cpython_only, import_helper
 from test.support.os_helper import EnvironmentVarGuard
 
 
@@ -22,6 +23,15 @@ def supports_virtual_terminal():
         return contextlib.nullcontext()
 
 
+class TestImportTime(unittest.TestCase):
+
+    @cpython_only
+    def test_lazy_import(self):
+        import_helper.ensure_lazy_imports(
+            "_colorize", {"copy", "re", "inspect", "typing"}
+        )
+
+
 class TestTheme(unittest.TestCase):
 
     def test_attributes(self):
@@ -40,6 +50,7 @@ class TestTheme(unittest.TestCase):
         unittest_no_colors = _colorize.Unittest.no_colors()
         copy = theme.copy_with(unittest=unittest_no_colors)
         self.assertEqual(copy.argparse, theme.argparse)
+        self.assertEqual(copy.difflib, theme.difflib)
         self.assertEqual(copy.syntax, theme.syntax)
         self.assertEqual(copy.traceback, theme.traceback)
         self.assertEqual(copy.unittest, unittest_no_colors)

Lib/test/test_asyncio/test_ssl.py

@@ -1544,6 +1544,9 @@ class TestSSL(test_utils.TestCase):
             # This triggers bug gh-115514, also tested using mocks in
             # test.test_asyncio.test_selector_events.SelectorSocketTransportTests.test_write_buffer_after_close
             socket_transport = writer.transport._ssl_protocol._transport
+            # connection_lost may have already cleared _transport.
+            if socket_transport is None:
+                return
 
             class SocketWrapper:
                 def __init__(self, sock) -> None:

Lib/test/test_asyncio/test_sslproto.py

@@ -70,6 +70,50 @@ class SslProtoHandshakeTests(test_utils.TestCase):
             sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter,
                                  ssl_handshake_timeout=-10)
 
+    def test_check_hostname_accepts_server_hostname(self):
+        # Supplying a server_hostname succeeds with check_hostname enabled.
+        sslcontext = test_utils.simple_client_sslcontext(disable_verify=False)
+        sslcontext.check_hostname = True
+        app_proto = mock.Mock()
+        waiter = mock.Mock()
+
+        # No ValueError is raised from SSLProtocol with 'server_hostname'.
+        ssl_proto = sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter,
+                             server_hostname='example.org')
+        self.addCleanup(ssl_proto._app_transport.close)
+
+    @support.subTests("server_hostname", [None, ''])
+    def test_check_hostname_requires_server_hostname(self, server_hostname):
+        # A caller-supplied context asking for hostname checking used to be
+        # taken through wrap_bio() with no name to check against, verifying
+        # the certificate chain but never the peer's identity.
+        # loop.start_tls() defaults server_hostname to None, and
+        # loop.create_connection() turns server_hostname='' into None here,
+        # so both reached that state.
+        sslcontext = test_utils.simple_client_sslcontext(disable_verify=False)
+        sslcontext.check_hostname = True
+        app_proto = mock.Mock()
+        waiter = mock.Mock()
+
+        # Supplying an empty server_hostname fails with check_hostname enabled.
+        with self.assertRaisesRegex(
+                ValueError,
+                'check_hostname requires server_hostname'):
+            sslproto.SSLProtocol(self.loop, app_proto, sslcontext,
+                                 waiter,
+                                 server_hostname=server_hostname)
+
+        # Disabling check_hostname allows for an empty or unset server_hostname.
+        sslcontext.check_hostname = False
+
+        ssl_proto = sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter)
+        self.addCleanup(ssl_proto._app_transport.close)
+
+        ssl_proto = sslproto.SSLProtocol(self.loop, app_proto, sslcontext,
+                             waiter,
+                             server_hostname=server_hostname)
+        self.addCleanup(ssl_proto._app_transport.close)
+
     def test_eof_received_waiter(self):
         waiter = self.loop.create_future()
         ssl_proto = self.ssl_protocol(waiter=waiter)

Lib/test/test_ctypes/test_macholib.py

@@ -108,5 +108,17 @@ class MachOTest(unittest.TestCase):
                          d('P', 'F.framework/Versions/A/F_debug', 'F', 'A', 'debug'))
 
 
+class TestModule(unittest.TestCase):
+    def test_deprecated__version__(self):
+        import ctypes.macholib
+
+        with self.assertWarnsRegex(
+            DeprecationWarning,
+            "'__version__' is deprecated and slated for removal in Python 3.20",
+        ) as cm:
+            getattr(ctypes.macholib, "__version__")
+        self.assertEqual(cm.filename, __file__)
+
+
 if __name__ == "__main__":
     unittest.main()

Lib/test/test_marshal.py

@@ -116,6 +116,14 @@ class StringTestCase(unittest.TestCase, HelperMixin):
         for s in [b"", b"Andr\xe8 Previn", b"abc", b" "*10000]:
             self.helper(s)
 
+    @support.cpython_only
+    def test_bytes_singleton(self):
+        for version in range(marshal.version + 1):
+            for sample in [b"", b"x"]:
+                new = marshal.loads(marshal.dumps(sample, version))
+                self.assertIs(new, sample)
+
+
 class ExceptionTestCase(unittest.TestCase):
     def test_exceptions(self):
         new = marshal.loads(marshal.dumps(StopIteration))
@@ -382,6 +390,12 @@ class BugsTestCase(unittest.TestCase):
         for v in range(marshal.version + 1):
             self.assertRaises(ValueError, marshal.dumps, a, v)
 
+    def test_reference_loop_frozendict(self):
+        a = frozendict({None: []})
+        a[None].append(a)
+        for v in range(marshal.version + 1):
+            self.assertRaises(ValueError, marshal.dumps, a, v)
+
     def test_loads_reference_loop_list(self):
         data = b'\xdb\x01\x00\x00\x00r\x00\x00\x00\x00' # [<R>]
         a = marshal.loads(data)
@@ -431,9 +445,13 @@ class BugsTestCase(unittest.TestCase):
             b'\xba{Nr\x00\x00\x00\x000NN', # slice({None: <R>}, None)
             b'\xbaN{Nr\x00\x00\x00\x000N', # slice(None, {None: <R>})
             b'\xbaNN{Nr\x00\x00\x00\x000', # slice(None, None, {None: <R>})
+            b'\xfdN[\x01\x00\x00\x00r\x00\x00\x00\x000', # frozendict({None: [<R>]})
+            b'\xfdN{Nr\x00\x00\x00\x0000', # frozendict({None: {None: <R>})
 
             # Direct self-references which cannot be created in Python.
             b'\xbe\x01\x00\x00\x00r\x00\x00\x00\x00', # frozenset({<R>})
+            b'\xfdNr\x00\x00\x00\x000', # frozendict({None: <R>})
+            b'\xfdr\x00\x00\x00\x00N0', # frozendict({<R>: None})
             b'\xbar\x00\x00\x00\x00NN', # slice(<R>, None)
             b'\xbaNr\x00\x00\x00\x00N', # slice(None, <R>)
             b'\xbaNNr\x00\x00\x00\x00', # slice(None, None, <R>)
@@ -695,6 +713,15 @@ class InstancingTestCase(unittest.TestCase, HelperMixin):
             self.helper(dictobj)
             self.helper3(dictobj)
 
+    def testFrozenDict(self):
+        for obj in self.keys:
+            dictobj = frozendict({"hello": obj, "goodbye": obj, obj: "hello"})
+            self.helper(dictobj)
+
+            for version in range(6):
+                with self.assertRaises(ValueError):
+                    marshal.dumps(dictobj, version)
+
     def testModule(self):
         with open(__file__, "rb") as f:
             code = f.read()
@@ -760,7 +787,7 @@ class SliceTestCase(unittest.TestCase, HelperMixin):
             with self.subTest(obj=str(obj)):
                 self.helper(obj)
 
-                for version in range(4):
+                for version in range(5):
                     with self.assertRaises(ValueError):
                         marshal.dumps(obj, version)
 

Lib/test/test_profiling/test_sampling_profiler/test_dump.py

@@ -0,0 +1,454 @@
+"""Tests for one-shot sampling profiler stack dumps."""
+
+from collections import namedtuple
+import os
+import opcode
+import tempfile
+import unittest
+
+import _colorize
+
+from profiling.sampling.constants import (
+    THREAD_STATUS_GIL_REQUESTED,
+    THREAD_STATUS_HAS_EXCEPTION,
+    THREAD_STATUS_HAS_GIL,
+    THREAD_STATUS_MAIN_THREAD,
+    THREAD_STATUS_ON_CPU,
+    THREAD_STATUS_UNKNOWN,
+)
+from profiling.sampling.dump import format_stack_dump
+
+from .mocks import (
+    LocationInfo as StructseqLocationInfo,
+    MockAwaitedInfo,
+    MockCoroInfo,
+    MockFrameInfo,
+    MockInterpreterInfo,
+    MockTaskInfo,
+    MockThreadInfo,
+)
+
+try:
+    import _remote_debugging  # noqa: F401
+except ImportError:
+    _remote_debugging = None
+
+
+StructseqInterpreterInfo = namedtuple(
+    "StructseqInterpreterInfo",
+    ["interpreter_id", "threads"],
+)
+StructseqThreadInfo = namedtuple(
+    "StructseqThreadInfo",
+    ["thread_id", "status", "frame_info"],
+)
+StructseqFrameInfo = namedtuple(
+    "StructseqFrameInfo",
+    ["filename", "location", "funcname", "opcode"],
+)
+
+
+class TestStackDumpFormatting(unittest.TestCase):
+    def test_format_stack_dump_single_thread(self):
+        frames = [
+            MockFrameInfo("leaf.py", 10, "leaf"),
+            MockFrameInfo("root.py", 1, "root"),
+        ]
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [
+                    MockThreadInfo(
+                        123,
+                        frames,
+                        status=(
+                            THREAD_STATUS_MAIN_THREAD
+                            | THREAD_STATUS_HAS_GIL
+                            | THREAD_STATUS_ON_CPU
+                        ),
+                    )
+                ],
+            )
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn(
+            "Stack dump for PID 42, thread 123 "
+            "(main thread, has GIL, on CPU; most recent call last):",
+            output,
+        )
+        self.assertNotIn("Interpreter 0", output)
+        self.assertLess(
+            output.index('File "root.py", line 1, in root'),
+            output.index('File "leaf.py", line 10, in leaf'),
+        )
+        self.assertIn('  File "root.py", line 1, in root', output)
+        self.assertIn('  File "leaf.py", line 10, in leaf', output)
+        self.assertNotIn("\x1b[", output)
+
+    def test_format_stack_dump_with_structseq_tuples(self):
+        stack_frames = [
+            StructseqInterpreterInfo(
+                0,
+                [
+                    StructseqThreadInfo(
+                        123,
+                        THREAD_STATUS_HAS_GIL,
+                        [
+                            StructseqFrameInfo(
+                                "file.py",
+                                StructseqLocationInfo(5, 5, -1, -1),
+                                "func",
+                                None,
+                            )
+                        ],
+                    )
+                ],
+            )
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn(
+            "Stack dump for PID 42, thread 123 "
+            "(has GIL; most recent call last):",
+            output,
+        )
+        self.assertIn('  File "file.py", line 5, in func', output)
+
+    def test_format_stack_dump_shows_interpreter_ids_when_multiple(self):
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [MockThreadInfo(100, [MockFrameInfo("main.py", 1, "main")])],
+            ),
+            MockInterpreterInfo(
+                1,
+                [MockThreadInfo(200, [MockFrameInfo("sub.py", 2, "sub")])],
+            ),
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn(
+            "Stack dump for PID 42, thread 100 "
+            "(interpreter 0; idle; most recent call last):",
+            output,
+        )
+        self.assertIn(
+            "Stack dump, thread 200 "
+            "(interpreter 1; idle; most recent call last):",
+            output,
+        )
+
+    def test_format_stack_dump_omits_unknown_status(self):
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [
+                    MockThreadInfo(
+                        123,
+                        [MockFrameInfo("file.py", 5, "func")],
+                        status=THREAD_STATUS_UNKNOWN,
+                    )
+                ],
+            )
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn(
+            "Stack dump for PID 42, thread 123 (most recent call last):",
+            output,
+        )
+        self.assertNotIn("unknown", output)
+
+    def test_format_stack_dump_omits_unknown_when_other_status_exists(self):
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [
+                    MockThreadInfo(
+                        123,
+                        [MockFrameInfo("file.py", 5, "func")],
+                        status=THREAD_STATUS_UNKNOWN | THREAD_STATUS_ON_CPU,
+                    )
+                ],
+            )
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn(
+            "Stack dump for PID 42, thread 123 "
+            "(on CPU; most recent call last):",
+            output,
+        )
+        self.assertNotIn("unknown", output)
+
+    def test_format_stack_dump_labels_known_idle_status(self):
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [
+                    MockThreadInfo(
+                        123,
+                        [MockFrameInfo("file.py", 5, "func")],
+                        status=0,
+                    )
+                ],
+            )
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn(
+            "Stack dump for PID 42, thread 123 "
+            "(idle; most recent call last):",
+            output,
+        )
+
+    def test_format_stack_dump_status_does_not_add_idle_to_waiting_thread(self):
+        status = (
+            THREAD_STATUS_MAIN_THREAD
+            | THREAD_STATUS_GIL_REQUESTED
+            | THREAD_STATUS_HAS_EXCEPTION
+        )
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [MockThreadInfo(123, [MockFrameInfo("file.py", 5, "func")], status=status)],
+            )
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn(
+            "Stack dump for PID 42, thread 123 "
+            "(main thread, waiting for GIL, has exception; most recent call last):",
+            output,
+        )
+        self.assertNotIn("idle", output)
+
+    def test_format_stack_dump_formats_opcode_name(self):
+        load_const = opcode.opmap["LOAD_CONST"]
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [
+                    MockThreadInfo(
+                        123,
+                        [
+                            StructseqFrameInfo(
+                                "file.py",
+                                StructseqLocationInfo(5, 5, -1, -1),
+                                "func",
+                                load_const,
+                            )
+                        ],
+                    )
+                ],
+            )
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn("opcode=LOAD_CONST", output)
+        self.assertNotIn(f"opcode={load_const}", output)
+
+    def test_format_stack_dump_formats_unknown_opcode(self):
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [
+                    MockThreadInfo(
+                        123,
+                        [
+                            StructseqFrameInfo(
+                                "file.py",
+                                StructseqLocationInfo(5, 5, -1, -1),
+                                "func",
+                                999,
+                            )
+                        ],
+                    )
+                ],
+            )
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn("opcode=<999>", output)
+
+    def test_format_stack_dump_prefers_qualname_attribute(self):
+        class Frame:
+            filename = "file.py"
+            location = StructseqLocationInfo(5, 5, -1, -1)
+            funcname = "inner"
+            qualname = "Outer.inner"
+            opcode = None
+
+        stack_frames = [
+            MockInterpreterInfo(0, [MockThreadInfo(123, [Frame()])])
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn('  File "file.py", line 5, in Outer.inner', output)
+        self.assertNotIn("in inner", output)
+
+    def test_format_stack_dump_filters_internal_frames(self):
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [
+                    MockThreadInfo(
+                        123,
+                        [
+                            MockFrameInfo("user.py", 10, "user"),
+                            MockFrameInfo("_sync_coordinator.py", 20, "internal"),
+                        ],
+                    )
+                ],
+            )
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn('  File "user.py", line 10, in user', output)
+        self.assertNotIn("_sync_coordinator.py", output)
+
+    @unittest.skipIf(_remote_debugging is None, "requires _remote_debugging")
+    def test_format_stack_dump_async_task(self):
+        task = MockTaskInfo(
+            task_id=1,
+            task_name="Task-1",
+            coroutine_stack=[
+                MockCoroInfo(
+                    "Task-1",
+                    [MockFrameInfo("task.py", 5, "waiter")],
+                )
+            ],
+        )
+        stack_frames = [MockAwaitedInfo(thread_id=123, awaited_by=[task])]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn(
+            "Stack dump for PID 42, thread 123 "
+            "(task 1; most recent call last):",
+            output,
+        )
+        self.assertIn('  File "task.py", line 5, in waiter', output)
+
+    def test_format_stack_dump_strips_source_like_traceback(self):
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [
+                    MockThreadInfo(
+                        123,
+                        [
+                            StructseqFrameInfo(
+                                __file__,
+                                StructseqLocationInfo(1, 1, -1, -1),
+                                "<module>",
+                                None,
+                            )
+                        ],
+                    )
+                ],
+            )
+        ]
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=False)
+
+        self.assertIn('  File "', output)
+        self.assertIn('", line 1, in <module>', output)
+        self.assertIn('"""Tests for one-shot sampling profiler stack dumps."""', output)
+        self.assertNotIn("^^^^^^^^", output)
+
+    def test_format_stack_dump_highlights_source_range(self):
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [
+                    MockThreadInfo(
+                        123,
+                        [
+                            StructseqFrameInfo(
+                                __file__,
+                                StructseqLocationInfo(1, 1, 0, 3),
+                                "<module>",
+                                None,
+                            )
+                        ],
+                    )
+                ],
+            )
+        ]
+        theme = _colorize.get_theme(force_color=True).profiler_dump
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=True)
+
+        self.assertIn(
+            f"{theme.source_highlight}\"\"\"{theme.reset}",
+            output,
+        )
+        self.assertNotIn("^^^^^^^^", _colorize.decolor(output))
+
+    def test_format_stack_dump_highlights_source_range_after_trimming(self):
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            filename = os.path.join(tmp_dir, "target.py")
+            with open(filename, "w", encoding="utf-8") as file:
+                file.write("    result = call(arg)\n")
+            stack_frames = [
+                MockInterpreterInfo(
+                    0,
+                    [
+                        MockThreadInfo(
+                            123,
+                            [
+                                StructseqFrameInfo(
+                                    filename,
+                                    StructseqLocationInfo(1, 1, 13, 17),
+                                    "<module>",
+                                    None,
+                                )
+                            ],
+                        )
+                    ],
+                )
+            ]
+            theme = _colorize.get_theme(force_color=True).profiler_dump
+
+            output = format_stack_dump(stack_frames, pid=42, colorize=True)
+
+        self.assertIn(f"{theme.source_highlight}call{theme.reset}", output)
+        self.assertIn("\n    result = call(arg)\n", _colorize.decolor(output))
+        self.assertNotIn("\n        result = call(arg)\n", _colorize.decolor(output))
+
+    def test_format_stack_dump_empty(self):
+        output = format_stack_dump([], pid=42, colorize=False)
+
+        self.assertEqual("No Python stacks found for PID 42\n", output)
+
+    def test_format_stack_dump_colorized(self):
+        stack_frames = [
+            MockInterpreterInfo(
+                0,
+                [MockThreadInfo(123, [MockFrameInfo("file.py", 5, "func")])],
+            )
+        ]
+        theme = _colorize.get_theme(force_color=True).profiler_dump
+
+        output = format_stack_dump(stack_frames, pid=42, colorize=True)
+
+        self.assertIn(theme.header, output)
+        self.assertIn(theme.filename, output)
+        self.assertIn(theme.reset, output)
+
+
+if __name__ == "__main__":
+    unittest.main()