"""Tests for tasks.py."""

import collections

import contextlib

import contextvars

import gc

import io

import random

import re

import sys

import traceback

import types

import unittest

from unittest import mock

from types import GenericAlias

import asyncio

from asyncio import futures

from asyncio import tasks

from test.test_asyncio import utils as test_utils

from test import support

from test.support.script_helper import assert_python_ok

from test.support.warnings_helper import ignore_warnings

def tearDownModule():

asyncio.events._set_event_loop_policy(None)

async def coroutine_function():

pass

def format_coroutine(qualname, state, src, source_traceback, generator=False):

if generator:

state = '%s' % state

else:

state = '%s, defined' % state

if source_traceback is not None:

frame = source_traceback[-1]

return ('coro=<%s() %s at %s> created at %s:%s'

% (qualname, state, src, frame[0], frame[1]))

else:

return 'coro=<%s() %s at %s>' % (qualname, state, src)

def get_innermost_context(exc):

"""

Return information about the innermost exception context in the chain.

"""

depth = 0

while True:

context = exc.__context__

if context is None:

break

exc = context

depth += 1

return (type(exc), exc.args, depth)

class Dummy:

def __repr__(self):

return '<Dummy>'

def __call__(self, *args):

pass

class CoroLikeObject:

def send(self, v):

raise StopIteration(42)

def throw(self, *exc):

pass

def close(self):

pass

def __await__(self):

return self

class BaseTaskTests:

Task = None

Future = None

all_tasks = None

def new_task(self, loop, coro, name='TestTask', context=None, eager_start=None):

return self.__class__.Task(coro, loop=loop, name=name, context=context, eager_start=eager_start)

def new_future(self, loop):

return self.__class__.Future(loop=loop)

def setUp(self):

super().setUp()

self.loop = self.new_test_loop()

self.loop.set_task_factory(self.new_task)

self.loop.create_future = lambda: self.new_future(self.loop)

def test_generic_alias(self):

task = self.__class__.Task[str]

self.assertEqual(task.__args__, (str,))

self.assertIsInstance(task, GenericAlias)

def test_task_cancel_message_getter(self):

async def coro():

pass

t = self.new_task(self.loop, coro())

self.assertHasAttr(t, '_cancel_message')

self.assertEqual(t._cancel_message, None)

t.cancel('my message')

self.assertEqual(t._cancel_message, 'my message')

with self.assertRaises(asyncio.CancelledError) as cm:

self.loop.run_until_complete(t)

self.assertEqual('my message', cm.exception.args[0])

def test_task_cancel_message_setter(self):

async def coro():

pass

t = self.new_task(self.loop, coro())

t.cancel('my message')

t._cancel_message = 'my new message'

self.assertEqual(t._cancel_message, 'my new message')

with self.assertRaises(asyncio.CancelledError) as cm:

self.loop.run_until_complete(t)

self.assertEqual('my new message', cm.exception.args[0])

def test_task_del_collect(self):

class Evil:

def __del__(self):

gc.collect()

async def run():

return Evil()

self.loop.run_until_complete(

asyncio.gather(*[

self.new_task(self.loop, run()) for _ in range(100)

]))

def test_other_loop_future(self):

other_loop = asyncio.new_event_loop()

fut = self.new_future(other_loop)

async def run(fut):

await fut

try:

with self.assertRaisesRegex(RuntimeError,

r'Task .* got Future .* attached'):

self.loop.run_until_complete(run(fut))

finally:

other_loop.close()

def test_task_awaits_on_itself(self):

async def test():

await task

task = asyncio.ensure_future(test(), loop=self.loop)

with self.assertRaisesRegex(RuntimeError,

'Task cannot await on itself'):

self.loop.run_until_complete(task)

def test_task_class(self):

async def notmuch():

return 'ok'

t = self.new_task(self.loop, notmuch())

self.loop.run_until_complete(t)

self.assertTrue(t.done())

self.assertEqual(t.result(), 'ok')

self.assertIs(t._loop, self.loop)

self.assertIs(t.get_loop(), self.loop)

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

t = self.new_task(loop, notmuch())

self.assertIs(t._loop, loop)

loop.run_until_complete(t)

loop.close()

def test_ensure_future_coroutine(self):

async def notmuch():

return 'ok'

