Stowage/cpython
2
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2026-06-28 03:41:13 +00:00
Code Issues Projects Releases Packages Wiki Activity

[3.15] gh-151496: Use process groups in TraceBackend in test_dtrace (GH-152039) (#152082)

Browse source 
gh-151496: Use process groups in TraceBackend in test_dtrace (GH-152039)

Run the generic DTrace/SystemTap commands in a new process group and
terminate the whole group on timeout.

This prevents a forked tracer child from keeping stdout/stderr pipes
open after the direct tracer process is killed.
(cherry picked from commit 1785f4b35f)

Co-authored-by: stratakis <cstratak@redhat.com>
This commit is contained in:
Miss Islington (bot) 2026-06-24 16:15:43 +02:00 • committed by GitHub
parent 8146fa4727
commit c93039d453
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 24 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

35
Lib/test/test_dtrace.py
View file

@ -75,10 +75,13 @@ class TraceBackend:
COMMAND_ARGS = []
def run_case(self, name, optimize_python=None):
actual_output = normalize_trace_output(self.trace_python(
script_file=abspath(name + self.EXTENSION),
python_file=abspath(name + ".py"),
optimize_python=optimize_python))
try:
actual_output = normalize_trace_output(self.trace_python(
script_file=abspath(name + self.EXTENSION),
python_file=abspath(name + ".py"),
optimize_python=optimize_python))
except subprocess.TimeoutExpired:
raise AssertionError(f"{self.COMMAND[0]} timed out")
with open(abspath(name + self.EXTENSION + ".expected")) as f:
expected_output = f.read().rstrip()
@ -91,12 +94,17 @@ def generate_trace_command(self, script_file, subcommand=None):
command += ["-c", subcommand]
return command
def trace(self, script_file, subcommand=None):
def trace(self, script_file, subcommand=None, *, timeout=None):
command = self.generate_trace_command(script_file, subcommand)
stdout, _ = subprocess.Popen(command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True).communicate()
proc = create_process_group(command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
universal_newlines=True)
try:
stdout, _ = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
kill_process_group(proc)
raise
return stdout
def trace_python(self, script_file, python_file, optimize_python=None):
@ -104,12 +112,17 @@ def trace_python(self, script_file, python_file, optimize_python=None):
if optimize_python:
python_flags.extend(["-O"] * optimize_python)
subcommand = " ".join([sys.executable] + python_flags + [python_file])
return self.trace(script_file, subcommand)
return self.trace(script_file, subcommand, timeout=60)
def assert_usable(self):
try:
output = self.trace(abspath("assert_usable" + self.EXTENSION))
output = self.trace(abspath("assert_usable" + self.EXTENSION),
timeout=10)
output = output.strip()
except subprocess.TimeoutExpired:
raise unittest.SkipTest(
f"{self.COMMAND[0]} timed out during usability check"
)
except (FileNotFoundError, NotADirectoryError, PermissionError) as fnfe:
output = str(fnfe)
if output != "probe: success":