[Python-checkins] python/dist/src/Lib/test test_dummy_thread.py,1.2,1.3 test_dummy_threading.py,1.2,1.3

[bcannon@users.sourceforge.net]

Update of /cvsroot/python/python/dist/src/Lib/test
In directory sc8-pr-cvs1:/tmp/cvs-serv27564/Lib/test

Modified Files:
	test_dummy_thread.py test_dummy_threading.py 
Log Message:
Make time.sleep calls go to 0 for common testing.

Index: test_dummy_thread.py
===================================================================
RCS file: /cvsroot/python/python/dist/src/Lib/test/test_dummy_thread.py,v
retrieving revision 1.2
retrieving revision 1.3
diff -C2 -d -r1.2 -r1.3
*** test_dummy_thread.py	19 Feb 2003 02:35:05 -0000	1.2
--- test_dummy_thread.py	30 Apr 2003 03:03:37 -0000	1.3
***************
*** 13,16 ****
--- 13,18 ----
  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):
***************
*** 68,83 ****
  
          self.lock.acquire()
-         delay = 1  #In seconds
          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.")
  
--- 70,84 ----
  
          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.")
  
***************
*** 135,158 ****
  
          thread_count = 5
-         delay = 1.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):
              _thread.start_new_thread(queue_mark,
!                                      (testing_queue, round(random.random(), 1)))
!         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
      if imported_module:
          _thread = imported_module
      if test_support.verbose:
          print
--- 136,163 ----
  
          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

Index: test_dummy_threading.py
===================================================================
RCS file: /cvsroot/python/python/dist/src/Lib/test/test_dummy_threading.py,v
retrieving revision 1.2
retrieving revision 1.3
diff -C2 -d -r1.2 -r1.3
*** test_dummy_threading.py	19 Feb 2003 02:35:05 -0000	1.2
--- test_dummy_threading.py	30 Apr 2003 03:03:37 -0000	1.3
***************
*** 13,17 ****
      def run(self):
          global running
!         delay = random.random() * 2
          if verbose:
              print 'task', self.getName(), 'will run for', delay, 'sec'
--- 13,20 ----
      def run(self):
          global running
!         # Uncomment if testing another module, such as the real 'threading'
!         # module.
!         #delay = random.random() * 2
!         delay = 0
          if verbose:
              print 'task', self.getName(), 'will run for', delay, 'sec'