Method test_uncancel_basic

Lib/test/test_asyncio/test_tasks.py:542–582

(self)

            loop.close()

    def test_uncancel_basic(self):
        loop = asyncio.new_event_loop()

        async def task():
            try:
                await asyncio.sleep(10)
            except asyncio.CancelledError:
                self.current_task().uncancel()
                await asyncio.sleep(10)

        try:
            t = self.new_task(loop, task())
            loop.run_until_complete(asyncio.sleep(0.01))

            # Cancel first sleep
            self.assertTrue(t.cancel())
            self.assertIn(" cancelling ", repr(t))
            self.assertEqual(t.cancelling(), 1)
            self.assertFalse(t.cancelled())  # Task is still not complete
            loop.run_until_complete(asyncio.sleep(0.01))

            # after .uncancel()
            self.assertNotIn(" cancelling ", repr(t))
            self.assertEqual(t.cancelling(), 0)
            self.assertFalse(t.cancelled())  # Task is still not complete

            # Cancel second sleep
            self.assertTrue(t.cancel())
            self.assertEqual(t.cancelling(), 1)
            self.assertFalse(t.cancelled())  # Task is still not complete
            with self.assertRaises(asyncio.CancelledError):
                loop.run_until_complete(t)
            self.assertTrue(t.cancelled())  # Finally, task complete
            self.assertTrue(t.done())

            # uncancel is no longer effective after the task is complete
            t.uncancel()
            self.assertTrue(t.cancelled())
            self.assertTrue(t.done())
        finally:
            loop.close()

    def test_uncancel_structured_blocks(self):
        # This test recreates the following high-level structure using uncancel()::