Home | History | Annotate | Download | only in test 
      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()
    183