Skip to content

gh-152548: Convert the test_signal wakeup tests to @runInSubprocess() #152639

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
serhiy-storchaka wants to merge 1 commit into
base: main
Choose a base branch
from
+87 −146
Open

gh-152548: Convert the test_signal wakeup tests to @runInSubprocess() #152639

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
233 changes: 87 additions & 146 deletions Lib/test/test_signal.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
import time
import unittest
from test import support
from test.support import isolation
from test.support import (
force_not_colorized, is_apple, is_apple_mobile, os_helper, threading_helper
)
Expand Down Expand Up @@ -312,59 +313,51 @@ def test_set_wakeup_fd_blocking(self):

@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
class WakeupSignalTests(unittest.TestCase):
@unittest.skipIf(_testcapi is None, 'need _testcapi')
def check_wakeup(self, test_body, *signals, ordered=True):
# use a subprocess to have only one thread
code = """if 1:
import _testcapi
import os
import signal
def check_wakeup(self, test, *signals, ordered=True):
# The caller is decorated with @runInSubprocess(), so this runs in a fresh
# subprocess that has only one thread for the signal to be delivered to.
import struct

signals = {!r}
signals = tuple(int(s) for s in signals)

def handler(signum, frame):
pass

def check_signum(signals):
data = os.read(read, len(signals)+1)
raised = struct.unpack('%uB' % len(data), data)
if not {!r}:
raised = set(raised)
signals = set(signals)
if raised != signals:
raise Exception("%r != %r" % (raised, signals))

{}

signal.signal(signal.SIGALRM, handler)
read, write = os.pipe()
os.set_blocking(write, False)
signal.set_wakeup_fd(write)

test()
check_signum(signals)
test(read)

data = os.read(read, len(signals) + 1)
raised = struct.unpack('%uB' % len(data), data)
if ordered:
self.assertEqual(raised, signals)
else:
self.assertEqual(set(raised), set(signals))

os.close(read)
os.close(write)
""".format(tuple(map(int, signals)), ordered, test_body)

assert_python_ok('-c', code)

@unittest.skipIf(_testcapi is None, 'need _testcapi')
@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
@force_not_colorized
@isolation.runInSubprocess()
def test_wakeup_write_error(self):
# Issue #16105: write() errors in the C signal handler should not
# pass silently.
# Use a subprocess to have only one thread.
code = """if 1:
import _testcapi
import errno
import os
import signal
import sys
from test.support import captured_stderr
# Use @runInSubprocess() to run in a subprocess with only one thread.
r, w = os.pipe()
try:
os.write(r, b'x')
except OSError:
pass
else:
self.skipTest("OS doesn't report write() error on the read end of a pipe")
finally:
os.close(r)
os.close(w)

def handler(signum, frame):
1/0
Expand All @@ -375,38 +368,23 @@ def handler(signum, frame):

# Set wakeup_fd a read-only file descriptor to trigger the error
signal.set_wakeup_fd(r)
try:
with captured_stderr() as err:
with support.captured_stderr() as err:
with self.assertRaises(ZeroDivisionError):
signal.raise_signal(signal.SIGALRM)
except ZeroDivisionError:
# An ignored exception should have been printed out on stderr
err = err.getvalue()
if ('Exception ignored while trying to write to the signal wakeup fd'
not in err):
raise AssertionError(err)
if ('OSError: [Errno %d]' % errno.EBADF) not in err:
raise AssertionError(err)
else:
raise AssertionError("ZeroDivisionError not raised")
# An ignored exception should have been printed out on stderr
err = err.getvalue()
self.assertIn(
'Exception ignored while trying to write to the signal wakeup fd',
err)
self.assertIn('OSError: [Errno %d]' % errno.EBADF, err)

os.close(r)
os.close(w)
"""
r, w = os.pipe()
try:
os.write(r, b'x')
except OSError:
pass
else:
self.skipTest("OS doesn't report write() error on the read end of a pipe")
finally:
os.close(r)
os.close(w)

