metadata

Author: maurycy

Lib/test/test_external_inspection.py

@@ -3804,5 +3804,170 @@ def test_get_stats_disabled_raises(self):
             client_socket.sendall(b"done")
 
 
+@requires_remote_subprocess_debugging()
+class TestMetadataDegradation(RemoteInspectionTestBase):
+    """Tests for graceful degradation of oversized code-object metadata."""
+
+    @contextmanager
+    def _running_target(self, script_body):
+        """Run a target script (socket handshake prepended), yield (process, socket)."""
+        port = find_unused_port()
+        script = (
+            textwrap.dedent(
+                f"""\
+                import time, socket
+                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+                sock.connect(('localhost', {port}))
+                """
+            )
+            + textwrap.dedent(script_body)
+        )
+
+        with os_helper.temp_dir() as work_dir:
+            script_dir = os.path.join(work_dir, "script_pkg")
+            os.mkdir(script_dir)
+
+            server_socket = _create_server_socket(port)
+            script_name = _make_test_script(script_dir, "script", script)
+            client_socket = None
+
+            try:
+                with _managed_subprocess([sys.executable, script_name]) as p:
+                    client_socket, _ = server_socket.accept()
+                    server_socket.close()
+                    server_socket = None
+                    yield p, client_socket
+            finally:
+                _cleanup_sockets(client_socket, server_socket)
+
+    def _sample_until_frame(self, pid, predicate):
+        """Sample until a frame matching predicate appears; return that frame."""
+        unwinder = RemoteUnwinder(pid, all_threads=True)
+        traces = _get_stack_trace_with_retry(
+            unwinder,
+            condition=lambda t: self._find_frame_in_trace(t, predicate)
+            is not None,
+        )
+        return self._find_frame_in_trace(traces, predicate)
+
+    @skip_if_not_supported
+    @unittest.skipIf(
+        sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
+        "Test only runs on Linux with process_vm_readv support",
+    )
+    def test_long_qualname_truncated_not_dropped(self):
+        """A qualname longer than 1024 chars is truncated with a marker
+        instead of failing the whole sample."""
+        script_body = """\
+            src = ("def " + "f" * 1100 + "():\\n"
+                   "    sock.sendall(b'ready')\\n"
+                   "    time.sleep(10_000)\\n")
+            ns = {"sock": sock, "time": time}
+            exec(src, ns)
+            ns["f" * 1100]()
+            """
+        with self._running_target(script_body) as (p, client_socket):
+            _wait_for_signal(client_socket, b"ready")
+            frame = self._sample_until_frame(
+                p.pid, lambda f: f.funcname.startswith("fff")
+            )
+            self.assertEqual(
+                frame.funcname, "f" * 1024 + "…(len=1100)"
+            )
+
+    @skip_if_not_supported
+    @unittest.skipIf(
+        sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
+        "Test only runs on Linux with process_vm_readv support",
+    )
+    def test_long_filename_truncated(self):
+        """A filename longer than 1024 chars is truncated with a marker
+        instead of failing the whole sample."""
+        script_body = """\
+            src = ("def g():\\n"
+                   "    sock.sendall(b'ready')\\n"
+                   "    time.sleep(10_000)\\n")
+            ns = {"sock": sock, "time": time}
+            exec(compile(src, "x" * 1500 + ".py", "exec"), ns)
+            ns["g"]()
+            """
+        with self._running_target(script_body) as (p, client_socket):
+            _wait_for_signal(client_socket, b"ready")
+            frame = self._sample_until_frame(
+                p.pid, lambda f: f.funcname == "g"
+            )
+            self.assertEqual(
+                frame.filename, "x" * 1024 + "…(len=1503)"
+            )
+
+    @skip_if_not_supported
+    @unittest.skipIf(
+        sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
+        "Test only runs on Linux with process_vm_readv support",
+    )
+    def test_large_linetable_keeps_line_numbers(self):
+        """A linetable larger than the old 4096-byte cap keeps line numbers."""
+        script_body = """\
+            body = "\\n".join("    x%d = %d" % (i, i) for i in range(4000))
+            src = ("def big():\\n" + body + "\\n"
+                   "    sock.sendall(b'ready')\\n"
+                   "    time.sleep(10_000)\\n")
+            ns = {"sock": sock, "time": time}
+            exec(src, ns)
+            sock.sendall(b"lt:%d\\n" % len(ns["big"].__code__.co_linetable))
+            ns["big"]()
+            """
+        with self._running_target(script_body) as (p, client_socket):
+            buffer = _wait_for_signal(client_socket, [b"lt:", b"ready"])
+            linetable_size = int(
+                buffer.partition(b"lt:")[2].partition(b"\n")[0]
+            )
+            self.assertGreater(linetable_size, 4096)
+
+            frame = self._sample_until_frame(
+                p.pid, lambda f: f.funcname == "big"
+            )
+            self.assertIsNotNone(
+                frame.location, "line info degraded for large linetable"
+            )
+            self.assertGreater(frame.location.lineno, 4000)
+
+    @skip_if_not_supported
+    @unittest.skipIf(
+        sys.platform == "linux" and not PROCESS_VM_READV_SUPPORTED,
+        "Test only runs on Linux with process_vm_readv support",
+    )
+    @unittest.skipIf(
+        sys.platform == "win32",
+        "Process death maps to ProcessLookupError only on POSIX platforms",
+    )
+    def test_dead_process_raises_not_degrades(self):
+        """Death of the target raises ProcessLookupError instead of
+        degrading to synthetic frames."""
+        script_body = """\
+            sock.sendall(b"ready")
+            time.sleep(10_000)
+            """
+        with self._running_target(script_body) as (p, client_socket):
+            _wait_for_signal(client_socket, b"ready")
+            unwinder = RemoteUnwinder(p.pid, all_threads=True)
+            _get_stack_trace_with_retry(unwinder)
+
+            p.kill()
+            p.wait()
+
+            for _ in busy_retry(SHORT_TIMEOUT, error=False):
+                try:
+                    unwinder.get_stack_trace()
+                except ProcessLookupError:
+                    break
+                except RuntimeError:
+                    continue
+            else:
+                self.fail(
+                    "ProcessLookupError never raised for dead process"
+                )
+
+
 if __name__ == "__main__":
     unittest.main()

