diff options
-rw-r--r-- | Lib/asyncio/tasks.py | 12 | ||||
-rw-r--r-- | 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 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 |