mirror of
https://github.com/python/cpython.git
synced 2025-07-08 03:45:36 +00:00
1bc535dc78
svn+ssh://pythondev@svn.python.org/python/branches/p3yk ........ r55329 | brett.cannon | 2007-05-14 16:36:56 -0700 (Mon, 14 May 2007) | 3 lines Implement the removal of tuple parameter unpacking (PEP 3113). Thanks, Tony Lownds for the patch. ........ r55331 | neal.norwitz | 2007-05-14 16:40:30 -0700 (Mon, 14 May 2007) | 1 line Update to use Python 3.0 ........ r55332 | brett.cannon | 2007-05-14 16:47:18 -0700 (Mon, 14 May 2007) | 2 lines Mention PEP 3113. And thanks to Tony Lownds for the PEP 3113 patch. ........ r55333 | neal.norwitz | 2007-05-14 16:57:06 -0700 (Mon, 14 May 2007) | 1 line Fix exception printing (no more exceptions module) ........ r55334 | neal.norwitz | 2007-05-14 17:11:10 -0700 (Mon, 14 May 2007) | 1 line Remove popen* functions from os ........ r55335 | neal.norwitz | 2007-05-14 18:03:38 -0700 (Mon, 14 May 2007) | 1 line Get rid of most of popen. There are still some uses I need to cleanup. ........ r55336 | neal.norwitz | 2007-05-14 21:11:34 -0700 (Mon, 14 May 2007) | 1 line Remove a few more remnants of the compiler package ........ r55337 | neal.norwitz | 2007-05-14 22:28:27 -0700 (Mon, 14 May 2007) | 1 line Get test_[cx]pickle working on 64-bit platforms (avoid overflow int/long) ........
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""PyUnit testing that threads honor our signal semantics"""
|
|
|
|
import unittest
|
|
import thread
|
|
import signal
|
|
import os
|
|
import sys
|
|
from test.test_support import run_unittest, TestSkipped
|
|
|
|
if sys.platform[:3] in ('win', 'os2') or sys.platform=='riscos':
|
|
raise TestSkipped, "Can't test signal on %s" % sys.platform
|
|
|
|
process_pid = os.getpid()
|
|
signalled_all=thread.allocate_lock()
|
|
|
|
|
|
def registerSignals(for_usr1, for_usr2, for_alrm):
|
|
usr1 = signal.signal(signal.SIGUSR1, for_usr1)
|
|
usr2 = signal.signal(signal.SIGUSR2, for_usr2)
|
|
alrm = signal.signal(signal.SIGALRM, for_alrm)
|
|
return usr1, usr2, alrm
|
|
|
|
|
|
# The signal handler. Just note that the signal occurred and
|
|
# from who.
|
|
def handle_signals(sig,frame):
|
|
signal_blackboard[sig]['tripped'] += 1
|
|
signal_blackboard[sig]['tripped_by'] = thread.get_ident()
|
|
|
|
# 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)
|
|
signalled_all.release()
|
|
|
|
class ThreadSignals(unittest.TestCase):
|
|
"""Test signal handling semantics of threads.
|
|
We spawn a thread, have the thread send two signals, and
|
|
wait for it to finish. Check that we got both signals
|
|
and that they were run by the main thread.
|
|
"""
|
|
def test_signals(self):
|
|
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:
|
|
signal.alarm(1)
|
|
signal.pause()
|
|
signal.alarm(0)
|
|
|
|
self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1)
|
|
self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'],
|
|
thread.get_ident())
|
|
self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped'], 1)
|
|
self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped_by'],
|
|
thread.get_ident())
|
|
signalled_all.release()
|
|
|
|
def spawnSignallingThread(self):
|
|
thread.start_new_thread(send_signals, ())
|
|
|
|
|
|
def test_main():
|
|
global signal_blackboard
|
|
|
|
signal_blackboard = { signal.SIGUSR1 : {'tripped': 0, 'tripped_by': 0 },
|
|
signal.SIGUSR2 : {'tripped': 0, 'tripped_by': 0 },
|
|
signal.SIGALRM : {'tripped': 0, 'tripped_by': 0 } }
|
|
|
|
oldsigs = registerSignals(handle_signals, handle_signals, handle_signals)
|
|
try:
|
|
run_unittest(ThreadSignals)
|
|
finally:
|
|
registerSignals(*oldsigs)
|
|
|
|
if __name__ == '__main__':
|
|
test_main()
|