summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--Lib/asyncio/tasks.py12
-rw-r--r--Lib/test/test_asyncio/test_tasks.py47
2 files changed, 59 insertions, 0 deletions
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index 0785e10..c6c22dd 100644
--- a/Lib/asyncio/tasks.py
+++ b/Lib/asyncio/tasks.py
@@ -63,6 +63,18 @@ class CoroWrapper:
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
index 2b90a10..80571b4 100644
--- 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 @@ class TaskTests(unittest.TestCase):
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