1 from test import test_support

     2 import unittest

     3 import dummy_threading as _threading

     4 import time

     5 

     6 class DummyThreadingTestCase(unittest.TestCase):

     7 

     8     class TestThread(_threading.Thread):

     9 

    10         def run(self):

    11             global running

    12             global sema

    13             global mutex

    14             # Uncomment if testing another module, such as the real 'threading'

    15             # module.

    16             #delay = random.random() * 2

    17             delay = 0

    18             if test_support.verbose:

    19                 print 'task', self.name, 'will run for', delay, 'sec'

    20             sema.acquire()

    21             mutex.acquire()

    22             running += 1

    23             if test_support.verbose:

    24                 print running, 'tasks are running'

    25             mutex.release()

    26             time.sleep(delay)

    27             if test_support.verbose:

    28                 print 'task', self.name, 'done'

    29             mutex.acquire()

    30             running -= 1

    31             if test_support.verbose:

    32                 print self.name, 'is finished.', running, 'tasks are running'

    33             mutex.release()

    34             sema.release()

    35 

    36     def setUp(self):

    37         self.numtasks = 10

    38         global sema

    39         sema = _threading.BoundedSemaphore(value=3)

    40         global mutex

    41         mutex = _threading.RLock()

    42         global running

    43         running = 0

    44         self.threads = []

    45 

    46     def test_tasks(self):

    47         for i in range(self.numtasks):

    48             t = self.TestThread(name="<thread %d>"%i)

    49             self.threads.append(t)

    50             t.start()

    51 

    52         if test_support.verbose:

    53             print 'waiting for all tasks to complete'

    54         for t in self.threads:

    55             t.join()

    56         if test_support.verbose:

    57             print 'all tasks done'

    58 

    59 def test_main():

    60     test_support.run_unittest(DummyThreadingTestCase)

    61 

    62 if __name__ == '__main__':

    63     test_main()