[Python-checkins] cpython: asyncio: Add gi_{frame,running,code} properties to CoroWrapper (upstream #163).

[guido.van.rossum]

http://hg.python.org/cpython/rev/c776ed8a8e75
changeset:   90398:c776ed8a8e75
user:        Guido van Rossum <guido at python.org>
date:        Tue Apr 15 12:06:34 2014 -0700
summary:
  asyncio: Add gi_{frame,running,code} properties to CoroWrapper (upstream #163).

files:
  Lib/asyncio/tasks.py                |  12 ++++
  Lib/test/test_asyncio/test_tasks.py |  47 +++++++++++++++++
  2 files changed, 59 insertions(+), 0 deletions(-)


diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -63,6 +63,18 @@
     def close(self):
         return self.gen.close()
 
+    @property
+    def gi_frame(self):
+        return self.gen.gi_frame
+
+    @property
+    def gi_running(self):
+        return self.gen.gi_running
+
+    @property
+    def gi_code(self):
+        return self.gen.gi_code
+
     def __del__(self):
         frame = self.gen.gi_frame
         if frame is not None and frame.f_lasti == -1:
diff --git a/Lib/test/test_asyncio/test_tasks.py b/Lib/test/test_asyncio/test_tasks.py
--- a/Lib/test/test_asyncio/test_tasks.py
+++ b/Lib/test/test_asyncio/test_tasks.py
@@ -2,6 +2,7 @@
 
 import gc
 import os.path
+import types
 import unittest
 from test.script_helper import assert_python_ok
 
@@ -1386,6 +1387,52 @@
         self.assertRaises(ValueError, self.loop.run_until_complete,
             asyncio.wait([], loop=self.loop))
 
+    def test_corowrapper_mocks_generator(self):
+
+        def check():
+            # A function that asserts various things.
+            # Called twice, with different debug flag values.
+
+            @asyncio.coroutine
+            def coro():
+                # The actual coroutine.
+                self.assertTrue(gen.gi_running)
+                yield from fut
+
+            # A completed Future used to run the coroutine.
+            fut = asyncio.Future(loop=self.loop)
+            fut.set_result(None)
+
+            # Call the coroutine.
+            gen = coro()
+
+            # Check some properties.
+            self.assertTrue(asyncio.iscoroutine(gen))
+            self.assertIsInstance(gen.gi_frame, types.FrameType)
+            self.assertFalse(gen.gi_running)
+            self.assertIsInstance(gen.gi_code, types.CodeType)
+
+            # Run it.
+            self.loop.run_until_complete(gen)
+
+            # The frame should have changed.
+            self.assertIsNone(gen.gi_frame)
+
+        # Save debug flag.
+        old_debug = asyncio.tasks._DEBUG
+        try:
+            # Test with debug flag cleared.
+            asyncio.tasks._DEBUG = False
+            check()
+
+            # Test with debug flag set.
+            asyncio.tasks._DEBUG = True
+            check()
+
+        finally:
+            # Restore original debug flag.
+            asyncio.tasks._DEBUG = old_debug
+
     def test_yield_from_corowrapper(self):
         old_debug = asyncio.tasks._DEBUG
         asyncio.tasks._DEBUG = True

-- 
Repository URL: http://hg.python.org/cpython