Merged in 1st1/cpython350 (pull request #5)

Issue #24867: Fix asyncio.Task.get_stack() for 'async def' coroutines
This commit is contained in:
Larry Hastings 2015-08-24 16:53:45 -07:00
commit 7250d02b73
4 changed files with 56 additions and 1 deletions

View File

@ -128,6 +128,10 @@ class Task(futures.Future):
returned for a suspended coroutine. returned for a suspended coroutine.
""" """
frames = [] frames = []
try:
# 'async def' coroutines
f = self._coro.cr_frame
except AttributeError:
f = self._coro.gi_frame f = self._coro.gi_frame
if f is not None: if f is not None:
while f is not None: while f is not None:

View File

@ -186,6 +186,23 @@ class CoroutineTests(BaseTest):
data = self.loop.run_until_complete(coro()) data = self.loop.run_until_complete(coro())
self.assertEqual(data, 'spam') self.assertEqual(data, 'spam')
def test_task_print_stack(self):
T = None
async def foo():
f = T.get_stack(limit=1)
try:
self.assertEqual(f[0].f_code.co_name, 'foo')
finally:
f = None
async def runner():
nonlocal T
T = asyncio.ensure_future(foo(), loop=self.loop)
await T
self.loop.run_until_complete(runner())
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()

View File

@ -2,6 +2,7 @@
import contextlib import contextlib
import functools import functools
import io
import os import os
import re import re
import sys import sys
@ -162,6 +163,37 @@ class TaskTests(test_utils.TestCase):
'function is deprecated, use ensure_'): 'function is deprecated, use ensure_'):
self.assertIs(f, asyncio.async(f)) self.assertIs(f, asyncio.async(f))
def test_get_stack(self):
T = None
@asyncio.coroutine
def foo():
yield from bar()
@asyncio.coroutine
def bar():
# test get_stack()
f = T.get_stack(limit=1)
try:
self.assertEqual(f[0].f_code.co_name, 'foo')
finally:
f = None
# test print_stack()
file = io.StringIO()
T.print_stack(limit=1, file=file)
file.seek(0)
tb = file.read()
self.assertRegex(tb, r'foo\(\) running')
@asyncio.coroutine
def runner():
nonlocal T
T = asyncio.ensure_future(foo(), loop=self.loop)
yield from T
self.loop.run_until_complete(runner())
def test_task_repr(self): def test_task_repr(self):
self.loop.set_debug(False) self.loop.set_debug(False)

View File

@ -79,6 +79,8 @@ Library
- Issue #24791: Fix grammar regression for call syntax: 'g(*a or b)'. - Issue #24791: Fix grammar regression for call syntax: 'g(*a or b)'.
- Issue #24867: Fix Task.get_stack() for 'async def' coroutines
Documentation Documentation
------------- -------------