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

1 files changed, 17 insertions, 0 deletions

diff --git a/Lib/test/test_asyncio/test_tasks.py b/Lib/test/test_asyncio/test_tasks.py
index a18d49a..4f05319 100644
--- a/Lib/test/test_asyncio/test_tasks.py
+++ b/Lib/test/test_asyncio/test_tasks.py
@@ -1462,6 +1462,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))
@@ -1806,8 +1814,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)