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:

"""

depth = 0

while True:

context = exc.__context__

if context is None:

break

exc = context

depth += 1

def __repr__(self):

return '<Dummy>'

def __call__(self, *args):

def send(self, v):

raise StopIteration(42)

def throw(self, *exc):

pass

def close(self):

pass

def __await__(self):

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