"""
# On Python 3.5, set the coroutine flag on our generator, to allow it
# to be used with 'await'.
+ wrapped = func
if hasattr(types, 'coroutine'):
func = types.coroutine(func)
- @functools.wraps(func)
+ @functools.wraps(wrapped)
def wrapper(*args, **kwargs):
future = TracebackFuture()
future = None
future.set_result(result)
return future
+
+ wrapper.__wrapped__ = wrapped
+ wrapper.__tornado_coroutine__ = True
return wrapper
+def is_coroutine_function(func):
+ """Return whether *func* is a coroutine function, i.e. a function
+ wrapped with `~.gen.coroutine`.
+ """
+ return getattr(func, '__tornado_coroutine__', False)
+
+
class Return(Exception):
"""Special exception to return a value from a `coroutine`.
super(GenCoroutineTest, self).tearDown()
assert self.finished
+ def test_attributes(self):
+ self.finished = True
+
+ def f():
+ yield gen.moment
+
+ coro = gen.coroutine(f)
+ self.assertEqual(coro.__name__, f.__name__)
+ self.assertEqual(coro.__module__, f.__module__)
+ self.assertIs(coro.__wrapped__, f)
+
+ def test_is_coroutine_function(self):
+ self.finished = True
+
+ def f():
+ yield gen.moment
+
+ coro = gen.coroutine(f)
+ self.assertFalse(gen.is_coroutine_function(f))
+ self.assertTrue(gen.is_coroutine_function(coro))
+ self.assertFalse(gen.is_coroutine_function(coro()))
+
@gen_test
def test_sync_gen_return(self):
@gen.coroutine