python · stratakis · Jul 8, 2026 · Jul 8, 2026
diff --git a/Lib/test/dtracedata/call_stack.stp b/Lib/test/dtracedata/call_stack.stp
@@ -10,7 +10,7 @@ function basename:string(path:string)
     return last_token;
 }
 
-probe process.mark("function__entry")
+probe @PYTHON_SYSTEMTAP_PROBE@("function__entry")
 {
     funcname = user_string($arg2);
 
@@ -19,7 +19,8 @@ probe process.mark("function__entry")
     }
 }
 
-probe process.mark("function__entry"), process.mark("function__return")
+probe @PYTHON_SYSTEMTAP_PROBE@("function__entry"),
+      @PYTHON_SYSTEMTAP_PROBE@("function__return")
 {
     filename = user_string($arg1);
     funcname = user_string($arg2);
@@ -31,7 +32,7 @@ probe process.mark("function__entry"), process.mark("function__return")
     }
 }
 
-probe process.mark("function__return")
+probe @PYTHON_SYSTEMTAP_PROBE@("function__return")
 {
     funcname = user_string($arg2);
 

diff --git a/Lib/test/dtracedata/gc.stp b/Lib/test/dtracedata/gc.stp
@@ -1,6 +1,6 @@
 global tracing
 
-probe process.mark("function__entry")
+probe @PYTHON_SYSTEMTAP_PROBE@("function__entry")
 {
     funcname = user_string($arg2);
 
@@ -9,14 +9,15 @@ probe process.mark("function__entry")
     }
 }
 
-probe process.mark("gc__start"), process.mark("gc__done")
+probe @PYTHON_SYSTEMTAP_PROBE@("gc__start"),
+      @PYTHON_SYSTEMTAP_PROBE@("gc__done")
 {
     if (tracing) {
         printf("%d\t%s:%ld\n", gettimeofday_us(), $$name, $arg1);
     }
 }
 
