EmbeddedPkg: Extend NvVarStoreFormattedLib LIBRARY_CLASS

[mirror_edk2.git] / AppPkg / Applications / Python / Python-2.7.2 / Lib / test / test_threadsignals.py

"""PyUnit testing that threads honor our signal semantics"""\r
\r
import unittest\r
import signal\r
import os\r
import sys\r
from test.test_support import run_unittest, import_module, reap_threads\r
thread = import_module('thread')\r
\r
if sys.platform[:3] in ('win', 'os2') or sys.platform=='riscos':\r
    raise unittest.SkipTest, "Can't test signal on %s" % sys.platform\r
\r
process_pid = os.getpid()\r
signalled_all=thread.allocate_lock()\r
\r
\r
def registerSignals(for_usr1, for_usr2, for_alrm):\r
    usr1 = signal.signal(signal.SIGUSR1, for_usr1)\r
    usr2 = signal.signal(signal.SIGUSR2, for_usr2)\r
    alrm = signal.signal(signal.SIGALRM, for_alrm)\r
    return usr1, usr2, alrm\r
\r
\r
# The signal handler. Just note that the signal occurred and\r
# from who.\r
def handle_signals(sig,frame):\r
    signal_blackboard[sig]['tripped'] += 1\r
    signal_blackboard[sig]['tripped_by'] = thread.get_ident()\r
\r
# a function that will be spawned as a separate thread.\r
def send_signals():\r
    os.kill(process_pid, signal.SIGUSR1)\r
    os.kill(process_pid, signal.SIGUSR2)\r
    signalled_all.release()\r
\r
class ThreadSignals(unittest.TestCase):\r
    """Test signal handling semantics of threads.\r
       We spawn a thread, have the thread send two signals, and\r
       wait for it to finish. Check that we got both signals\r
       and that they were run by the main thread.\r
    """\r
    @reap_threads\r
    def test_signals(self):\r
        signalled_all.acquire()\r
        self.spawnSignallingThread()\r
        signalled_all.acquire()\r
        # the signals that we asked the kernel to send\r
        # will come back, but we don't know when.\r
        # (it might even be after the thread exits\r
        # and might be out of order.)  If we haven't seen\r
        # the signals yet, send yet another signal and\r
        # wait for it return.\r
        if signal_blackboard[signal.SIGUSR1]['tripped'] == 0 \\r
           or signal_blackboard[signal.SIGUSR2]['tripped'] == 0:\r
            signal.alarm(1)\r
            signal.pause()\r
            signal.alarm(0)\r
\r
        self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1)\r
        self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'],\r
                           thread.get_ident())\r
        self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped'], 1)\r
        self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped_by'],\r
                           thread.get_ident())\r
        signalled_all.release()\r
\r
    def spawnSignallingThread(self):\r
        thread.start_new_thread(send_signals, ())\r
\r
\r
def test_main():\r
    global signal_blackboard\r
\r
    signal_blackboard = { signal.SIGUSR1 : {'tripped': 0, 'tripped_by': 0 },\r
                          signal.SIGUSR2 : {'tripped': 0, 'tripped_by': 0 },\r
                          signal.SIGALRM : {'tripped': 0, 'tripped_by': 0 } }\r
\r
    oldsigs = registerSignals(handle_signals, handle_signals, handle_signals)\r
    try:\r
        run_unittest(ThreadSignals)\r
    finally:\r
        registerSignals(*oldsigs)\r
\r
if __name__ == '__main__':\r
    test_main()\r