t = asyncio.ensure_future(notmuch(), loop=self.loop)

self.assertIs(t._loop, self.loop)

self.loop.run_until_complete(t)

self.assertTrue(t.done())

self.assertEqual(t.result(), 'ok')

a = notmuch()

self.addCleanup(a.close)

with self.assertRaisesRegex(RuntimeError, 'no current event loop'):

asyncio.ensure_future(a)

async def test():

return asyncio.ensure_future(notmuch())

t = self.loop.run_until_complete(test())

self.assertIs(t._loop, self.loop)

self.loop.run_until_complete(t)

self.assertTrue(t.done())

self.assertEqual(t.result(), 'ok')

# Deprecated in 3.10, undeprecated in 3.12

asyncio.set_event_loop(self.loop)

self.addCleanup(asyncio.set_event_loop, None)

t = asyncio.ensure_future(notmuch())

self.assertIs(t._loop, self.loop)

self.loop.run_until_complete(t)

self.assertTrue(t.done())

self.assertEqual(t.result(), 'ok')

def test_ensure_future_future(self):

f_orig = self.new_future(self.loop)

f_orig.set_result('ko')

f = asyncio.ensure_future(f_orig)

self.loop.run_until_complete(f)

self.assertTrue(f.done())

self.assertEqual(f.result(), 'ko')

self.assertIs(f, f_orig)

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

with self.assertRaises(ValueError):

f = asyncio.ensure_future(f_orig, loop=loop)

loop.close()

f = asyncio.ensure_future(f_orig, loop=self.loop)

self.assertIs(f, f_orig)

def test_ensure_future_task(self):

async def notmuch():

return 'ok'

t_orig = self.new_task(self.loop, notmuch())

t = asyncio.ensure_future(t_orig)

self.loop.run_until_complete(t)

self.assertTrue(t.done())

self.assertEqual(t.result(), 'ok')

self.assertIs(t, t_orig)

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

with self.assertRaises(ValueError):

t = asyncio.ensure_future(t_orig, loop=loop)

loop.close()

t = asyncio.ensure_future(t_orig, loop=self.loop)

self.assertIs(t, t_orig)

def test_ensure_future_awaitable(self):

class Aw:

def __init__(self, coro):

self.coro = coro

def __await__(self):

return self.coro.__await__()

async def coro():

return 'ok'

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

fut = asyncio.ensure_future(Aw(coro()), loop=loop)

loop.run_until_complete(fut)

self.assertEqual(fut.result(), 'ok')

def test_ensure_future_task_awaitable(self):

class Aw:

def __await__(self):

return asyncio.sleep(0, result='ok').__await__()

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

task = asyncio.ensure_future(Aw(), loop=loop)

loop.run_until_complete(task)

self.assertTrue(task.done())

self.assertEqual(task.result(), 'ok')

self.assertIsInstance(task.get_coro(), types.CoroutineType)

loop.close()

def test_ensure_future_neither(self):

with self.assertRaises(TypeError):

asyncio.ensure_future('ok')

def test_ensure_future_error_msg(self):

loop = asyncio.new_event_loop()

f = self.new_future(self.loop)

with self.assertRaisesRegex(ValueError, 'The future belongs to a '

'different loop than the one specified as '

'the loop argument'):

asyncio.ensure_future(f, loop=loop)

loop.close()

def test_get_stack(self):

T = None

async def foo():

await bar()

async def bar():

# test get_stack()

f = T.get_stack(limit=1)

try:

self.assertEqual(f[0].f_code.co_name, 'foo')

finally:

f = None

# test print_stack()

file = io.StringIO()

T.print_stack(limit=1, file=file)

file.seek(0)

tb = file.read()

self.assertRegex(tb, r'foo\(\) running')

async def runner():

nonlocal T

T = asyncio.ensure_future(foo(), loop=self.loop)

await T

self.loop.run_until_complete(runner())

def test_task_repr(self):

self.loop.set_debug(False)

async def notmuch():

return 'abc'

# test coroutine function

self.assertEqual(notmuch.__name__, 'notmuch')

self.assertRegex(notmuch.__qualname__,

r'\w+.test_task_repr.<locals>.notmuch')

self.assertEqual(notmuch.__module__, __name__)

filename, lineno = test_utils.get_function_source(notmuch)