assert_python_ok('-c', code)

@unittest.skipIf(_testcapi is None, 'need _testcapi')
@isolation.runInSubprocess()
def test_wakeup_fd_early(self):
self.check_wakeup("""def test():
def test(read):
import select
import time

Expand All @@ -429,18 +407,19 @@ def handler(signum, frame):
except InterruptSelect:
pass
else:
raise Exception("select() was not interrupted")
self.fail("select() was not interrupted")

before_time = time.monotonic()
select.select([read], [], [], TIMEOUT_FULL)
after_time = time.monotonic()
dt = after_time - before_time
if dt >= TIMEOUT_HALF:
raise Exception("%s >= %s" % (dt, TIMEOUT_HALF))
""", signal.SIGALRM)
self.assertLess(dt, TIMEOUT_HALF)
self.check_wakeup(test, signal.SIGALRM)

@unittest.skipIf(_testcapi is None, 'need _testcapi')
@isolation.runInSubprocess()
def test_wakeup_fd_during(self):
self.check_wakeup("""def test():
def test(read):
import select
import time

Expand All @@ -462,27 +441,34 @@ def handler(signum, frame):
except InterruptSelect:
pass
else:
raise Exception("select() was not interrupted")
self.fail("select() was not interrupted")
after_time = time.monotonic()
dt = after_time - before_time
if dt >= TIMEOUT_HALF:
raise Exception("%s >= %s" % (dt, TIMEOUT_HALF))
""", signal.SIGALRM)
self.assertLess(dt, TIMEOUT_HALF)
self.check_wakeup(test, signal.SIGALRM)

@unittest.skipIf(_testcapi is None, 'need _testcapi')
@isolation.runInSubprocess()
def test_signum(self):
self.check_wakeup("""def test():
def test(read):
def handler(signum, frame):
pass
signal.signal(signal.SIGUSR1, handler)
signal.raise_signal(signal.SIGUSR1)
signal.raise_signal(signal.SIGALRM)
""", signal.SIGUSR1, signal.SIGALRM)
self.check_wakeup(test, signal.SIGUSR1, signal.SIGALRM)

@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
'need signal.pthread_sigmask()')
@unittest.skipIf(_testcapi is None, 'need _testcapi')
@isolation.runInSubprocess()
def test_pending(self):
self.check_wakeup("""def test():
def test(read):
signum1 = signal.SIGUSR1
signum2 = signal.SIGUSR2

def handler(signum, frame):
pass
signal.signal(signum1, handler)
signal.signal(signum2, handler)

Expand All @@ -491,20 +477,17 @@ def test_pending(self):
signal.raise_signal(signum2)
# Unblocking the 2 signals calls the C signal handler twice
signal.pthread_sigmask(signal.SIG_UNBLOCK, (signum1, signum2))
""", signal.SIGUSR1, signal.SIGUSR2, ordered=False)
self.check_wakeup(test, signal.SIGUSR1, signal.SIGUSR2, ordered=False)


@unittest.skipUnless(hasattr(socket, 'socketpair'), 'need socket.socketpair')
class WakeupSocketSignalTests(unittest.TestCase):

@unittest.skipIf(_testcapi is None, 'need _testcapi')
@isolation.runInSubprocess()
def test_socket(self):
# use a subprocess to have only one thread
code = """if 1:
import signal
import socket
# Use @runInSubprocess() to run in a subprocess with only one thread.
import struct
import _testcapi

signum = signal.SIGINT
signals = (signum,)
Expand All @@ -521,33 +504,21 @@ def handler(signum, frame):
signal.raise_signal(signum)

data = read.recv(1)
if not data:
raise Exception("no signum written")
self.assertTrue(data, "no signum written")
raised = struct.unpack('B', data)
if raised != signals:
raise Exception("%r != %r" % (raised, signals))
self.assertEqual(raised, signals)

read.close()
write.close()
"""

assert_python_ok('-c', code)

