Method test_async_timeout

Lib/test/_test_multiprocessing.py:3010–3022

(self)

    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
    def test_async_timeout(self):
        p = self.Pool(3)
        try:
            event = threading.Event() if self.TYPE == 'threads' else None
            res = p.apply_async(sqr, (6, TIMEOUT2 + support.SHORT_TIMEOUT, event))
            get = TimingWrapper(res.get)
            self.assertRaises(multiprocessing.TimeoutError, get, timeout=TIMEOUT2)
            self.assertTimingAlmostEqual(get.elapsed, TIMEOUT2)
        finally:
            if event is not None:
                event.set()
            p.terminate()
            p.join()

    def test_imap(self):
        it = self.pool.imap(sqr, list(range(10)))