diff --git a/Lib/asyncio/__main__.py b/Lib/asyncio/__main__.py index 44e14771b..cbb052630 100644 --- a/Lib/asyncio/__main__.py +++ b/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="") @@ -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) diff --git a/Lib/ctypes/macholib/__init__.py b/Lib/ctypes/macholib/__init__.py index 5621defcc..f00e55f81 100644 --- a/Lib/ctypes/macholib/__init__.py +++ b/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}") diff --git a/Lib/profiling/sampling/dump.py b/Lib/profiling/sampling/dump.py new file mode 100644 index 000000000..7cf43bd29 --- /dev/null +++ b/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 "" + 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", "") + 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)) diff --git a/Lib/test/test__colorize.py b/Lib/test/test__colorize.py index 026277267..c7bc6914a 100644 --- a/Lib/test/test__colorize.py +++ b/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) diff --git a/Lib/test/test_asyncio/test_ssl.py b/Lib/test/test_asyncio/test_ssl.py index ca15fc3bd..784c784d2 100644 --- a/Lib/test/test_asyncio/test_ssl.py +++ b/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: diff --git a/Lib/test/test_asyncio/test_sslproto.py b/Lib/test/test_asyncio/test_sslproto.py index 3e304c166..455397024 100644 --- a/Lib/test/test_asyncio/test_sslproto.py +++ b/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) diff --git a/Lib/test/test_ctypes/test_macholib.py b/Lib/test/test_ctypes/test_macholib.py index 9d9061799..9a5e18ed5 100644 --- a/Lib/test/test_ctypes/test_macholib.py +++ b/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() diff --git a/Lib/test/test_marshal.py b/Lib/test/test_marshal.py index 9f5cef9be..057dff1e6 100644 --- a/Lib/test/test_marshal.py +++ b/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' # [] a = marshal.loads(data) @@ -431,9 +445,13 @@ class BugsTestCase(unittest.TestCase): b'\xba{Nr\x00\x00\x00\x000NN', # slice({None: }, None) b'\xbaN{Nr\x00\x00\x00\x000N', # slice(None, {None: }) b'\xbaNN{Nr\x00\x00\x00\x000', # slice(None, None, {None: }) + b'\xfdN[\x01\x00\x00\x00r\x00\x00\x00\x000', # frozendict({None: []}) + b'\xfdN{Nr\x00\x00\x00\x0000', # frozendict({None: {None: }) # Direct self-references which cannot be created in Python. b'\xbe\x01\x00\x00\x00r\x00\x00\x00\x00', # frozenset({}) + b'\xfdNr\x00\x00\x00\x000', # frozendict({None: }) + b'\xfdr\x00\x00\x00\x00N0', # frozendict({: None}) b'\xbar\x00\x00\x00\x00NN', # slice(, None) b'\xbaNr\x00\x00\x00\x00N', # slice(None, ) b'\xbaNNr\x00\x00\x00\x00', # slice(None, None, ) @@ -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) diff --git a/Lib/test/test_profiling/test_sampling_profiler/test_dump.py b/Lib/test/test_profiling/test_sampling_profiler/test_dump.py new file mode 100644 index 000000000..7a0f8df38 --- /dev/null +++ b/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), + "", + None, + ) + ], + ) + ], + ) + ] + + output = format_stack_dump(stack_frames, pid=42, colorize=False) + + self.assertIn(' File "', output) + self.assertIn('", line 1, in ', 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), + "", + 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), + "", + 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()