mirror of
https://github.com/python/cpython.git
synced 2025-08-04 00:48:58 +00:00
[3.11] gh-71052: Use raise_signal
in ThreadSignals.test_signals
(GH-116423) (#116617)
gh-71052: Use `raise_signal` in `ThreadSignals.test_signals` (GH-116423)
Use `raise_signal` rather than `kill` in `ThreadSignals.test_signals`
(cherry picked from commit 34920f3691
)
Co-authored-by: Malcolm Smith <smith@chaquo.com>
This commit is contained in:
parent
6ae6d46e61
commit
88ed804ae7
1 changed files with 7 additions and 18 deletions
|
@ -32,39 +32,28 @@ def handle_signals(sig,frame):
|
|||
|
||||
# a function that will be spawned as a separate thread.
|
||||
def send_signals():
|
||||
os.kill(process_pid, signal.SIGUSR1)
|
||||
os.kill(process_pid, signal.SIGUSR2)
|
||||
# We use `raise_signal` rather than `kill` because:
|
||||
# * It verifies that a signal delivered to a background thread still has
|
||||
# its Python-level handler called on the main thread.
|
||||
# * It ensures the signal is handled before the thread exits.
|
||||
signal.raise_signal(signal.SIGUSR1)
|
||||
signal.raise_signal(signal.SIGUSR2)
|
||||
signalled_all.release()
|
||||
|
||||
|
||||
@threading_helper.requires_working_threading()
|
||||
@unittest.skipUnless(hasattr(signal, "alarm"), "test requires signal.alarm")
|
||||
class ThreadSignals(unittest.TestCase):
|
||||
|
||||
def test_signals(self):
|
||||
with threading_helper.wait_threads_exit():
|
||||
# Test signal handling semantics of threads.
|
||||
# We spawn a thread, have the thread send two signals, and
|
||||
# We spawn a thread, have the thread send itself two signals, and
|
||||
# wait for it to finish. Check that we got both signals
|
||||
# and that they were run by the main thread.
|
||||
signalled_all.acquire()
|
||||
self.spawnSignallingThread()
|
||||
signalled_all.acquire()
|
||||
|
||||
# the signals that we asked the kernel to send
|
||||
# will come back, but we don't know when.
|
||||
# (it might even be after the thread exits
|
||||
# and might be out of order.) If we haven't seen
|
||||
# the signals yet, send yet another signal and
|
||||
# wait for it return.
|
||||
if signal_blackboard[signal.SIGUSR1]['tripped'] == 0 \
|
||||
or signal_blackboard[signal.SIGUSR2]['tripped'] == 0:
|
||||
try:
|
||||
signal.alarm(1)
|
||||
signal.pause()
|
||||
finally:
|
||||
signal.alarm(0)
|
||||
|
||||
self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1)
|
||||
self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'],
|
||||
thread.get_ident())
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue