(self)
| 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): |
nothing calls this directly
no test coverage detected