(self)
| 142 | |
| 143 | @incr |
| 144 | def test_counter(self): |
| 145 | counter = multiprocessing.Value("i", 0) |
| 146 | process_manager = salt.utils.process.ProcessManager() |
| 147 | self.addCleanup(process_manager.terminate) |
| 148 | process_manager.add_process(self.incr_counter, args=(counter, 2)) |
| 149 | time.sleep(1) |
| 150 | process_manager.check_children() |
| 151 | time.sleep(1) |
| 152 | # we should have had 2 processes go at it |
| 153 | assert counter.value == 4 |
| 154 | |
| 155 | |
| 156 | class TestThreadPool(TestCase): |
nothing calls this directly
no test coverage detected