test_dummy_thread.py 文件源码

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.assertEqual(testing_queue.qsize(), thread_count,
                        "Not all %s threads executed properly after %s sec." %
                        (thread_count, DELAY))