add some tests

Author: graingert

Lib/test/test_asyncio/test_eager_task_factory.py

@@ -263,6 +263,39 @@ async def run():
 
         self.run_coro(run())
 
+    def test_eager_start_false(self):
+        name = None
+
+        async def asyncfn():
+            nonlocal name
+            name = asyncio.current_task().get_name()
+
+        async def main():
+            t = asyncio.create_task(asyncfn(), eager_start=False, name="example")
+            self.assertFalse(t.done())
+            self.assertIsNone(name)
+            await t
+            self.assertEqual(name, "example")
+
+        self.run_coro(main())
+
+
+    def test_eager_start_false_tg(self):
+        name = None
+
+        async def asyncfn():
+            nonlocal name
+            name = asyncio.current_task().get_name()
+
+        async def main():
+            async with asyncio.TaskGroup() as tg:
+                t = tg.create_task(asyncfn(), eager_start=False, name="example")
+                self.assertFalse(t.done())
+                self.assertIsNone(name)
+            self.assertEqual(name, "example")
+
+        self.run_coro(main())
+
 
 class PyEagerTaskFactoryLoopTests(EagerTaskFactoryLoopTests, test_utils.TestCase):
     Task = tasks._PyTask

Lib/test/test_asyncio/test_tasks.py

@@ -89,8 +89,8 @@ class BaseTaskTests:
     Future = None
     all_tasks = None
 
-    def new_task(self, loop, coro, name='TestTask', context=None):
-        return self.__class__.Task(coro, loop=loop, name=name, context=context)
+    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)
@@ -2667,6 +2667,35 @@ async def main():
 
         self.assertEqual([None, 1, 2], ret)
 
+    def test_eager_start_true(self):
+        name = None
+
+        async def asyncfn():
+            nonlocal name
+            name = asyncio.current_task().get_name()
+
+        async def main():
+            t = self.new_task(coro=asyncfn(), loop=asyncio.get_running_loop(), eager_start=True, name="example")
+            self.assertTrue(t.done())
+            self.assertEqual(name, "example")
+            await t
+
+    def test_eager_start_false(self):
+        name = None
+
+        async def asyncfn():
+            nonlocal name
+            name = asyncio.current_task().get_name()
+
+        async def main():
+            t = self.new_task(coro=asyncfn(), loop=asyncio.get_running_loop(), eager_start=False, name="example")
+            self.assertFalse(t.done())
+            self.assertIsNone(name)
+            await t
+            self.assertEqual(name, "example")
+
+        asyncio.run(main(), loop_factory=asyncio.EventLoop)
+
     def test_get_coro(self):
         loop = asyncio.new_event_loop()
         coro = coroutine_function()