diff -r ffa851df0825 -r 2fb8b9db1c86 symbian-qemu-0.9.1-12/python-2.6.1/Lib/test/test_dummy_thread.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/symbian-qemu-0.9.1-12/python-2.6.1/Lib/test/test_dummy_thread.py Fri Jul 31 15:01:17 2009 +0100 @@ -0,0 +1,182 @@ +"""Generic thread tests. + +Meant to be used by dummy_thread and thread. To allow for different modules +to be used, test_main() can be called with the module to use as the thread +implementation as its sole argument. + +""" +import dummy_thread as _thread +import time +import Queue +import random +import unittest +from test import test_support + +DELAY = 0 # Set > 0 when testing a module other than dummy_thread, such as + # the 'thread' module. + +class LockTests(unittest.TestCase): + """Test lock objects.""" + + def setUp(self): + # Create a lock + self.lock = _thread.allocate_lock() + + def test_initlock(self): + #Make sure locks start locked + self.failUnless(not self.lock.locked(), + "Lock object is not initialized unlocked.") + + def test_release(self): + # Test self.lock.release() + self.lock.acquire() + self.lock.release() + self.failUnless(not self.lock.locked(), + "Lock object did not release properly.") + + def test_improper_release(self): + #Make sure release of an unlocked thread raises _thread.error + self.failUnlessRaises(_thread.error, self.lock.release) + + def test_cond_acquire_success(self): + #Make sure the conditional acquiring of the lock works. + self.failUnless(self.lock.acquire(0), + "Conditional acquiring of the lock failed.") + + def test_cond_acquire_fail(self): + #Test acquiring locked lock returns False + self.lock.acquire(0) + self.failUnless(not self.lock.acquire(0), + "Conditional acquiring of a locked lock incorrectly " + "succeeded.") + + def test_uncond_acquire_success(self): + #Make sure unconditional acquiring of a lock works. + self.lock.acquire() + self.failUnless(self.lock.locked(), + "Uncondional locking failed.") + + def test_uncond_acquire_return_val(self): + #Make sure that an unconditional locking returns True. + self.failUnless(self.lock.acquire(1) is True, + "Unconditional locking did not return True.") + self.failUnless(self.lock.acquire() is True) + + def test_uncond_acquire_blocking(self): + #Make sure that unconditional acquiring of a locked lock blocks. + def delay_unlock(to_unlock, delay): + """Hold on to lock for a set amount of time before unlocking.""" + time.sleep(delay) + to_unlock.release() + + self.lock.acquire() + start_time = int(time.time()) + _thread.start_new_thread(delay_unlock,(self.lock, DELAY)) + if test_support.verbose: + print + print "*** Waiting for thread to release the lock "\ + "(approx. %s sec.) ***" % DELAY + self.lock.acquire() + end_time = int(time.time()) + if test_support.verbose: + print "done" + self.failUnless((end_time - start_time) >= DELAY, + "Blocking by unconditional acquiring failed.") + +class MiscTests(unittest.TestCase): + """Miscellaneous tests.""" + + def test_exit(self): + #Make sure _thread.exit() raises SystemExit + self.failUnlessRaises(SystemExit, _thread.exit) + + def test_ident(self): + #Test sanity of _thread.get_ident() + self.failUnless(isinstance(_thread.get_ident(), int), + "_thread.get_ident() returned a non-integer") + self.failUnless(_thread.get_ident() != 0, + "_thread.get_ident() returned 0") + + def test_LockType(self): + #Make sure _thread.LockType is the same type as _thread.allocate_locke() + self.failUnless(isinstance(_thread.allocate_lock(), _thread.LockType), + "_thread.LockType is not an instance of what is " + "returned by _thread.allocate_lock()") + + def test_interrupt_main(self): + #Calling start_new_thread with a function that executes interrupt_main + # should raise KeyboardInterrupt upon completion. + def call_interrupt(): + _thread.interrupt_main() + self.failUnlessRaises(KeyboardInterrupt, _thread.start_new_thread, + call_interrupt, tuple()) + + def test_interrupt_in_main(self): + # Make sure that if interrupt_main is called in main threat that + # KeyboardInterrupt is raised instantly. + self.failUnlessRaises(KeyboardInterrupt, _thread.interrupt_main) + +class ThreadTests(unittest.TestCase): + """Test thread creation.""" + + def test_arg_passing(self): + #Make sure that parameter passing works. + def arg_tester(queue, arg1=False, arg2=False): + """Use to test _thread.start_new_thread() passes args properly.""" + queue.put((arg1, arg2)) + + testing_queue = Queue.Queue(1) + _thread.start_new_thread(arg_tester, (testing_queue, True, True)) + result = testing_queue.get() + self.failUnless(result[0] and result[1], + "Argument passing for thread creation using tuple failed") + _thread.start_new_thread(arg_tester, tuple(), {'queue':testing_queue, + 'arg1':True, 'arg2':True}) + result = testing_queue.get() + self.failUnless(result[0] and result[1], + "Argument passing for thread creation using kwargs failed") + _thread.start_new_thread(arg_tester, (testing_queue, True), {'arg2':True}) + result = testing_queue.get() + self.failUnless(result[0] and result[1], + "Argument passing for thread creation using both tuple" + " and kwargs failed") + + def test_multi_creation(self): + #Make sure multiple threads can be created. + def queue_mark(queue, delay): + """Wait for ``delay`` seconds and then put something into ``queue``""" + time.sleep(delay) + queue.put(_thread.get_ident()) + + thread_count = 5 + testing_queue = Queue.Queue(thread_count) + if test_support.verbose: + print + print "*** Testing multiple thread creation "\ + "(will take approx. %s to %s sec.) ***" % (DELAY, thread_count) + for count in xrange(thread_count): + if DELAY: + local_delay = round(random.random(), 1) + else: + local_delay = 0 + _thread.start_new_thread(queue_mark, + (testing_queue, local_delay)) + time.sleep(DELAY) + if test_support.verbose: + print 'done' + self.failUnless(testing_queue.qsize() == thread_count, + "Not all %s threads executed properly after %s sec." % + (thread_count, DELAY)) + +def test_main(imported_module=None): + global _thread, DELAY + if imported_module: + _thread = imported_module + DELAY = 2 + if test_support.verbose: + print + print "*** Using %s as _thread module ***" % _thread + test_support.run_unittest(LockTests, MiscTests, ThreadTests) + +if __name__ == '__main__': + test_main()