src = "%s:%s" % (filename, lineno)

# test coroutine object

gen = notmuch()

coro_qualname = 'BaseTaskTests.test_task_repr.<locals>.notmuch'

self.assertEqual(gen.__name__, 'notmuch')

self.assertEqual(gen.__qualname__, coro_qualname)

# test pending Task

t = self.new_task(self.loop, gen)

t.add_done_callback(Dummy())

coro = format_coroutine(coro_qualname, 'running', src,

t._source_traceback, generator=True)

self.assertEqual(repr(t),

"<Task pending name='TestTask' %s cb=[<Dummy>()]>" % coro)

# test cancelling Task

t.cancel() # Does not take immediate effect!

self.assertEqual(repr(t),

"<Task cancelling name='TestTask' %s cb=[<Dummy>()]>" % coro)

# test cancelled Task

self.assertRaises(asyncio.CancelledError,

self.loop.run_until_complete, t)

coro = format_coroutine(coro_qualname, 'done', src,

t._source_traceback)

self.assertEqual(repr(t),

"<Task cancelled name='TestTask' %s>" % coro)

# test finished Task

t = self.new_task(self.loop, notmuch())

self.loop.run_until_complete(t)

coro = format_coroutine(coro_qualname, 'done', src,

t._source_traceback)

self.assertEqual(repr(t),

"<Task finished name='TestTask' %s result='abc'>" % coro)

def test_task_repr_autogenerated(self):

async def notmuch():

return 123

t1 = self.new_task(self.loop, notmuch(), None)

t2 = self.new_task(self.loop, notmuch(), None)

self.assertNotEqual(repr(t1), repr(t2))

match1 = re.match(r"^<Task pending name='Task-(\d+)'", repr(t1))

self.assertIsNotNone(match1)

match2 = re.match(r"^<Task pending name='Task-(\d+)'", repr(t2))

self.assertIsNotNone(match2)

# Autogenerated task names should have monotonically increasing numbers

self.assertLess(int(match1.group(1)), int(match2.group(1)))

self.loop.run_until_complete(t1)

self.loop.run_until_complete(t2)

def test_task_set_name_pylong(self):

# test that setting the task name to a PyLong explicitly doesn't

# incorrectly trigger the deferred name formatting logic

async def notmuch():

return 123

t = self.new_task(self.loop, notmuch(), name=987654321)

self.assertEqual(t.get_name(), '987654321')

t.set_name(123456789)

self.assertEqual(t.get_name(), '123456789')

self.loop.run_until_complete(t)

def test_task_repr_name_not_str(self):

async def notmuch():

return 123

t = self.new_task(self.loop, notmuch())

t.set_name({6})

self.assertEqual(t.get_name(), '{6}')

self.loop.run_until_complete(t)

def test_task_repr_wait_for(self):

self.loop.set_debug(False)

async def wait_for(fut):

return await fut

fut = self.new_future(self.loop)

task = self.new_task(self.loop, wait_for(fut))

test_utils.run_briefly(self.loop)

self.assertRegex(repr(task),

'<Task .* wait_for=%s>' % re.escape(repr(fut)))

fut.set_result(None)

self.loop.run_until_complete(task)

def test_task_basics(self):

async def outer():

a = await inner1()

b = await inner2()

return a+b

async def inner1():

return 42

async def inner2():

return 1000

t = outer()

self.assertEqual(self.loop.run_until_complete(t), 1042)

def test_exception_chaining_after_await(self):

# Test that when awaiting on a task when an exception is already

# active, if the task raises an exception it will be chained

# with the original.

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

async def raise_error():

raise ValueError

async def run():

try:

raise KeyError(3)

except Exception as exc:

task = self.new_task(loop, raise_error())

try:

await task

except Exception as exc:

self.assertEqual(type(exc), ValueError)

chained = exc.__context__

self.assertEqual((type(chained), chained.args),

(KeyError, (3,)))

try:

task = self.new_task(loop, run())

loop.run_until_complete(task)

finally:

loop.close()

def test_exception_chaining_after_await_with_context_cycle(self):

# Check trying to create an exception context cycle:

# https://bugs.python.org/issue40696

has_cycle = None

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

async def process_exc(exc):

raise exc

async def run():

nonlocal has_cycle

try:

raise KeyError('a')

except Exception as exc:

