(self)
| 226 | self.assertTrue(t0 <= t1 <= t0 + 1) |
| 227 | |
| 228 | async def test_reschedule(self): |
| 229 | loop = asyncio.get_running_loop() |
| 230 | fut = loop.create_future() |
| 231 | deadline1 = loop.time() + 10 |
| 232 | deadline2 = deadline1 + 20 |
| 233 | |
| 234 | async def f(): |
| 235 | async with asyncio.timeout_at(deadline1) as cm: |
| 236 | fut.set_result(cm) |
| 237 | await asyncio.sleep(50) |
| 238 | |
| 239 | task = asyncio.create_task(f()) |
| 240 | cm = await fut |
| 241 | |
| 242 | self.assertEqual(cm.when(), deadline1) |
| 243 | cm.reschedule(deadline2) |
| 244 | self.assertEqual(cm.when(), deadline2) |
| 245 | cm.reschedule(None) |
| 246 | self.assertIsNone(cm.when()) |
| 247 | |
| 248 | task.cancel() |
| 249 | |
| 250 | with self.assertRaises(asyncio.CancelledError): |
| 251 | await task |
| 252 | self.assertFalse(cm.expired()) |
| 253 | |
| 254 | async def test_repr_active(self): |
| 255 | async with asyncio.timeout(10) as cm: |
nothing calls this directly
no test coverage detected