0001 """Generic thread tests. 0002 0003 Meant to be used by dummy_thread and thread. To allow for different modules 0004 to be used, test_main() can be called with the module to use as the thread 0005 implementation as its sole argument. 0006 0007 """ 0008 import dummy_thread as _thread 0009 import time 0010 import Queue 0011 import random 0012 import unittest 0013 from test import test_support 0014 0015 DELAY = 0 # Set > 0 when testing a module other than dummy_thread, such as 0016 # the 'thread' module. 0017 0018 class LockTests(unittest.TestCase): 0019 """Test lock objects.""" 0020 0021 def setUp(self): 0022 # Create a lock 0023 self.lock = _thread.allocate_lock() 0024 0025 def test_initlock(self): 0026 #Make sure locks start locked 0027 self.failUnless(not self.lock.locked(), 0028 "Lock object is not initialized unlocked.") 0029 0030 def test_release(self): 0031 # Test self.lock.release() 0032 self.lock.acquire() 0033 self.lock.release() 0034 self.failUnless(not self.lock.locked(), 0035 "Lock object did not release properly.") 0036 0037 def test_improper_release(self): 0038 #Make sure release of an unlocked thread raises _thread.error 0039 self.failUnlessRaises(_thread.error, self.lock.release) 0040 0041 def test_cond_acquire_success(self): 0042 #Make sure the conditional acquiring of the lock works. 0043 self.failUnless(self.lock.acquire(0), 0044 "Conditional acquiring of the lock failed.") 0045 0046 def test_cond_acquire_fail(self): 0047 #Test acquiring locked lock returns False 0048 self.lock.acquire(0) 0049 self.failUnless(not self.lock.acquire(0), 0050 "Conditional acquiring of a locked lock incorrectly " 0051 "succeeded.") 0052 0053 def test_uncond_acquire_success(self): 0054 #Make sure unconditional acquiring of a lock works. 0055 self.lock.acquire() 0056 self.failUnless(self.lock.locked(), 0057 "Uncondional locking failed.") 0058 0059 def test_uncond_acquire_return_val(self): 0060 #Make sure that an unconditional locking returns True. 0061 self.failUnless(self.lock.acquire(1) is True, 0062 "Unconditional locking did not return True.") 0063 0064 def test_uncond_acquire_blocking(self): 0065 #Make sure that unconditional acquiring of a locked lock blocks. 0066 def delay_unlock(to_unlock, delay): 0067 """Hold on to lock for a set amount of time before unlocking.""" 0068 time.sleep(delay) 0069 to_unlock.release() 0070 0071 self.lock.acquire() 0072 start_time = int(time.time()) 0073 _thread.start_new_thread(delay_unlock,(self.lock, DELAY)) 0074 if test_support.verbose: 0075 print 0076 print "*** Waiting for thread to release the lock "\ 0077 "(approx. %s sec.) ***" % DELAY 0078 self.lock.acquire() 0079 end_time = int(time.time()) 0080 if test_support.verbose: 0081 print "done" 0082 self.failUnless((end_time - start_time) >= DELAY, 0083 "Blocking by unconditional acquiring failed.") 0084 0085 class MiscTests(unittest.TestCase): 0086 """Miscellaneous tests.""" 0087 0088 def test_exit(self): 0089 #Make sure _thread.exit() raises SystemExit 0090 self.failUnlessRaises(SystemExit, _thread.exit) 0091 0092 def test_ident(self): 0093 #Test sanity of _thread.get_ident() 0094 self.failUnless(isinstance(_thread.get_ident(), int), 0095 "_thread.get_ident() returned a non-integer") 0096 self.failUnless(_thread.get_ident() != 0, 0097 "_thread.get_ident() returned 0") 0098 0099 def test_LockType(self): 0100 #Make sure _thread.LockType is the same type as _thread.allocate_locke() 0101 self.failUnless(isinstance(_thread.allocate_lock(), _thread.LockType), 0102 "_thread.LockType is not an instance of what is " 0103 "returned by _thread.allocate_lock()") 0104 0105 def test_interrupt_main(self): 0106 #Calling start_new_thread with a function that executes interrupt_main 0107 # should raise KeyboardInterrupt upon completion. 0108 def call_interrupt(): 0109 _thread.interrupt_main() 0110 self.failUnlessRaises(KeyboardInterrupt, _thread.start_new_thread, 0111 call_interrupt, tuple()) 0112 0113 def test_interrupt_in_main(self): 0114 # Make sure that if interrupt_main is called in main threat that 0115 # KeyboardInterrupt is raised instantly. 0116 self.failUnlessRaises(KeyboardInterrupt, _thread.interrupt_main) 0117 0118 class ThreadTests(unittest.TestCase): 0119 """Test thread creation.""" 0120 0121 def test_arg_passing(self): 0122 #Make sure that parameter passing works. 0123 def arg_tester(queue, arg1=False, arg2=False): 0124 """Use to test _thread.start_new_thread() passes args properly.""" 0125 queue.put((arg1, arg2)) 0126 0127 testing_queue = Queue.Queue(1) 0128 _thread.start_new_thread(arg_tester, (testing_queue, True, True)) 0129 result = testing_queue.get() 0130 self.failUnless(result[0] and result[1], 0131 "Argument passing for thread creation using tuple failed") 0132 _thread.start_new_thread(arg_tester, tuple(), {'queue':testing_queue, 0133 'arg1':True, 'arg2':True}) 0134 result = testing_queue.get() 0135 self.failUnless(result[0] and result[1], 0136 "Argument passing for thread creation using kwargs failed") 0137 _thread.start_new_thread(arg_tester, (testing_queue, True), {'arg2':True}) 0138 result = testing_queue.get() 0139 self.failUnless(result[0] and result[1], 0140 "Argument passing for thread creation using both tuple" 0141 " and kwargs failed") 0142 0143 def test_multi_creation(self): 0144 #Make sure multiple threads can be created. 0145 def queue_mark(queue, delay): 0146 """Wait for ``delay`` seconds and then put something into ``queue``""" 0147 time.sleep(delay) 0148 queue.put(_thread.get_ident()) 0149 0150 thread_count = 5 0151 testing_queue = Queue.Queue(thread_count) 0152 if test_support.verbose: 0153 print 0154 print "*** Testing multiple thread creation "\ 0155 "(will take approx. %s to %s sec.) ***" % (DELAY, thread_count) 0156 for count in xrange(thread_count): 0157 if DELAY: 0158 local_delay = round(random.random(), 1) 0159 else: 0160 local_delay = 0 0161 _thread.start_new_thread(queue_mark, 0162 (testing_queue, local_delay)) 0163 time.sleep(DELAY) 0164 if test_support.verbose: 0165 print 'done' 0166 self.failUnless(testing_queue.qsize() == thread_count, 0167 "Not all %s threads executed properly after %s sec." % 0168 (thread_count, DELAY)) 0169 0170 def test_main(imported_module=None): 0171 global _thread, DELAY 0172 if imported_module: 0173 _thread = imported_module 0174 DELAY = 2 0175 if test_support.verbose: 0176 print 0177 print "*** Using %s as _thread module ***" % _thread 0178 test_support.run_unittest(LockTests, MiscTests, ThreadTests) 0179 0180 if __name__ == '__main__': 0181 test_main() 0182
Generated by PyXR 0.9.4