(self)
| 113 | self.assertEqual(foo_running, False) |
| 114 | |
| 115 | async def test_wait_for(self): |
| 116 | foo_running = None |
| 117 | |
| 118 | async def foo(): |
| 119 | nonlocal foo_running |
| 120 | foo_running = True |
| 121 | try: |
| 122 | await asyncio.sleep(support.LONG_TIMEOUT) |
| 123 | finally: |
| 124 | foo_running = False |
| 125 | return 'done' |
| 126 | |
| 127 | fut = asyncio.create_task(foo()) |
| 128 | |
| 129 | with self.assertRaises(asyncio.TimeoutError): |
| 130 | await asyncio.wait_for(fut, 0.1) |
| 131 | self.assertTrue(fut.done()) |
| 132 | # it should have been cancelled due to the timeout |
| 133 | self.assertTrue(fut.cancelled()) |
| 134 | self.assertEqual(foo_running, False) |
| 135 | |
| 136 | async def test_wait_for_blocking(self): |
| 137 | async def coro(): |
nothing calls this directly
no test coverage detected