task = self.new_task(loop, process_exc(exc))

try:

await task

except BaseException as exc:

has_cycle = (exc is exc.__context__)

# Prevent a hang if has_cycle is True.

exc.__context__ = None

try:

task = self.new_task(loop, run())

loop.run_until_complete(task)

finally:

loop.close()

# This also distinguishes from the initial has_cycle=None.

self.assertEqual(has_cycle, False)

def test_cancelling(self):

loop = asyncio.new_event_loop()

async def task():

await asyncio.sleep(10)

try:

t = self.new_task(loop, task())

self.assertFalse(t.cancelling())

self.assertNotIn(" cancelling ", repr(t))

self.assertTrue(t.cancel())

self.assertTrue(t.cancelling())

self.assertIn(" cancelling ", repr(t))

# Since we commented out two lines from Task.cancel(),

# this t.cancel() call now returns True.

# self.assertFalse(t.cancel())

self.assertTrue(t.cancel())

with self.assertRaises(asyncio.CancelledError):

loop.run_until_complete(t)

finally:

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()::

#

# async def make_request_with_timeout():

# try:

# async with asyncio.timeout(1):

# # Structured block affected by the timeout:

# await make_request()

# await make_another_request()

# except TimeoutError:

# pass # There was a timeout

# # Outer code not affected by the timeout:

# await unrelated_code()

loop = asyncio.new_event_loop()

async def make_request_with_timeout(*, sleep: float, timeout: float):

task = self.current_task()

loop = task.get_loop()

timed_out = False

structured_block_finished = False

outer_code_reached = False

def on_timeout():

nonlocal timed_out

timed_out = True

task.cancel()

timeout_handle = loop.call_later(timeout, on_timeout)

try:

try:

# Structured block affected by the timeout

await asyncio.sleep(sleep)

structured_block_finished = True

finally:

timeout_handle.cancel()

if (

timed_out

and task.uncancel() == 0

and type(sys.exception()) is asyncio.CancelledError

):

# Note the five rules that are needed here to satisfy proper

# uncancellation:

#

# 1. handle uncancellation in a `finally:` block to allow for

# plain returns;

# 2. our `timed_out` flag is set, meaning that it was our event

# that triggered the need to uncancel the task, regardless of

# what exception is raised;

# 3. we can call `uncancel()` because *we* called `cancel()`

# before;

# 4. we call `uncancel()` but we only continue converting the

# CancelledError to TimeoutError if `uncancel()` caused the

# cancellation request count go down to 0. We need to look

# at the counter vs having a simple boolean flag because our

# code might have been nested (think multiple timeouts). See

# commit 7fce1063b6e5a366f8504e039a8ccdd6944625cd for

# details.

# 5. we only convert CancelledError to TimeoutError; for other

# exceptions raised due to the cancellation (like

# a ConnectionLostError from a database client), simply

# propagate them.

#

# Those checks need to take place in this exact order to make

# sure the `cancelling()` counter always stays in sync.

#

# Additionally, the original stimulus to `cancel()` the task

# needs to be unscheduled to avoid re-cancelling the task later.

# Here we do it by cancelling `timeout_handle` in the `finally:`

# block.

raise TimeoutError

except TimeoutError:

self.assertTrue(timed_out)

# Outer code not affected by the timeout:

outer_code_reached = True

await asyncio.sleep(0)

return timed_out, structured_block_finished, outer_code_reached

try:

# Test which timed out.

t1 = self.new_task(loop, make_request_with_timeout(sleep=10.0, timeout=0.1))

timed_out, structured_block_finished, outer_code_reached = (

loop.run_until_complete(t1)

)

self.assertTrue(timed_out)

self.assertFalse(structured_block_finished) # it was cancelled

self.assertTrue(outer_code_reached) # task got uncancelled after leaving

# the structured block and continued until

# completion

self.assertEqual(t1.cancelling(), 0) # no pending cancellation of the outer task

# Test which did not time out.

t2 = self.new_task(loop, make_request_with_timeout(sleep=0, timeout=10.0))

timed_out, structured_block_finished, outer_code_reached = (

loop.run_until_complete(t2)

)

self.assertFalse(timed_out)

self.assertTrue(structured_block_finished)

self.assertTrue(outer_code_reached)

self.assertEqual(t2.cancelling(), 0)

