66import subprocess
77import sys
88import sysconfig
9+ import tempfile
910import types
1011import unittest
1112
1213from test import support
1314from test .support import findfile , MS_WINDOWS
15+ from test .support import os_helper
1416
1517
1618if not support .has_subprocess_support :
@@ -25,6 +27,31 @@ def abspath(filename):
2527 return os .path .abspath (findfile (filename , subdir = "dtracedata" ))
2628
2729
30+ def get_probe_binary ():
31+ binary = sys .executable
32+ if sysconfig .get_config_var ("Py_ENABLE_SHARED" ):
33+ lib_dir = sysconfig .get_config_var ("LIBDIR" )
34+ if not lib_dir or sysconfig .is_python_build ():
35+ lib_dir = os .path .abspath (os .path .dirname (sys .executable ))
36+
37+ lib_names = []
38+ for name in (
39+ sysconfig .get_config_var ("INSTSONAME" ),
40+ sysconfig .get_config_var ("LDLIBRARY" ),
41+ ):
42+ if name and name not in lib_names :
43+ lib_names .append (name )
44+
45+ if lib_dir :
46+ for name in lib_names :
47+ libpython_path = os .path .join (lib_dir , name )
48+ if os .path .exists (libpython_path ):
49+ binary = libpython_path
50+ break
51+
52+ return binary
53+
54+
2855def normalize_trace_output (output ):
2956 """Normalize DTrace output for comparison.
3057
@@ -180,6 +207,45 @@ class DTraceBackend(TraceBackend):
180207class SystemTapBackend (TraceBackend ):
181208 EXTENSION = ".stp"
182209 COMMAND = ["stap" , "-g" ]
210+ PROBE_PLACEHOLDER = "@PYTHON_SYSTEMTAP_PROBE@"
211+
212+ @staticmethod
213+ def quote_systemtap_string (value ):
214+ return value .replace ("\\ " , "\\ \\ " ).replace ('"' , '\\ "' )
215+
216+ def python_probe (self ):
217+ executable = self .quote_systemtap_string (sys .executable )
218+ probe_binary = get_probe_binary ()
219+ if probe_binary == sys .executable :
220+ return f'process("{ executable } ").mark'
221+
222+ # Python built with --enable-shared
223+ probe_binary = self .quote_systemtap_string (probe_binary )
224+ return f'process("{ executable } ").library("{ probe_binary } ").mark'
225+
226+ def render_script (self , filename ):
227+ with open (filename ) as fp :
228+ script = fp .read ()
229+
230+ return script .replace (self .PROBE_PLACEHOLDER , self .python_probe ())
231+
232+ def trace (self , script_file , subcommand = None , * , timeout = None ,
233+ check_returncode = False ):
234+ with tempfile .NamedTemporaryFile (
235+ mode = "w" , encoding = "utf-8" , suffix = self .EXTENSION , delete = False
236+ ) as script :
237+ script .write (self .render_script (script_file ))
238+ generated_script_file = script .name
239+
240+ try :
241+ return super ().trace (
242+ generated_script_file ,
243+ subcommand ,
244+ timeout = timeout ,
245+ check_returncode = check_returncode ,
246+ )
247+ finally :
248+ os_helper .unlink (generated_script_file )
183249
184250
185251class BPFTraceBackend (TraceBackend ):
@@ -273,7 +339,7 @@ def run_case(self, name, optimize_python=None):
273339 python_flags .extend (["-O" ] * optimize_python )
274340
275341 subcommand = [sys .executable ] + python_flags + [python_file ]
276- program = self .PROGRAMS [name ].format (python = sys . executable )
342+ program = self .PROGRAMS [name ].format (python = get_probe_binary () )
277343
278344 try :
279345 proc = create_process_group (
@@ -312,7 +378,7 @@ def run_case(self, name, optimize_python=None):
312378
313379 def assert_usable (self ):
314380 # Check if bpftrace is available and can attach to USDT probes
315- program = f'usdt:{ sys . executable } :python:function__entry {{ printf("probe: success\\ n"); exit(); }}'
381+ program = f'usdt:{ get_probe_binary () } :python:function__entry {{ printf("probe: success\\ n"); exit(); }}'
316382 try :
317383 proc = create_process_group (
318384 ["bpftrace" , "-e" , program , "-c" ,
@@ -455,28 +521,7 @@ def get_readelf_version():
455521 return int (match .group (1 )), int (match .group (2 ))
456522
457523 def get_readelf_output (self ):
458- binary = sys .executable
459- if sysconfig .get_config_var ("Py_ENABLE_SHARED" ):
460- lib_dir = sysconfig .get_config_var ("LIBDIR" )
461- if not lib_dir or sysconfig .is_python_build ():
462- lib_dir = os .path .abspath (os .path .dirname (sys .executable ))
463-
464- lib_names = []
465- for name in (
466- sysconfig .get_config_var ("INSTSONAME" ),
467- sysconfig .get_config_var ("LDLIBRARY" ),
468- ):
469- if name and name not in lib_names :
470- lib_names .append (name )
471-
472- if lib_dir :
473- for name in lib_names :
474- libpython_path = os .path .join (lib_dir , name )
475- if os .path .exists (libpython_path ):
476- binary = libpython_path
477- break
478-
479- return run_readelf (["readelf" , "-n" , binary ])
524+ return run_readelf (["readelf" , "-n" , get_probe_binary ()])
480525
481526 def test_check_probes (self ):
482527 readelf_output = self .get_readelf_output ()
0 commit comments