test_threadsignals.py 2.92 KB
Newer Older
1 2 3 4 5
"""PyUnit testing that threads honor our signal semantics"""

import unittest
import signal
import os
Fred Drake's avatar
Fred Drake committed
6
import sys
7
from test.test_support import run_unittest, import_module, reap_threads
8
thread = import_module('thread')
9

10
if sys.platform[:3] in ('win', 'os2') or sys.platform=='riscos':
11
    raise unittest.SkipTest, "Can't test signal on %s" % sys.platform
12

13 14 15 16
process_pid = os.getpid()
signalled_all=thread.allocate_lock()


17
def registerSignals(for_usr1, for_usr2, for_alrm):
18 19 20 21 22 23
    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


24
# The signal handler. Just note that the signal occurred and
25 26
# from who.
def handle_signals(sig,frame):
Tim Peters's avatar
Tim Peters committed
27
    signal_blackboard[sig]['tripped'] += 1
28 29 30 31 32 33 34 35 36 37 38 39 40 41
    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.
    """
42
    @reap_threads
43 44 45 46 47 48 49 50 51 52
    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.
53
        if signal_blackboard[signal.SIGUSR1]['tripped'] == 0 \
54
           or signal_blackboard[signal.SIGUSR2]['tripped'] == 0:
Tim Peters's avatar
Tim Peters committed
55 56 57
            signal.alarm(1)
            signal.pause()
            signal.alarm(0)
58 59

        self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1)
Tim Peters's avatar
Tim Peters committed
60
        self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'],
61 62
                           thread.get_ident())
        self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped'], 1)
Tim Peters's avatar
Tim Peters committed
63
        self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped_by'],
64
                           thread.get_ident())
65
        signalled_all.release()
66 67 68

    def spawnSignallingThread(self):
        thread.start_new_thread(send_signals, ())
Tim Peters's avatar
Tim Peters committed
69

70 71

def test_main():
72
    global signal_blackboard
Tim Peters's avatar
Tim Peters committed
73

74 75 76 77
    signal_blackboard = { signal.SIGUSR1 : {'tripped': 0, 'tripped_by': 0 },
                          signal.SIGUSR2 : {'tripped': 0, 'tripped_by': 0 },
                          signal.SIGALRM : {'tripped': 0, 'tripped_by': 0 } }

78
    oldsigs = registerSignals(handle_signals, handle_signals, handle_signals)
79
    try:
80
        run_unittest(ThreadSignals)
81
    finally:
82
        registerSignals(*oldsigs)
83 84 85

if __name__ == '__main__':
    test_main()