Issue #24867: Fix Task.get_stack() for 'async def' coroutines

This commit is contained in:
Yury Selivanov 2015-08-14 15:30:59 -04:00
parent ac37ba0742
commit 233983380d
2 changed files with 37 additions and 1 deletions

View File

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

View File

@ -2,6 +2,7 @@
import contextlib
import functools
import io
import os
import re
import sys
@ -162,6 +163,37 @@ class TaskTests(test_utils.TestCase):
'function is deprecated, use ensure_'):
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):
self.loop.set_debug(False)