"""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()