mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-30 21:48:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

bpo-29271: Fix Task.current_task and Task.all_tasks to accept None. (#406)

Browse source
This commit is contained in:
Yury Selivanov 2017-03-02 22:16:33 -05:00 • committed by GitHub
parent ba7e1f9a4e
commit 8d26aa930c
4 changed files with 29 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

17
Lib/test/test_asyncio/test_tasks.py
View file

@ -1461,6 +1461,14 @@ class BaseTaskTests:
def coro(loop):
self.assertTrue(Task.current_task(loop=loop) is task)
# See http://bugs.python.org/issue29271 for details:
asyncio.set_event_loop(loop)
try:
self.assertIs(Task.current_task(None), task)
self.assertIs(Task.current_task(), task)
finally:
asyncio.set_event_loop(None)
task = self.new_task(self.loop, coro(self.loop))
self.loop.run_until_complete(task)
self.assertIsNone(Task.current_task(loop=self.loop))
@ -1805,8 +1813,17 @@ class BaseTaskTests:
# schedule the task
coro = kill_me(self.loop)
task = asyncio.ensure_future(coro, loop=self.loop)
self.assertEqual(Task.all_tasks(loop=self.loop), {task})
# See http://bugs.python.org/issue29271 for details:
asyncio.set_event_loop(self.loop)
try:
self.assertEqual(Task.all_tasks(), {task})
self.assertEqual(Task.all_tasks(None), {task})
finally:
asyncio.set_event_loop(None)
# execute the task so it waits for future
self.loop._run_once()
self.assertEqual(len(self.loop._ready), 0)