(self)
| 5655 | |
| 5656 | @support.requires_resource('walltime') |
| 5657 | def test_wait_timeout(self): |
| 5658 | from multiprocessing.connection import wait |
| 5659 | |
| 5660 | timeout = 5.0 # seconds |
| 5661 | a, b = multiprocessing.Pipe() |
| 5662 | |
| 5663 | start = time.monotonic() |
| 5664 | res = wait([a, b], timeout) |
| 5665 | delta = time.monotonic() - start |
| 5666 | |
| 5667 | self.assertEqual(res, []) |
| 5668 | self.assertGreater(delta, timeout - CLOCK_RES) |
| 5669 | |
| 5670 | b.send(None) |
| 5671 | res = wait([a, b], 20) |
| 5672 | self.assertEqual(res, [a]) |
| 5673 | |
| 5674 | @classmethod |
| 5675 | def signal_and_sleep(cls, sem, period): |
nothing calls this directly
no test coverage detected