(self)
| 3001 | self.pool.map(sqr, SpecialIterable(), 1) |
| 3002 | |
| 3003 | def test_async(self): |
| 3004 | res = self.pool.apply_async(sqr, (7, TIMEOUT1,)) |
| 3005 | get = TimingWrapper(res.get) |
| 3006 | self.assertEqual(get(), 49) |
| 3007 | self.assertTimingAlmostEqual(get.elapsed, TIMEOUT1) |
| 3008 | |
| 3009 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
| 3010 | def test_async_timeout(self): |
nothing calls this directly
no test coverage detected