-probe process.mark("function__return")
+probe @PYTHON_SYSTEMTAP_PROBE@("function__return")
 {
     funcname = user_string($arg2);
 

diff --git a/Lib/test/test_dtrace.py b/Lib/test/test_dtrace.py
@@ -5,6 +5,7 @@
 import subprocess
 import sys
 import sysconfig
+import tempfile
 import types
 import unittest
 
@@ -24,6 +25,42 @@ def abspath(filename):
     return os.path.abspath(findfile(filename, subdir="dtracedata"))
 
 
+def get_probe_binary():
+    binary = sys.executable
+    if sysconfig.get_config_var("Py_ENABLE_SHARED"):
+        lib_dir = sysconfig.get_config_var("LIBDIR")
+        if not lib_dir or sysconfig.is_python_build():
+            lib_dir = os.path.abspath(os.path.dirname(sys.executable))
+
+        lib_names = []
+        for name in (
+            sysconfig.get_config_var("INSTSONAME"),
+            sysconfig.get_config_var("LDLIBRARY"),
+        ):
+            if name and name not in lib_names:
+                lib_names.append(name)
+
+        if lib_dir:
+            for name in lib_names:
+                libpython_path = os.path.join(lib_dir, name)
+                if os.path.exists(libpython_path):
+                    binary = libpython_path
+                    break
+
+    return binary
+
+
+def truncate_output(output, *, max_lines=3, max_chars=500):
+    lines = [line.strip() for line in output.splitlines() if line.strip()]
+    omitted = len(lines) - max_lines
+    output = "; ".join(lines[:max_lines])
+    if omitted > 0:
+        output += f"; ... ({omitted} lines omitted)"
+    if len(output) > max_chars:
+        output = output[:max_chars - 3] + "..."
+    return output
+
+
 def normalize_trace_output(output):
     """Normalize DTrace output for comparison.
 
@@ -94,7 +131,8 @@ def run_readelf(cmd):
         raise AssertionError(
             f"Command {' '.join(cmd)!r} failed "
             f"with exit code {proc.returncode}: "
-            f"stdout={stdout!r} stderr={stderr!r}"
+            f"stdout={truncate_output(stdout)!r} "
+            f"stderr={truncate_output(stderr)!r}"
         )
 
     return stdout
@@ -140,7 +178,8 @@ def trace(self, script_file, subcommand=None, *, timeout=None,
         if check_returncode and proc.returncode:
             raise AssertionError(
                 f"Command {' '.join(command)!r} failed "
-                f"with exit code {proc.returncode}: output={stdout!r}"
+                f"with exit code {proc.returncode}: "
+                f"output={truncate_output(stdout)!r}"
             )
         return stdout
 
@@ -165,7 +204,9 @@ def assert_usable(self):
             output = str(fnfe)
         if output != "probe: success":
             raise unittest.SkipTest(
-                "{}(1) failed: {}".format(self.COMMAND[0], output)
+                "{}(1) failed: {}".format(
+                    self.COMMAND[0], truncate_output(output)
+                )
             )
 
 
@@ -179,6 +220,43 @@ class DTraceBackend(TraceBackend):
 class SystemTapBackend(TraceBackend):
     EXTENSION = ".stp"
     COMMAND = ["stap", "-g"]
+    PROBE_PLACEHOLDER = "@PYTHON_SYSTEMTAP_PROBE@"
+
+    @staticmethod
+    def _quote_systemtap_string(value):
+        return value.replace("\\", "\\\\").replace('"', '\\"')
+
+    def _python_probe(self):
+        executable = self._quote_systemtap_string(sys.executable)
+        probe_binary = get_probe_binary()
+        if probe_binary != sys.executable:
+            probe_binary = self._quote_systemtap_string(probe_binary)
+            return f'process("{executable}").library("{probe_binary}").mark'
+        return f'process("{executable}").mark'
+
+    def _render_script(self, script_file):
+        with open(script_file) as script:
+            return script.read().replace(
+                self.PROBE_PLACEHOLDER, self._python_probe()
+            )
+
+    def trace(self, script_file, subcommand=None, *, timeout=None,
+              check_returncode=False):
+        with tempfile.NamedTemporaryFile(
+            mode="w", encoding="utf-8", suffix=self.EXTENSION, delete=False
+        ) as script:
+            script.write(self._render_script(script_file))
+            generated_script_file = script.name
+
+        try:
+            return super().trace(
+                generated_script_file,
+                subcommand,
+                timeout=timeout,
+                check_returncode=check_returncode,
+            )
+        finally:
+            os.unlink(generated_script_file)
 
 
 class BPFTraceBackend(TraceBackend):
@@ -272,7 +350,7 @@ def run_case(self, name, optimize_python=None):
             python_flags.extend(["-O"] * optimize_python)
 
         subcommand = [sys.executable] + python_flags + [python_file]
-        program = self.PROGRAMS[name].format(python=sys.executable)
+        program = self.PROGRAMS[name].format(python=get_probe_binary())
 
         try:
             proc = create_process_group(
@@ -290,7 +368,8 @@ def run_case(self, name, optimize_python=None):
 
         if proc.returncode != 0:
             raise AssertionError(
-                f"bpftrace failed with code {proc.returncode}:\n{stderr}"
+                f"bpftrace failed with code {proc.returncode}: "
+                f"{truncate_output(stderr)}"
             )
 
         stdout = self._filter_probe_rows(stdout)
@@ -311,7 +390,7 @@ def run_case(self, name, optimize_python=None):
 
     def assert_usable(self):
         # Check if bpftrace is available and can attach to USDT probes
-        program = f'usdt:{sys.executable}:python:function__entry {{ printf("probe: success\\n"); exit(); }}'
+        program = f'usdt:{get_probe_binary()}:python:function__entry {{ printf("probe: success\\n"); exit(); }}'
         try:
             proc = create_process_group(
                 ["bpftrace", "-e", program, "-c", f"{sys.executable} -c pass"],
@@ -329,12 +408,15 @@ def assert_usable(self):
         # Check for permission errors (bpftrace usually requires root)
         if proc.returncode != 0:
             raise unittest.SkipTest(
-                f"bpftrace(1) failed with code {proc.returncode}: {stderr}"
+                f"bpftrace(1) failed with code {proc.returncode}: "
+                f"{truncate_output(stderr)}"
             )
 
         if "probe: success" not in stdout:
             raise unittest.SkipTest(
-                f"bpftrace(1) failed: stdout={stdout!r} stderr={stderr!r}"
+                f"bpftrace(1) failed: "
+                f"stdout={truncate_output(stdout)!r} "
+                f"stderr={truncate_output(stderr)!r}"
             )
 
 
@@ -453,28 +535,7 @@ def get_readelf_version():
         return int(match.group(1)), int(match.group(2))
 
     def get_readelf_output(self):
-        binary = sys.executable
-        if sysconfig.get_config_var("Py_ENABLE_SHARED"):
-            lib_dir = sysconfig.get_config_var("LIBDIR")
-            if not lib_dir or sysconfig.is_python_build():
-                lib_dir = os.path.abspath(os.path.dirname(sys.executable))
-
-            lib_names = []
-            for name in (
-                sysconfig.get_config_var("INSTSONAME"),
-                sysconfig.get_config_var("LDLIBRARY"),
-            ):
-                if name and name not in lib_names:
-                    lib_names.append(name)
-
-            if lib_dir:
-                for name in lib_names:
-                    libpython_path = os.path.join(lib_dir, name)
-                    if os.path.exists(libpython_path):
-                        binary = libpython_path
-                        break
-
-        return run_readelf(["readelf", "-n", binary])
+        return run_readelf(["readelf", "-n", get_probe_binary()])
 
     def test_check_probes(self):
         readelf_output = self.get_readelf_output()