(self)
| 30 | self.assertEqual(deadline, cm.when()) |
| 31 | |
| 32 | async def test_nested_timeouts(self): |
| 33 | loop = asyncio.get_running_loop() |
| 34 | cancelled = False |
| 35 | with self.assertRaises(TimeoutError): |
| 36 | deadline = loop.time() + 0.01 |
| 37 | async with asyncio.timeout_at(deadline) as cm1: |
| 38 | # Only the topmost context manager should raise TimeoutError |
| 39 | try: |
| 40 | async with asyncio.timeout_at(deadline) as cm2: |
| 41 | await asyncio.sleep(10) |
| 42 | except asyncio.CancelledError: |
| 43 | cancelled = True |
| 44 | raise |
| 45 | self.assertTrue(cancelled) |
| 46 | self.assertTrue(cm1.expired()) |
| 47 | self.assertTrue(cm2.expired()) |
| 48 | |
| 49 | async def test_waiter_cancelled(self): |
| 50 | cancelled = False |
nothing calls this directly
no test coverage detected