Lib/test/test_profiling/test_sampling_profiler/test_binary_format.py

@@ -628,6 +628,41 @@ def test_same_line_different_columns(self):
         collector, count = self.roundtrip(samples)
         self.assertEqual(count, 3)
 
+    def test_synthetic_frames_roundtrip(self):
+        """Degraded/sentinel frames (location=None) survive the binary format."""
+        frames = [
+            FrameInfo(("~", None, name, None))
+            for name in _remote_debugging.SYNTHETIC_FRAME_NAMES
+        ]
+        frames.append(FrameInfo(("app.py", None, "<unknown function>", None)))
+        frames.append(FrameInfo(("<unknown file>", None, "real_func", None)))
+        samples = [[make_interpreter(0, [make_thread(1, frames)])]]
+
+        collector, count = self.roundtrip(samples)
+        self.assertEqual(count, 1)
+        self.assert_samples_equal(samples, collector)
+
+        class RawFrameCollector:
+            def __init__(self):
+                self.frames = []
+
+            def collect(self, stack_frames, timestamps_us):
+                for interp in stack_frames:
+                    for thread in interp.threads:
+                        self.frames.extend(thread.frame_info)
+
+            def export(self, filename):
+                pass
+
+        raw = RawFrameCollector()
+        with BinaryReader(self.create_binary_file(samples)) as reader:
+            reader.replay_samples(raw)
+        self.assertEqual(len(raw.frames), len(frames))
+        for original, replayed in zip(frames, raw.frames):
+            self.assertEqual(replayed.filename, original.filename)
+            self.assertEqual(replayed.funcname, original.funcname)
+            self.assertIsNone(replayed.location)
+
 
 class TestBinaryEdgeCases(BinaryFormatTestBase):
     """Tests for edge cases in binary format."""

Lib/test/test_profiling/test_sampling_profiler/test_collectors.py

@@ -20,12 +20,17 @@
     from profiling.sampling.jsonl_collector import JsonlCollector
     from profiling.sampling.gecko_collector import GeckoCollector
     from profiling.sampling.heatmap_collector import _TemplateLoader
-    from profiling.sampling.collector import extract_lineno, normalize_location
+    from profiling.sampling.collector import (
+        extract_lineno,
+        is_synthetic,
+        normalize_location,
+    )
     from profiling.sampling.opcode_utils import get_opcode_info, format_opcode
     from profiling.sampling.constants import (
         PROFILING_MODE_WALL,
         PROFILING_MODE_CPU,
         DEFAULT_LOCATION,
+        SYNTHETIC_FRAME_NAMES,
     )
     from _remote_debugging import (
         THREAD_STATUS_HAS_GIL,
@@ -3204,3 +3209,49 @@ def test_jsonl_collector_filters_internal_frames(self):
 
         for path in paths:
             self.assertNotIn("_sync_coordinator", path)
+
+
+class TestSyntheticFrames(unittest.TestCase):
+    """Tests for SYNTHETIC_FRAME_NAMES and the is_synthetic predicate."""
+
+    def test_synthetic_frame_names_exact_contract(self):
+        expected = (
+            "<GC>",
+            "<native>",
+            "<unknown function>",
+            "<unknown file>",
+            "<unreadable frame>",
+        )
+        self.assertEqual(_remote_debugging.SYNTHETIC_FRAME_NAMES, expected)
+
+    def test_synthetic_frame_names_parity_with_fallback(self):
+        self.assertEqual(
+            frozenset(_remote_debugging.SYNTHETIC_FRAME_NAMES),
+            SYNTHETIC_FRAME_NAMES,
+        )
+
+    def test_is_synthetic_sentinel_funcnames(self):
+        for name in SYNTHETIC_FRAME_NAMES:
+            with self.subTest(name=name):
+                self.assertTrue(is_synthetic(("app.py", None, name, None)))
+
+    def test_is_synthetic_tilde_filename(self):
+        self.assertTrue(is_synthetic(("~", None, "anything", None)))
+
+    def test_is_synthetic_rejects_real_angle_names(self):
+        for name in ("<module>", "<lambda>", "<listcomp>", "<genexpr>"):
+            with self.subTest(name=name):
+                self.assertFalse(
+                    is_synthetic(("app.py", (1, 1, 0, 0), name, None))
+                )
+
+    def test_is_synthetic_unknown_file_with_real_funcname(self):
+        self.assertFalse(
+            is_synthetic(("<unknown file>", None, "real_func", None))
+        )
+
+    def test_is_synthetic_frameinfo_object(self):
+        self.assertTrue(
+            is_synthetic(MockFrameInfo("app.py", 1, "<unknown function>"))
+        )
+        self.assertFalse(is_synthetic(MockFrameInfo("app.py", 1, "work")))