finally:

loop.close()

def test_uncancel_resets_must_cancel(self):

async def coro():

await fut

return 42

loop = asyncio.new_event_loop()

fut = asyncio.Future(loop=loop)

task = self.new_task(loop, coro())

loop.run_until_complete(asyncio.sleep(0)) # Get task waiting for fut

fut.set_result(None) # Make task runnable

try:

task.cancel() # Enter cancelled state

self.assertEqual(task.cancelling(), 1)

self.assertTrue(task._must_cancel)

task.uncancel() # Undo cancellation

self.assertEqual(task.cancelling(), 0)

self.assertFalse(task._must_cancel)

finally:

res = loop.run_until_complete(task)

self.assertEqual(res, 42)

loop.close()

def test_cancel(self):

def gen():

when = yield

self.assertAlmostEqual(10.0, when)

yield 0

loop = self.new_test_loop(gen)

async def task():

await asyncio.sleep(10.0)

return 12

t = self.new_task(loop, task())

loop.call_soon(t.cancel)

with self.assertRaises(asyncio.CancelledError):

loop.run_until_complete(t)

self.assertTrue(t.done())

self.assertTrue(t.cancelled())

self.assertFalse(t.cancel())

def test_cancel_with_message_then_future_result(self):

# Test Future.result() after calling cancel() with a message.

cases = [

((), ()),

((None,), ()),

(('my message',), ('my message',)),

# Non-string values should roundtrip.

((5,), (5,)),

]

for cancel_args, expected_args in cases:

with self.subTest(cancel_args=cancel_args):

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

async def sleep():

await asyncio.sleep(10)

async def coro():

task = self.new_task(loop, sleep())

await asyncio.sleep(0)

task.cancel(*cancel_args)

done, pending = await asyncio.wait([task])

task.result()

task = self.new_task(loop, coro())

with self.assertRaises(asyncio.CancelledError) as cm:

loop.run_until_complete(task)

exc = cm.exception

self.assertEqual(exc.args, expected_args)

actual = get_innermost_context(exc)

self.assertEqual(actual,

(asyncio.CancelledError, expected_args, 0))

def test_cancel_with_message_then_future_exception(self):

# Test Future.exception() after calling cancel() with a message.

cases = [

((), ()),

((None,), ()),

(('my message',), ('my message',)),

# Non-string values should roundtrip.

((5,), (5,)),

]

for cancel_args, expected_args in cases:

with self.subTest(cancel_args=cancel_args):

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

async def sleep():

await asyncio.sleep(10)

async def coro():

task = self.new_task(loop, sleep())

await asyncio.sleep(0)

task.cancel(*cancel_args)

done, pending = await asyncio.wait([task])

task.exception()

task = self.new_task(loop, coro())

with self.assertRaises(asyncio.CancelledError) as cm:

loop.run_until_complete(task)

exc = cm.exception

self.assertEqual(exc.args, expected_args)

actual = get_innermost_context(exc)

self.assertEqual(actual,

(asyncio.CancelledError, expected_args, 0))

def test_cancellation_exception_context(self):

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

fut = loop.create_future()

async def sleep():

fut.set_result(None)

await asyncio.sleep(10)

async def coro():

inner_task = self.new_task(loop, sleep())

await fut

loop.call_soon(inner_task.cancel, 'msg')

try:

await inner_task

except asyncio.CancelledError as ex:

raise ValueError("cancelled") from ex

task = self.new_task(loop, coro())

with self.assertRaises(ValueError) as cm:

loop.run_until_complete(task)

exc = cm.exception

self.assertEqual(exc.args, ('cancelled',))

actual = get_innermost_context(exc)

self.assertEqual(actual,

(asyncio.CancelledError, ('msg',), 1))

def test_cancel_with_message_before_starting_task(self):

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

async def sleep():

await asyncio.sleep(10)

async def coro():

task = self.new_task(loop, sleep())

# We deliberately leave out the sleep here.

task.cancel('my message')

done, pending = await asyncio.wait([task])

task.exception()

task = self.new_task(loop, coro())

with self.assertRaises(asyncio.CancelledError) as cm:

loop.run_until_complete(task)

exc = cm.exception

self.assertEqual(exc.args, ('my message',))

actual = get_innermost_context(exc)

