summaryrefslogtreecommitdiff
path: root/lib/python2.7/test/test_threadsignals.py
diff options
context:
space:
mode:
Diffstat (limited to 'lib/python2.7/test/test_threadsignals.py')
-rw-r--r--lib/python2.7/test/test_threadsignals.py85
1 files changed, 0 insertions, 85 deletions
diff --git a/lib/python2.7/test/test_threadsignals.py b/lib/python2.7/test/test_threadsignals.py
deleted file mode 100644
index 2f7eb60..0000000
--- a/lib/python2.7/test/test_threadsignals.py
+++ /dev/null
@@ -1,85 +0,0 @@
-"""PyUnit testing that threads honor our signal semantics"""
-
-import unittest
-import signal
-import os
-import sys
-from test.test_support import run_unittest, import_module, reap_threads
-thread = import_module('thread')
-
-if sys.platform[:3] in ('win', 'os2') or sys.platform=='riscos':
- raise unittest.SkipTest, "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.
- """
- @reap_threads
- 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()