AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_dummy_thread.py

   1 """Generic thread tests.
   2
   3 Meant to be used by dummy_thread and thread.  To allow for different modules
   4 to be used, test_main() can be called with the module to use as the thread
   5 implementation as its sole argument.
   6
   7 """
   8 import dummy_thread as _thread
   9 import time
  10 import Queue
  11 import random
  12 import unittest
  13 from test import test_support
  14
  15 DELAY = 0 # Set > 0 when testing a module other than dummy_thread, such as
  16           # the 'thread' module.
  17
  18 class LockTests(unittest.TestCase):
  19     """Test lock objects."""
  20
  21     def setUp(self):
  22         # Create a lock
  23         self.lock = _thread.allocate_lock()
  24
  25     def test_initlock(self):
  26         #Make sure locks start locked
  27         self.assertTrue(not self.lock.locked(),
  28                         "Lock object is not initialized unlocked.")
  29
  30     def test_release(self):
  31         # Test self.lock.release()
  32         self.lock.acquire()
  33         self.lock.release()
  34         self.assertTrue(not self.lock.locked(),
  35                         "Lock object did not release properly.")
  36
  37     def test_improper_release(self):
  38         #Make sure release of an unlocked thread raises _thread.error
  39         self.assertRaises(_thread.error, self.lock.release)
  40
  41     def test_cond_acquire_success(self):
  42         #Make sure the conditional acquiring of the lock works.
  43         self.assertTrue(self.lock.acquire(0),
  44                         "Conditional acquiring of the lock failed.")
  45
  46     def test_cond_acquire_fail(self):
  47         #Test acquiring locked lock returns False
  48         self.lock.acquire(0)
  49         self.assertTrue(not self.lock.acquire(0),
  50                         "Conditional acquiring of a locked lock incorrectly "
  51                          "succeeded.")
  52
  53     def test_uncond_acquire_success(self):
  54         #Make sure unconditional acquiring of a lock works.
  55         self.lock.acquire()
  56         self.assertTrue(self.lock.locked(),
  57                         "Uncondional locking failed.")
  58
  59     def test_uncond_acquire_return_val(self):
  60         #Make sure that an unconditional locking returns True.
  61         self.assertTrue(self.lock.acquire(1) is True,
  62                         "Unconditional locking did not return True.")
  63         self.assertTrue(self.lock.acquire() is True)
  64
  65     def test_uncond_acquire_blocking(self):
  66         #Make sure that unconditional acquiring of a locked lock blocks.
  67         def delay_unlock(to_unlock, delay):
  68             """Hold on to lock for a set amount of time before unlocking."""
  69             time.sleep(delay)
  70             to_unlock.release()
  71
  72         self.lock.acquire()
  73         start_time = int(time.time())
  74         _thread.start_new_thread(delay_unlock,(self.lock, DELAY))
  75         if test_support.verbose:
  76             print
  77             print "*** Waiting for thread to release the lock "\
  78             "(approx. %s sec.) ***" % DELAY
  79         self.lock.acquire()
  80         end_time = int(time.time())
  81         if test_support.verbose:
  82             print "done"
  83         self.assertTrue((end_time - start_time) >= DELAY,
  84                         "Blocking by unconditional acquiring failed.")
  85
  86 class MiscTests(unittest.TestCase):
  87     """Miscellaneous tests."""
  88
  89     def test_exit(self):
  90         #Make sure _thread.exit() raises SystemExit
  91         self.assertRaises(SystemExit, _thread.exit)
  92
  93     def test_ident(self):
  94         #Test sanity of _thread.get_ident()
  95         self.assertIsInstance(_thread.get_ident(), int,
  96                               "_thread.get_ident() returned a non-integer")
  97         self.assertTrue(_thread.get_ident() != 0,
  98                         "_thread.get_ident() returned 0")
  99
 100     def test_LockType(self):
 101         #Make sure _thread.LockType is the same type as _thread.allocate_locke()
 102         self.assertIsInstance(_thread.allocate_lock(), _thread.LockType,
 103                               "_thread.LockType is not an instance of what "
 104                               "is returned by _thread.allocate_lock()")
 105
 106     def test_interrupt_main(self):
 107         #Calling start_new_thread with a function that executes interrupt_main
 108         # should raise KeyboardInterrupt upon completion.
 109         def call_interrupt():
 110             _thread.interrupt_main()
 111         self.assertRaises(KeyboardInterrupt, _thread.start_new_thread,
 112                               call_interrupt, tuple())
 113
 114     def test_interrupt_in_main(self):
 115         # Make sure that if interrupt_main is called in main threat that
 116         # KeyboardInterrupt is raised instantly.
 117         self.assertRaises(KeyboardInterrupt, _thread.interrupt_main)
 118
 119 class ThreadTests(unittest.TestCase):
 120     """Test thread creation."""
 121
 122     def test_arg_passing(self):
 123         #Make sure that parameter passing works.
 124         def arg_tester(queue, arg1=False, arg2=False):
 125             """Use to test _thread.start_new_thread() passes args properly."""
 126             queue.put((arg1, arg2))
 127
 128         testing_queue = Queue.Queue(1)
 129         _thread.start_new_thread(arg_tester, (testing_queue, True, True))
 130         result = testing_queue.get()
 131         self.assertTrue(result[0] and result[1],
 132                         "Argument passing for thread creation using tuple failed")
 133         _thread.start_new_thread(arg_tester, tuple(), {'queue':testing_queue,
 134                                                        'arg1':True, 'arg2':True})
 135         result = testing_queue.get()
 136         self.assertTrue(result[0] and result[1],
 137                         "Argument passing for thread creation using kwargs failed")
 138         _thread.start_new_thread(arg_tester, (testing_queue, True), {'arg2':True})
 139         result = testing_queue.get()
 140         self.assertTrue(result[0] and result[1],
 141                         "Argument passing for thread creation using both tuple"
 142                         " and kwargs failed")
 143
 144     def test_multi_creation(self):
 145         #Make sure multiple threads can be created.
 146         def queue_mark(queue, delay):
 147             """Wait for ``delay`` seconds and then put something into ``queue``"""
 148             time.sleep(delay)
 149             queue.put(_thread.get_ident())
 150
 151         thread_count = 5
 152         testing_queue = Queue.Queue(thread_count)
 153         if test_support.verbose:
 154             print
 155             print "*** Testing multiple thread creation "\
 156             "(will take approx. %s to %s sec.) ***" % (DELAY, thread_count)
 157         for count in xrange(thread_count):
 158             if DELAY:
 159                 local_delay = round(random.random(), 1)
 160             else:
 161                 local_delay = 0
 162             _thread.start_new_thread(queue_mark,
 163                                      (testing_queue, local_delay))
 164         time.sleep(DELAY)
 165         if test_support.verbose:
 166             print 'done'
 167         self.assertTrue(testing_queue.qsize() == thread_count,
 168                         "Not all %s threads executed properly after %s sec." %
 169                         (thread_count, DELAY))
 170
 171 def test_main(imported_module=None):
 172     global _thread, DELAY
 173     if imported_module:
 174         _thread = imported_module
 175         DELAY = 2
 176     if test_support.verbose:
 177         print
 178         print "*** Using %s as _thread module ***" % _thread
 179     test_support.run_unittest(LockTests, MiscTests, ThreadTests)
 180
 181 if __name__ == '__main__':
 182     test_main()