| 76 | return self.value |
| 77 | |
| 78 | class TestThread(threading.Thread): |
| 79 | def __init__(self, name, testcase, sema, mutex, nrunning): |
| 80 | threading.Thread.__init__(self, name=name) |
| 81 | self.testcase = testcase |
| 82 | self.sema = sema |
| 83 | self.mutex = mutex |
| 84 | self.nrunning = nrunning |
| 85 | |
| 86 | def run(self): |
| 87 | delay = random.random() / 10000.0 |
| 88 | if verbose: |
| 89 | print('task %s will run for %.1f usec' % |
| 90 | (self.name, delay * 1e6)) |
| 91 | |
| 92 | with self.sema: |
| 93 | with self.mutex: |
| 94 | self.nrunning.inc() |
| 95 | if verbose: |
| 96 | print(self.nrunning.get(), 'tasks are running') |
| 97 | self.testcase.assertLessEqual(self.nrunning.get(), 3) |
| 98 | |
| 99 | time.sleep(delay) |
| 100 | if verbose: |
| 101 | print('task', self.name, 'done') |
| 102 | |
| 103 | with self.mutex: |
| 104 | self.nrunning.dec() |
| 105 | self.testcase.assertGreaterEqual(self.nrunning.get(), 0) |
| 106 | if verbose: |
| 107 | print('%s is finished. %d tasks are running' % |
| 108 | (self.name, self.nrunning.get())) |
| 109 | |
| 110 | |
| 111 | class BaseTestCase(unittest.TestCase): |
no outgoing calls
searching dependent graphs…