bpo-35568: add 'raise_signal' function (GH-11335)
As in title, expose C `raise` function as `raise_function` in `signal` module. Also drop existing `raise_signal` in `_testcapi` module and replace all usages with new function.
https://bugs.python.org/issue35568
diff --git a/Lib/test/test_asyncio/test_windows_events.py b/Lib/test/test_asyncio/test_windows_events.py
index a200a8a..05f8515 100644
--- a/Lib/test/test_asyncio/test_windows_events.py
+++ b/Lib/test/test_asyncio/test_windows_events.py
@@ -45,7 +45,7 @@
def SIGINT_after_delay():
time.sleep(1)
- _testcapi.raise_signal(signal.SIGINT)
+ signal.raise_signal(signal.SIGINT)
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
l = asyncio.get_event_loop()
diff --git a/Lib/test/test_faulthandler.py b/Lib/test/test_faulthandler.py
index 6aee22b..f0be918 100644
--- a/Lib/test/test_faulthandler.py
+++ b/Lib/test/test_faulthandler.py
@@ -198,14 +198,13 @@
@skip_segfault_on_android
def test_sigbus(self):
self.check_fatal_error("""
- import _testcapi
import faulthandler
import signal
faulthandler.enable()
- _testcapi.raise_signal(signal.SIGBUS)
+ signal.raise_signal(signal.SIGBUS)
""",
- 6,
+ 5,
'Bus error')
@unittest.skipIf(_testcapi is None, 'need _testcapi')
@@ -213,14 +212,13 @@
@skip_segfault_on_android
def test_sigill(self):
self.check_fatal_error("""
- import _testcapi
import faulthandler
import signal
faulthandler.enable()
- _testcapi.raise_signal(signal.SIGILL)
+ signal.raise_signal(signal.SIGILL)
""",
- 6,
+ 5,
'Illegal instruction')
def test_fatal_error(self):
diff --git a/Lib/test/test_posix.py b/Lib/test/test_posix.py
index 86c04b9..d7e512c 100644
--- a/Lib/test/test_posix.py
+++ b/Lib/test/test_posix.py
@@ -1596,8 +1596,8 @@
'need signal.pthread_sigmask()')
def test_setsigmask(self):
code = textwrap.dedent("""\
- import _testcapi, signal
- _testcapi.raise_signal(signal.SIGUSR1)""")
+ import signal
+ signal.raise_signal(signal.SIGUSR1)""")
pid = posix.posix_spawn(
sys.executable,
@@ -1627,8 +1627,8 @@
def test_setsigdef(self):
original_handler = signal.signal(signal.SIGUSR1, signal.SIG_IGN)
code = textwrap.dedent("""\
- import _testcapi, signal
- _testcapi.raise_signal(signal.SIGUSR1)""")
+ import signal
+ signal.raise_signal(signal.SIGUSR1)""")
try:
pid = posix.posix_spawn(
sys.executable,
diff --git a/Lib/test/test_regrtest.py b/Lib/test/test_regrtest.py
index a674583..6193776 100644
--- a/Lib/test/test_regrtest.py
+++ b/Lib/test/test_regrtest.py
@@ -26,9 +26,8 @@
ROOT_DIR = os.path.abspath(os.path.normpath(ROOT_DIR))
TEST_INTERRUPTED = textwrap.dedent("""
- from signal import SIGINT
+ from signal import SIGINT, raise_signal
try:
- from _testcapi import raise_signal
raise_signal(SIGINT)
except ImportError:
import os
diff --git a/Lib/test/test_signal.py b/Lib/test/test_signal.py
index b10faa0..2a6217e 100644
--- a/Lib/test/test_signal.py
+++ b/Lib/test/test_signal.py
@@ -1,3 +1,4 @@
+import errno
import os
import random
import signal
@@ -254,7 +255,7 @@
signal.set_wakeup_fd(r)
try:
with captured_stderr() as err:
- _testcapi.raise_signal(signal.SIGALRM)
+ signal.raise_signal(signal.SIGALRM)
except ZeroDivisionError:
# An ignored exception should have been printed out on stderr
err = err.getvalue()
@@ -348,10 +349,9 @@
def test_signum(self):
self.check_wakeup("""def test():
- import _testcapi
signal.signal(signal.SIGUSR1, handler)
- _testcapi.raise_signal(signal.SIGUSR1)
- _testcapi.raise_signal(signal.SIGALRM)
+ signal.raise_signal(signal.SIGUSR1)
+ signal.raise_signal(signal.SIGALRM)
""", signal.SIGUSR1, signal.SIGALRM)
@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
@@ -365,8 +365,8 @@
signal.signal(signum2, handler)
signal.pthread_sigmask(signal.SIG_BLOCK, (signum1, signum2))
- _testcapi.raise_signal(signum1)
- _testcapi.raise_signal(signum2)
+ signal.raise_signal(signum1)
+ 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)
@@ -396,7 +396,7 @@
write.setblocking(False)
signal.set_wakeup_fd(write.fileno())
- _testcapi.raise_signal(signum)
+ signal.raise_signal(signum)
data = read.recv(1)
if not data:
@@ -445,7 +445,7 @@
write.close()
with captured_stderr() as err:
- _testcapi.raise_signal(signum)
+ signal.raise_signal(signum)
err = err.getvalue()
if ('Exception ignored when trying to {action} to the signal wakeup fd'
@@ -519,7 +519,7 @@
signal.set_wakeup_fd(write.fileno())
with captured_stderr() as err:
- _testcapi.raise_signal(signum)
+ signal.raise_signal(signum)
err = err.getvalue()
if msg not in err:
@@ -530,7 +530,7 @@
signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=True)
with captured_stderr() as err:
- _testcapi.raise_signal(signum)
+ signal.raise_signal(signum)
err = err.getvalue()
if msg not in err:
@@ -541,7 +541,7 @@
signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=False)
with captured_stderr() as err:
- _testcapi.raise_signal(signum)
+ signal.raise_signal(signum)
err = err.getvalue()
if err != "":
@@ -553,7 +553,7 @@
signal.set_wakeup_fd(write.fileno())
with captured_stderr() as err:
- _testcapi.raise_signal(signum)
+ signal.raise_signal(signum)
err = err.getvalue()
if msg not in err:
@@ -1214,6 +1214,38 @@
# Python handler
self.assertEqual(len(sigs), N, "Some signals were lost")
+class RaiseSignalTest(unittest.TestCase):
+
+ def test_sigint(self):
+ try:
+ signal.raise_signal(signal.SIGINT)
+ self.fail("Expected KeyInterrupt")
+ except KeyboardInterrupt:
+ pass
+
+ @unittest.skipIf(sys.platform != "win32", "Windows specific test")
+ def test_invalid_argument(self):
+ try:
+ SIGHUP = 1 # not supported on win32
+ signal.raise_signal(SIGHUP)
+ self.fail("OSError (Invalid argument) expected")
+ except OSError as e:
+ if e.errno == errno.EINVAL:
+ pass
+ else:
+ raise
+
+ def test_handler(self):
+ is_ok = False
+ def handler(a, b):
+ nonlocal is_ok
+ is_ok = True
+ old_signal = signal.signal(signal.SIGINT, handler)
+ self.addCleanup(signal.signal, signal.SIGINT, old_signal)
+
+ signal.raise_signal(signal.SIGINT)
+ self.assertTrue(is_ok)
+
def tearDownModule():
support.reap_children()