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.failUnless(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.failUnless(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.failUnlessRaises(_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.failUnless(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.failUnless(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.failUnless(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.failUnless(self.lock.acquire(1) is True,
  62                         "Unconditional locking did not return True.")
  63
  64     def test_uncond_acquire_blocking(self):
  65         #Make sure that unconditional acquiring of a locked lock blocks.
  66         def delay_unlock(to_unlock, delay):
  67             """Hold on to lock for a set amount of time before unlocking."""
  68             time.sleep(delay)
  69             to_unlock.release()
  70
  71         self.lock.acquire()
  72         start_time = int(time.time())
  73         _thread.start_new_thread(delay_unlock,(self.lock, DELAY))
  74         if test_support.verbose:
  75             print
  76             print "*** Waiting for thread to release the lock "\
  77             "(approx. %s sec.) ***" % DELAY
  78         self.lock.acquire()
  79         end_time = int(time.time())
  80         if test_support.verbose:
  81             print "done"
  82         self.failUnless((end_time - start_time) >= DELAY,
  83                         "Blocking by unconditional acquiring failed.")
  84
  85 class MiscTests(unittest.TestCase):
  86     """Miscellaneous tests."""
  87
  88     def test_exit(self):
  89         #Make sure _thread.exit() raises SystemExit
  90         self.failUnlessRaises(SystemExit, _thread.exit)
  91
  92     def test_ident(self):
  93         #Test sanity of _thread.get_ident()
  94         self.failUnless(isinstance(_thread.get_ident(), int),
  95                         "_thread.get_ident() returned a non-integer")
  96         self.failUnless(_thread.get_ident() != 0,
  97                         "_thread.get_ident() returned 0")
  98
  99     def test_LockType(self):
 100         #Make sure _thread.LockType is the same type as _thread.allocate_locke()
 101         self.failUnless(isinstance(_thread.allocate_lock(), _thread.LockType),
 102                         "_thread.LockType is not an instance of what is "
 103                          "returned by _thread.allocate_lock()")
 104
 105 class ThreadTests(unittest.TestCase):
 106     """Test thread creation."""
 107
 108     def test_arg_passing(self):
 109         #Make sure that parameter passing works.
 110         def arg_tester(queue, arg1=False, arg2=False):
 111             """Use to test _thread.start_new_thread() passes args properly."""
 112             queue.put((arg1, arg2))
 113
 114         testing_queue = Queue.Queue(1)
 115         _thread.start_new_thread(arg_tester, (testing_queue, True, True))
 116         result = testing_queue.get()
 117         self.failUnless(result[0] and result[1],
 118                         "Argument passing for thread creation using tuple failed")
 119         _thread.start_new_thread(arg_tester, tuple(), {'queue':testing_queue,
 120                                                        'arg1':True, 'arg2':True})
 121         result = testing_queue.get()
 122         self.failUnless(result[0] and result[1],
 123                         "Argument passing for thread creation using kwargs failed")
 124         _thread.start_new_thread(arg_tester, (testing_queue, True), {'arg2':True})
 125         result = testing_queue.get()
 126         self.failUnless(result[0] and result[1],
 127                         "Argument passing for thread creation using both tuple"
 128                         " and kwargs failed")
 129
 130     def test_multi_creation(self):
 131         #Make sure multiple threads can be created.
 132         def queue_mark(queue, delay):
 133             """Wait for ``delay`` seconds and then put something into ``queue``"""
 134             time.sleep(delay)
 135             queue.put(_thread.get_ident())
 136
 137         thread_count = 5
 138         testing_queue = Queue.Queue(thread_count)
 139         if test_support.verbose:
 140             print
 141             print "*** Testing multiple thread creation "\
 142             "(will take approx. %s to %s sec.) ***" % (DELAY, thread_count)
 143         for count in xrange(thread_count):
 144             if DELAY:
 145                 local_delay = round(random.random(), 1)
 146             else:
 147                 local_delay = 0
 148             _thread.start_new_thread(queue_mark,
 149                                      (testing_queue, local_delay))
 150         time.sleep(DELAY)
 151         if test_support.verbose:
 152             print 'done'
 153         self.failUnless(testing_queue.qsize() == thread_count,
 154                         "Not all %s threads executed properly after %s sec." %
 155                         (thread_count, DELAY))
 156
 157 def test_main(imported_module=None):
 158     global _thread, DELAY
 159     if imported_module:
 160         _thread = imported_module
 161         DELAY = 2
 162     if test_support.verbose:
 163         print
 164         print "*** Using %s as _thread module ***" % _thread
 165     test_support.run_unittest(LockTests, MiscTests, ThreadTests)
 166
 167 if __name__ == '__main__':
 168     test_main()