w = CoroWrapper(coro(*args, **kwds), func)
if w._source_traceback:
del w._source_traceback[-1]
- w.__name__ = func.__name__
+ if hasattr(func, '__name__'):
+ w.__name__ = func.__name__
if hasattr(func, '__qualname__'):
w.__qualname__ = func.__qualname__
w.__doc__ = func.__doc__
def _format_coroutine(coro):
assert iscoroutine(coro)
- coro_name = getattr(coro, '__qualname__', coro.__name__)
+
+ if isinstance(coro, CoroWrapper):
+ func = coro.func
+ else:
+ func = coro
+ coro_name = events._format_callback(func, ())
filename = coro.gi_code.co_filename
if (isinstance(coro, CoroWrapper)
and not inspect.isgeneratorfunction(coro.func)):
filename, lineno = events._get_function_source(coro.func)
if coro.gi_frame is None:
- coro_repr = ('%s() done, defined at %s:%s'
+ coro_repr = ('%s done, defined at %s:%s'
% (coro_name, filename, lineno))
else:
- coro_repr = ('%s() running, defined at %s:%s'
+ coro_repr = ('%s running, defined at %s:%s'
% (coro_name, filename, lineno))
elif coro.gi_frame is not None:
lineno = coro.gi_frame.f_lineno
- coro_repr = ('%s() running at %s:%s'
+ coro_repr = ('%s running at %s:%s'
% (coro_name, filename, lineno))
else:
lineno = coro.gi_code.co_firstlineno
- coro_repr = ('%s() done, defined at %s:%s'
+ coro_repr = ('%s done, defined at %s:%s'
% (coro_name, filename, lineno))
return coro_repr
suffix = _format_args(args) + suffix
return _format_callback(func.func, func.args, suffix)
- func_repr = getattr(func, '__qualname__', None)
- if not func_repr:
+ if hasattr(func, '__qualname__'):
+ func_repr = getattr(func, '__qualname__')
+ elif hasattr(func, '__name__'):
+ func_repr = getattr(func, '__name__')
+ else:
func_repr = repr(func)
if args is not None:
func_repr += _format_args(args)
if suffix:
func_repr += suffix
+ return func_repr
+def _format_callback_source(func, args):
+ func_repr = _format_callback(func, args)
source = _get_function_source(func)
if source:
func_repr += ' at %s:%s' % source
if self._cancelled:
info.append('cancelled')
if self._callback is not None:
- info.append(_format_callback(self._callback, self._args))
+ info.append(_format_callback_source(self._callback, self._args))
if self._source_traceback:
frame = self._source_traceback[-1]
info.append('created at %s:%s' % (frame[0], frame[1]))
try:
self._callback(*self._args)
except Exception as exc:
- cb = _format_callback(self._callback, self._args)
+ cb = _format_callback_source(self._callback, self._args)
msg = 'Exception in callback {}'.format(cb)
context = {
'message': msg,
cb = ''
def format_cb(callback):
- return events._format_callback(callback, ())
+ return events._format_callback_source(callback, ())
if size == 1:
cb = format_cb(cb[0])
"""Tests for tasks.py."""
+import contextlib
+import functools
import os
import re
import sys
pass
+@contextlib.contextmanager
+def set_coroutine_debug(enabled):
+ coroutines = asyncio.coroutines
+
+ old_debug = coroutines._DEBUG
+ try:
+ coroutines._DEBUG = enabled
+ yield
+ finally:
+ coroutines._DEBUG = old_debug
+
+
+
def format_coroutine(qualname, state, src, source_traceback, generator=False):
if generator:
state = '%s' % state
fut.set_result(None)
self.loop.run_until_complete(task)
+ def test_task_repr_partial_corowrapper(self):
+ # Issue #222: repr(CoroWrapper) must not fail in debug mode if the
+ # coroutine is a partial function
+ with set_coroutine_debug(True):
+ self.loop.set_debug(True)
+
+ @asyncio.coroutine
+ def func(x, y):
+ yield from asyncio.sleep(0)
+
+ partial_func = asyncio.coroutine(functools.partial(func, 1))
+ task = self.loop.create_task(partial_func(2))
+
+ # make warnings quiet
+ task._log_destroy_pending = False
+ self.addCleanup(task._coro.close)
+
+ coro_repr = repr(task._coro)
+ expected = ('<CoroWrapper TaskTests.test_task_repr_partial_corowrapper'
+ '.<locals>.func(1)() running, ')
+ self.assertTrue(coro_repr.startswith(expected),
+ coro_repr)
+
def test_task_basics(self):
@asyncio.coroutine
def outer():
# The frame should have changed.
self.assertIsNone(gen.gi_frame)
- # Save debug flag.
- old_debug = asyncio.coroutines._DEBUG
- try:
- # Test with debug flag cleared.
- asyncio.coroutines._DEBUG = False
+ # Test with debug flag cleared.
+ with set_coroutine_debug(False):
check()
- # Test with debug flag set.
- asyncio.coroutines._DEBUG = True
+ # Test with debug flag set.
+ with set_coroutine_debug(True):
check()
- finally:
- # Restore original debug flag.
- asyncio.coroutines._DEBUG = old_debug
-
def test_yield_from_corowrapper(self):
- old_debug = asyncio.coroutines._DEBUG
- asyncio.coroutines._DEBUG = True
- try:
+ with set_coroutine_debug(True):
@asyncio.coroutine
def t1():
return (yield from t2())
task = asyncio.Task(t1(), loop=self.loop)
val = self.loop.run_until_complete(task)
self.assertEqual(val, (1, 2, 3))
- finally:
- asyncio.coroutines._DEBUG = old_debug
def test_yield_from_corowrapper_send(self):
def foo():
@mock.patch('asyncio.coroutines.logger')
def test_coroutine_never_yielded(self, m_log):
- debug = asyncio.coroutines._DEBUG
- try:
- asyncio.coroutines._DEBUG = True
+ with set_coroutine_debug(True):
@asyncio.coroutine
def coro_noop():
pass
- finally:
- asyncio.coroutines._DEBUG = debug
tb_filename = __file__
tb_lineno = sys._getframe().f_lineno + 2
Library
-------
+- Asyncio issue 222 / PR 231 (Victor Stinner) -- fix @coroutine
+ functions without __name__.
+
- Issue #9246: On POSIX, os.getcwd() now supports paths longer than 1025 bytes.
Patch written by William Orr.
projects / python / commitdiff