(self)
| 5326 | |
| 5327 | @support.requires_resource('walltime') |
| 5328 | def test_wait_timeout(self): |
| 5329 | from multiprocessing.connection import wait |
| 5330 | |
| 5331 | timeout = 5.0 # seconds |
| 5332 | a, b = multiprocessing.Pipe() |
| 5333 | |
| 5334 | start = time.monotonic() |
| 5335 | res = wait([a, b], timeout) |
| 5336 | delta = time.monotonic() - start |
| 5337 | |
| 5338 | self.assertEqual(res, []) |
| 5339 | self.assertGreater(delta, timeout - CLOCK_RES) |
| 5340 | |
| 5341 | b.send(None) |
| 5342 | res = wait([a, b], 20) |
| 5343 | self.assertEqual(res, [a]) |
| 5344 | |
| 5345 | @classmethod |
| 5346 | def signal_and_sleep(cls, sem, period): |
nothing calls this directly
no test coverage detected