(self)
| 516 | |
| 517 | |
| 518 | def test_cancelling(self): |
| 519 | loop = asyncio.new_event_loop() |
| 520 | |
| 521 | async def task(): |
| 522 | await asyncio.sleep(10) |
| 523 | |
| 524 | try: |
| 525 | t = self.new_task(loop, task()) |
| 526 | self.assertFalse(t.cancelling()) |
| 527 | self.assertNotIn(" cancelling ", repr(t)) |
| 528 | self.assertTrue(t.cancel()) |
| 529 | self.assertTrue(t.cancelling()) |
| 530 | self.assertIn(" cancelling ", repr(t)) |
| 531 | |
| 532 | # Since we commented out two lines from Task.cancel(), |
| 533 | # this t.cancel() call now returns True. |
| 534 | # self.assertFalse(t.cancel()) |
| 535 | self.assertTrue(t.cancel()) |
| 536 | |
| 537 | with self.assertRaises(asyncio.CancelledError): |
| 538 | loop.run_until_complete(t) |
| 539 | finally: |
| 540 | loop.close() |
| 541 | |
| 542 | def test_uncancel_basic(self): |
| 543 | loop = asyncio.new_event_loop() |
nothing calls this directly
no test coverage detected