mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-01 10:45:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 15

asyncio: Add Task.current_task() class method.

Browse source
This commit is contained in:
Guido van Rossum 2013-12-06 12:57:40 -08:00
parent 2f8c83568c
commit 1a605ed5a3
3 changed files with 57 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1113,6 +1113,42 @@ class TaskTests(unittest.TestCase):
self.assertEqual(res, 'test')
self.assertIsNone(t2.result())
def test_current_task(self):
self.assertIsNone(tasks.Task.current_task(loop=self.loop))
@tasks.coroutine
def coro(loop):
self.assertTrue(tasks.Task.current_task(loop=loop) is task)
task = tasks.Task(coro(self.loop), loop=self.loop)
self.loop.run_until_complete(task)
self.assertIsNone(tasks.Task.current_task(loop=self.loop))
def test_current_task_with_interleaving_tasks(self):
self.assertIsNone(tasks.Task.current_task(loop=self.loop))
fut1 = futures.Future(loop=self.loop)
fut2 = futures.Future(loop=self.loop)
@tasks.coroutine
def coro1(loop):
self.assertTrue(tasks.Task.current_task(loop=loop) is task1)
yield from fut1
self.assertTrue(tasks.Task.current_task(loop=loop) is task1)
fut2.set_result(True)
@tasks.coroutine
def coro2(loop):
self.assertTrue(tasks.Task.current_task(loop=loop) is task2)
fut1.set_result(True)
yield from fut2
self.assertTrue(tasks.Task.current_task(loop=loop) is task2)
task1 = tasks.Task(coro1(self.loop), loop=self.loop)
task2 = tasks.Task(coro2(self.loop), loop=self.loop)
self.loop.run_until_complete(tasks.wait((task1, task2), loop=self.loop))
self.assertIsNone(tasks.Task.current_task(loop=self.loop))
# Some thorough tests for cancellation propagation through
# coroutines, tasks and wait().