self.assertEqual(actual,

(asyncio.CancelledError, ('my message',), 0))

def test_cancel_yield(self):

async def task():

await asyncio.sleep(0)

await asyncio.sleep(0)

return 12

t = self.new_task(self.loop, task())

test_utils.run_briefly(self.loop) # start coro

t.cancel()

self.assertRaises(

asyncio.CancelledError, self.loop.run_until_complete, t)

self.assertTrue(t.done())

self.assertTrue(t.cancelled())

self.assertFalse(t.cancel())

def test_cancel_inner_future(self):

f = self.new_future(self.loop)

async def task():

await f

return 12

t = self.new_task(self.loop, task())

test_utils.run_briefly(self.loop) # start task

f.cancel()

with self.assertRaises(asyncio.CancelledError):

self.loop.run_until_complete(t)

self.assertTrue(f.cancelled())

self.assertTrue(t.cancelled())

def test_cancel_both_task_and_inner_future(self):

f = self.new_future(self.loop)

async def task():

await f

return 12

t = self.new_task(self.loop, task())

test_utils.run_briefly(self.loop)

f.cancel()

t.cancel()

with self.assertRaises(asyncio.CancelledError):

self.loop.run_until_complete(t)

self.assertTrue(t.done())

self.assertTrue(f.cancelled())

self.assertTrue(t.cancelled())

def test_cancel_task_catching(self):

fut1 = self.new_future(self.loop)

fut2 = self.new_future(self.loop)

async def task():

await fut1

try:

await fut2

except asyncio.CancelledError:

return 42

t = self.new_task(self.loop, task())

test_utils.run_briefly(self.loop)

self.assertIs(t._fut_waiter, fut1) # White-box test.

fut1.set_result(None)

test_utils.run_briefly(self.loop)

self.assertIs(t._fut_waiter, fut2) # White-box test.

t.cancel()

self.assertTrue(fut2.cancelled())

res = self.loop.run_until_complete(t)

self.assertEqual(res, 42)

self.assertFalse(t.cancelled())

def test_cancel_task_ignoring(self):

fut1 = self.new_future(self.loop)

fut2 = self.new_future(self.loop)

fut3 = self.new_future(self.loop)

async def task():

await fut1

try:

await fut2

except asyncio.CancelledError:

pass

res = await fut3

return res

t = self.new_task(self.loop, task())

test_utils.run_briefly(self.loop)

self.assertIs(t._fut_waiter, fut1) # White-box test.

fut1.set_result(None)

test_utils.run_briefly(self.loop)

self.assertIs(t._fut_waiter, fut2) # White-box test.

t.cancel()

self.assertTrue(fut2.cancelled())

test_utils.run_briefly(self.loop)

self.assertIs(t._fut_waiter, fut3) # White-box test.

fut3.set_result(42)

res = self.loop.run_until_complete(t)

self.assertEqual(res, 42)

self.assertFalse(fut3.cancelled())

self.assertFalse(t.cancelled())

def test_cancel_current_task(self):

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

async def task():

t.cancel()

self.assertTrue(t._must_cancel) # White-box test.

# The sleep should be cancelled immediately.

await asyncio.sleep(100)

return 12

t = self.new_task(loop, task())

self.assertFalse(t.cancelled())

self.assertRaises(

asyncio.CancelledError, loop.run_until_complete, t)

self.assertTrue(t.done())

self.assertTrue(t.cancelled())

self.assertFalse(t._must_cancel) # White-box test.

self.assertFalse(t.cancel())

def test_cancel_at_end(self):

"""coroutine end right after task is cancelled"""

loop = asyncio.new_event_loop()

self.set_event_loop(loop)

async def task():

t.cancel()

self.assertTrue(t._must_cancel) # White-box test.

return 12

t = self.new_task(loop, task())

self.assertFalse(t.cancelled())

self.assertRaises(

asyncio.CancelledError, loop.run_until_complete, t)

self.assertTrue(t.done())

self.assertTrue(t.cancelled())

self.assertFalse(t._must_cancel) # White-box test.

self.assertFalse(t.cancel())

def test_cancel_awaited_task(self):

# This tests for a relatively rare condition when

# a task cancellation is requested for a task which is not

# currently blocked, such as a task cancelling itself.

# In this situation we must ensure that whatever next future