@unittest.skipIf(_testcapi is None, 'need _testcapi')
@isolation.runInSubprocess()
def test_send_error(self):
# Use a subprocess to have only one thread.
# Use @runInSubprocess() to run in a subprocess with only one thread.
if os.name == 'nt':
action = 'send'
else:
action = 'write'
code = """if 1:
import errno
import signal
import socket
import sys
import time
import _testcapi
from test.support import captured_stderr

signum = signal.SIGINT

Expand All @@ -566,31 +537,22 @@ def handler(signum, frame):
read.close()
write.close()

with captured_stderr() as err:
with support.captured_stderr() as err:
signal.raise_signal(signum)

err = err.getvalue()
if ('Exception ignored while trying to {action} to the signal wakeup fd'
not in err):
raise AssertionError(err)
""".format(action=action)
assert_python_ok('-c', code)
self.assertIn(
'Exception ignored while trying to %s to the signal wakeup fd'
% action,
err.getvalue())

@unittest.skipIf(_testcapi is None, 'need _testcapi')
@isolation.runInSubprocess()
def test_warn_on_full_buffer(self):
# Use a subprocess to have only one thread.
# Use @runInSubprocess() to run in a subprocess with only one thread.
if os.name == 'nt':
action = 'send'
else:
action = 'write'
code = """if 1:
import errno
import signal
import socket
import sys
import time
import _testcapi
from test.support import captured_stderr

signum = signal.SIGINT

Expand Down Expand Up @@ -627,67 +589,46 @@ def handler(signum, frame):
except (BlockingIOError, TimeoutError):
pass

print(f"%s bytes written into the socketpair" % written, flush=True)

write.setblocking(False)
try:
with self.assertRaises(BlockingIOError,
msg="%s bytes failed to fill the socketpair "
"buffer" % written):
write.send(b"x")
except BlockingIOError:
# The socketpair buffer seems full
pass
else:
raise AssertionError("%s bytes failed to fill the socketpair "
"buffer" % written)

# By default, we get a warning when a signal arrives
msg = ('Exception ignored while trying to {action} '
'to the signal wakeup fd')
msg = ('Exception ignored while trying to %s '
'to the signal wakeup fd' % action)
signal.set_wakeup_fd(write.fileno())

with captured_stderr() as err:
with support.captured_stderr() as err:
signal.raise_signal(signum)

err = err.getvalue()
if msg not in err:
raise AssertionError("first set_wakeup_fd() test failed, "
"stderr: %r" % err)
self.assertIn(msg, err.getvalue(),
"first set_wakeup_fd() test failed")

# And also if warn_on_full_buffer=True
signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=True)

with captured_stderr() as err:
with support.captured_stderr() as err:
signal.raise_signal(signum)

err = err.getvalue()
if msg not in err:
raise AssertionError("set_wakeup_fd(warn_on_full_buffer=True) "
"test failed, stderr: %r" % err)
self.assertIn(msg, err.getvalue(),
"set_wakeup_fd(warn_on_full_buffer=True) test failed")

# But not if warn_on_full_buffer=False
signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=False)

with captured_stderr() as err:
with support.captured_stderr() as err:
signal.raise_signal(signum)

err = err.getvalue()
if err != "":
raise AssertionError("set_wakeup_fd(warn_on_full_buffer=False) "
"test failed, stderr: %r" % err)
self.assertEqual(err.getvalue(), "",
"set_wakeup_fd(warn_on_full_buffer=False) test failed")

# And then check the default again, to make sure warn_on_full_buffer
# settings don't leak across calls.
signal.set_wakeup_fd(write.fileno())

with captured_stderr() as err:
with support.captured_stderr() as err:
signal.raise_signal(signum)

err = err.getvalue()
if msg not in err:
raise AssertionError("second set_wakeup_fd() test failed, "
"stderr: %r" % err)

""".format(action=action)
assert_python_ok('-c', code)
self.assertIn(msg, err.getvalue(),
"second set_wakeup_fd() test failed")


@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
Expand Down
Loading