mirror of
https://github.com/python/cpython
synced 2024-07-20 17:55:24 +00:00
45df820591
svn+ssh://pythondev@svn.python.org/python/trunk ........ r80552 | victor.stinner | 2010-04-27 23:46:03 +0200 (mar., 27 avril 2010) | 3 lines Issue #7449, part 1: fix test_support.py for Python compiled without thread ........ r80553 | victor.stinner | 2010-04-27 23:47:01 +0200 (mar., 27 avril 2010) | 1 line Issue #7449, part 2: regrtest.py -j option requires thread support ........ r80554 | victor.stinner | 2010-04-27 23:51:26 +0200 (mar., 27 avril 2010) | 9 lines Issue #7449 part 3, test_doctest: import trace module in test_coverage() Import trace module fail if the threading module is missing. test_coverage() is only used if test_doctest.py is used with the -c option. This commit allows to execute the test suite without thread support. Move "import trace" in test_coverage() and use test_support.import_module('trace'). ........ r80555 | victor.stinner | 2010-04-27 23:56:26 +0200 (mar., 27 avril 2010) | 6 lines Issue #7449, part 4: skip test_multiprocessing if thread support is disabled import threading after _multiprocessing to raise a more revelant error message: "No module named _multiprocessing". _multiprocessing is not compiled without thread support. ........ r80556 | victor.stinner | 2010-04-28 00:01:24 +0200 (mer., 28 avril 2010) | 8 lines Issue #7449, part 5: split Test.test_open() of ctypes/test/test_errno.py * Split Test.test_open() in 2 functions: test_open() and test_thread_open() * Skip test_open() and test_thread_open() if we are unable to find the C library * Skip test_thread_open() if thread support is disabled * Use unittest.skipUnless(os.name == "nt", ...) on test_GetLastError() ........ r80564 | victor.stinner | 2010-04-28 00:59:35 +0200 (mer., 28 avril 2010) | 4 lines Issue #7449, part 6: fix test_hashlib for missing threading module Move @test_support.reap_thread decorator from test_main() to test_threaded_hashing(). ........ r80565 | victor.stinner | 2010-04-28 01:01:29 +0200 (mer., 28 avril 2010) | 6 lines Issue #7449, part 7: simplify threading detection in test_capi * Skip TestPendingCalls if threading module is missing * Test if threading module is present or not, instead of test the presence of _testcapi._test_thread_state ........ r80566 | victor.stinner | 2010-04-28 01:03:16 +0200 (mer., 28 avril 2010) | 4 lines Issue #7449, part 8: don't skip the whole test_asynchat if threading is missing TestFifo can be executed without the threading module ........ r80568 | victor.stinner | 2010-04-28 01:14:58 +0200 (mer., 28 avril 2010) | 6 lines Issue #7449, part 9: fix test_xmlrpclib for missing threading module * Skip testcases using threads if threading module is missing * Use "http://" instead of URL in ServerProxyTestCase if threading is missing because URL is not set in this case ........ r80569 | victor.stinner | 2010-04-28 01:33:58 +0200 (mer., 28 avril 2010) | 6 lines Partial revert of r80556 (Issue #7449, part 5, fix ctypes test) Rewrite r80556: the thread test have to be executed just after the test on libc_open() and so the test cannot be splitted in two functions (without duplicating code, and I don't want to duplicate code). ........ r80570 | victor.stinner | 2010-04-28 01:51:16 +0200 (mer., 28 avril 2010) | 8 lines Issue #7449, part 10: test_cmd imports trace module using test_support.import_module() Use test_support.import_module() instead of import to raise a SkipTest exception if the import fail. Import trace fails if the threading module is missing. See also part 3: test_doctest: import trace module in test_coverage(). ........ r80571 | victor.stinner | 2010-04-28 01:55:59 +0200 (mer., 28 avril 2010) | 6 lines Issue #7449, last part (11): fix many tests if thread support is disabled * Use try/except ImportError or test_support.import_module() to import thread and threading modules * Add @unittest.skipUnless(threading, ...) to testcases using threads ........
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""PyUnit testing that threads honor our signal semantics"""
|
|
|
|
import unittest
|
|
import signal
|
|
import os
|
|
import sys
|
|
from test.support import run_unittest, import_module
|
|
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.
|
|
"""
|
|
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()
|