def test_default_exc_handler_coro(self):
self.loop._process_events = mock.Mock()
- @asyncio.coroutine
- def zero_error_coro():
- yield from asyncio.sleep(0.01)
+ async def zero_error_coro():
+ await asyncio.sleep(0.01)
1/0
# Test Future.__del__
class MyTask(asyncio.Task):
pass
- @asyncio.coroutine
- def coro():
+ async def coro():
pass
factory = lambda loop, coro: MyTask(coro, loop=loop)
class MyTask(asyncio.Task):
pass
- @asyncio.coroutine
- def test():
+ async def test():
pass
class EventLoop(base_events.BaseEventLoop):
# Python issue #22601: ensure that the temporary task created by
# run_forever() consumes the KeyboardInterrupt and so don't log
# a warning
- @asyncio.coroutine
- def raise_keyboard_interrupt():
+ async def raise_keyboard_interrupt():
raise KeyboardInterrupt
self.loop._process_events = mock.Mock()
def test_run_until_complete_baseexception(self):
# Python issue #22429: run_until_complete() must not schedule a pending
# call to stop() if the future raised a BaseException
- @asyncio.coroutine
- def raise_keyboard_interrupt():
+ async def raise_keyboard_interrupt():
raise KeyboardInterrupt
self.loop._process_events = mock.Mock()
class MyProto(asyncio.Protocol):
pass
- @asyncio.coroutine
- def getaddrinfo(*args, **kw):
- yield from []
+ async def getaddrinfo(*args, **kw):
return [(2, 1, 6, '', ('107.6.106.82', 80)),
(2, 1, 6, '', ('107.6.106.82', 80))]
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
def test_create_connection_no_getaddrinfo(self):
- @asyncio.coroutine
- def getaddrinfo(*args, **kw):
- yield from []
+ async def getaddrinfo(*args, **kw):
+ return []
def getaddrinfo_task(*args, **kwds):
return asyncio.Task(getaddrinfo(*args, **kwds), loop=self.loop)
OSError, self.loop.run_until_complete, coro)
def test_create_connection_multiple(self):
- @asyncio.coroutine
- def getaddrinfo(*args, **kw):
+ async def getaddrinfo(*args, **kw):
return [(2, 1, 6, '', ('0.0.0.1', 80)),
(2, 1, 6, '', ('0.0.0.2', 80))]
m_socket.socket.return_value.bind = bind
- @asyncio.coroutine
- def getaddrinfo(*args, **kw):
+ async def getaddrinfo(*args, **kw):
return [(2, 1, 6, '', ('0.0.0.1', 80)),
(2, 1, 6, '', ('0.0.0.2', 80))]
self.loop.run_until_complete(coro)
def test_create_connection_no_local_addr(self):
- @asyncio.coroutine
- def getaddrinfo(host, *args, **kw):
+ async def getaddrinfo(host, *args, **kw):
if host == 'example.com':
return [(2, 1, 6, '', ('107.6.106.82', 80)),
(2, 1, 6, '', ('107.6.106.82', 80))]
# if host is empty string use None instead
host = object()
- @asyncio.coroutine
- def getaddrinfo(*args, **kw):
+ async def getaddrinfo(*args, **kw):
nonlocal host
host = args[0]
- yield from []
+ return []
def getaddrinfo_task(*args, **kwds):
return asyncio.Task(getaddrinfo(*args, **kwds), loop=self.loop)
MyProto, sock, None, None, mock.ANY, mock.ANY)
def test_call_coroutine(self):
- @asyncio.coroutine
- def simple_coroutine():
- pass
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def simple_coroutine():
+ pass
self.loop.set_debug(True)
coro_func = simple_coroutine
def stop_loop_cb(loop):
loop.stop()
- @asyncio.coroutine
- def stop_loop_coro(loop):
- yield from ()
+ async def stop_loop_coro(loop):
loop.stop()
asyncio.set_event_loop(self.loop)
class RunningLoopTests(unittest.TestCase):
def test_running_loop_within_a_loop(self):
- @asyncio.coroutine
- def runner(loop):
+ async def runner(loop):
loop.run_forever()
loop = asyncio.new_event_loop()
self.assertTrue(repr(lock).endswith('[unlocked]>'))
self.assertTrue(RGX_REPR.match(repr(lock)))
- @asyncio.coroutine
- def acquire_lock():
- with self.assertWarns(DeprecationWarning):
- yield from lock
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def acquire_lock():
+ with self.assertWarns(DeprecationWarning):
+ yield from lock
self.loop.run_until_complete(acquire_lock())
self.assertTrue(repr(lock).endswith('[locked]>'))
def test_lock(self):
lock = asyncio.Lock(loop=self.loop)
- @asyncio.coroutine
- def acquire_lock():
- with self.assertWarns(DeprecationWarning):
- return (yield from lock)
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def acquire_lock():
+ with self.assertWarns(DeprecationWarning):
+ return (yield from lock)
res = self.loop.run_until_complete(acquire_lock())
asyncio.BoundedSemaphore(loop=loop),
]
- @asyncio.coroutine
- def test(lock):
- yield from asyncio.sleep(0.01)
- self.assertFalse(lock.locked())
- with self.assertWarns(DeprecationWarning):
- with (yield from lock) as _lock:
- self.assertIs(_lock, None)
- self.assertTrue(lock.locked())
- yield from asyncio.sleep(0.01)
- self.assertTrue(lock.locked())
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def test(lock):
+ yield from asyncio.sleep(0.01)
self.assertFalse(lock.locked())
+ with self.assertWarns(DeprecationWarning):
+ with (yield from lock) as _lock:
+ self.assertIs(_lock, None)
+ self.assertTrue(lock.locked())
+ yield from asyncio.sleep(0.01)
+ self.assertTrue(lock.locked())
+ self.assertFalse(lock.locked())
for primitive in primitives:
loop.run_until_complete(test(primitive))
def test_context_manager(self):
lock = asyncio.Lock(loop=self.loop)
- @asyncio.coroutine
- def acquire_lock():
- with self.assertWarns(DeprecationWarning):
- return (yield from lock)
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def acquire_lock():
+ with self.assertWarns(DeprecationWarning):
+ return (yield from lock)
with self.loop.run_until_complete(acquire_lock()):
self.assertTrue(lock.locked())
def test_context_manager_cant_reuse(self):
lock = asyncio.Lock(loop=self.loop)
- @asyncio.coroutine
- def acquire_lock():
- with self.assertWarns(DeprecationWarning):
- return (yield from lock)
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def acquire_lock():
+ with self.assertWarns(DeprecationWarning):
+ return (yield from lock)
# This spells "yield from lock" outside a generator.
cm = self.loop.run_until_complete(acquire_lock())
def test_context_manager(self):
cond = asyncio.Condition(loop=self.loop)
- @asyncio.coroutine
- def acquire_cond():
- with self.assertWarns(DeprecationWarning):
- return (yield from cond)
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def acquire_cond():
+ with self.assertWarns(DeprecationWarning):
+ return (yield from cond)
with self.loop.run_until_complete(acquire_cond()):
self.assertTrue(cond.locked())
sem = asyncio.Semaphore(loop=self.loop)
self.assertEqual(1, sem._value)
- @asyncio.coroutine
- def acquire_lock():
- with self.assertWarns(DeprecationWarning):
- return (yield from sem)
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def acquire_lock():
+ with self.assertWarns(DeprecationWarning):
+ return (yield from sem)
res = self.loop.run_until_complete(acquire_lock())
def test_context_manager(self):
sem = asyncio.Semaphore(2, loop=self.loop)
- @asyncio.coroutine
- def acquire_lock():
- with self.assertWarns(DeprecationWarning):
- return (yield from sem)
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def acquire_lock():
+ with self.assertWarns(DeprecationWarning):
+ return (yield from sem)
with self.loop.run_until_complete(acquire_lock()):
self.assertFalse(sem.locked())
asyncio.set_event_loop_policy(None)
-@asyncio.coroutine
-def coroutine_function():
+async def coroutine_function():
pass
def __del__(self):
gc.collect()
- @asyncio.coroutine
- def run():
+ async def run():
return Evil()
self.loop.run_until_complete(
self.loop.run_until_complete(task)
def test_task_class(self):
- @asyncio.coroutine
- def notmuch():
+ async def notmuch():
return 'ok'
t = self.new_task(self.loop, notmuch())
self.loop.run_until_complete(t)
loop.close()
def test_ensure_future_coroutine(self):
- @asyncio.coroutine
- def notmuch():
- return 'ok'
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def notmuch():
+ return 'ok'
t = asyncio.ensure_future(notmuch(), loop=self.loop)
self.loop.run_until_complete(t)
self.assertTrue(t.done())
self.assertIs(f, f_orig)
def test_ensure_future_task(self):
- @asyncio.coroutine
- def notmuch():
+ async def notmuch():
return 'ok'
t_orig = self.new_task(self.loop, notmuch())
t = asyncio.ensure_future(t_orig)
def __await__(self):
return (yield from self.coro)
- @asyncio.coroutine
- def coro():
- return 'ok'
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro():
+ return 'ok'
loop = asyncio.new_event_loop()
self.set_event_loop(loop)
def test_task_repr(self):
self.loop.set_debug(False)
- @asyncio.coroutine
- def notmuch():
- yield from []
+ async def notmuch():
return 'abc'
# test coroutine function
"<Task finished name='TestTask' %s result='abc'>" % coro)
def test_task_repr_autogenerated(self):
- @asyncio.coroutine
- def notmuch():
+ async def notmuch():
return 123
t1 = self.new_task(self.loop, notmuch(), None)
self.loop.run_until_complete(t2)
def test_task_repr_name_not_str(self):
- @asyncio.coroutine
- def notmuch():
+ async def notmuch():
return 123
t = self.new_task(self.loop, notmuch())
def test_task_repr_coro_decorator(self):
self.loop.set_debug(False)
- @asyncio.coroutine
- def notmuch():
- # notmuch() function doesn't use yield from: it will be wrapped by
- # @coroutine decorator
- return 123
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def notmuch():
+ # notmuch() function doesn't use yield from: it will be wrapped by
+ # @coroutine decorator
+ return 123
# test coroutine function
self.assertEqual(notmuch.__name__, 'notmuch')
async def func(x, y):
await asyncio.sleep(0)
- partial_func = asyncio.coroutine(functools.partial(func, 1))
+ with self.assertWarns(DeprecationWarning):
+ partial_func = asyncio.coroutine(functools.partial(func, 1))
task = self.loop.create_task(partial_func(2))
# make warnings quiet
self.assertFalse(t.cancel())
def test_cancel_yield(self):
- @asyncio.coroutine
- def task():
- yield
- yield
- return 12
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def task():
+ yield
+ yield
+ return 12
t = self.new_task(self.loop, task())
test_utils.run_briefly(self.loop) # start coro
loop = asyncio.new_event_loop()
self.set_event_loop(loop)
- @asyncio.coroutine
- def task():
+ async def task():
t.cancel()
self.assertTrue(t._must_cancel) # White-box test.
return 12
foo_started = False
- @asyncio.coroutine
- def foo():
+ async def foo():
nonlocal foo_started
foo_started = True
def test_wait_for_blocking(self):
loop = self.new_test_loop()
- @asyncio.coroutine
- def coro():
+ async def coro():
return 'done'
res = loop.run_until_complete(asyncio.wait_for(coro(), timeout=None))
def test_wait_duplicate_coroutines(self):
- @asyncio.coroutine
- def coro(s):
- return s
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro(s):
+ return s
c = coro('test')
task =self.new_task(
# there is possibility that some tasks in the pending list
# became done but their callbacks haven't all been called yet
- @asyncio.coroutine
- def coro1():
- yield
+ async def coro1():
+ await asyncio.sleep(0)
- @asyncio.coroutine
- def coro2():
- yield
- yield
+ async def coro2():
+ await asyncio.sleep(0)
+ await asyncio.sleep(0)
a = self.new_task(self.loop, coro1())
b = self.new_task(self.loop, coro2())
# first_exception, task already has exception
a = self.new_task(loop, asyncio.sleep(10.0))
- @asyncio.coroutine
- def exc():
+ async def exc():
raise ZeroDivisionError('err')
b = self.new_task(loop, exc())
a = self.new_task(loop, asyncio.sleep(0.1))
- @asyncio.coroutine
- def sleeper():
- yield from asyncio.sleep(0.15)
+ async def sleeper():
+ await asyncio.sleep(0.15)
raise ZeroDivisionError('really')
b = self.new_task(loop, sleeper())
completed = set()
time_shifted = False
- @asyncio.coroutine
- def sleeper(dt, x):
- nonlocal time_shifted
- yield from asyncio.sleep(dt)
- completed.add(x)
- if not time_shifted and 'a' in completed and 'b' in completed:
- time_shifted = True
- loop.advance_time(0.14)
- return x
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def sleeper(dt, x):
+ nonlocal time_shifted
+ yield from asyncio.sleep(dt)
+ completed.add(x)
+ if not time_shifted and 'a' in completed and 'b' in completed:
+ time_shifted = True
+ loop.advance_time(0.14)
+ return x
a = sleeper(0.01, 'a')
b = sleeper(0.01, 'b')
c = sleeper(0.15, 'c')
- @asyncio.coroutine
- def foo():
+ async def foo():
values = []
for f in asyncio.as_completed([b, c, a], loop=loop):
- values.append((yield from f))
+ values.append(await f)
return values
res = loop.run_until_complete(self.new_task(loop, foo()))
def test_as_completed_duplicate_coroutines(self):
- @asyncio.coroutine
- def coro(s):
- return s
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro(s):
+ return s
- @asyncio.coroutine
- def runner():
- result = []
- c = coro('ham')
- for f in asyncio.as_completed([c, c, coro('spam')],
- loop=self.loop):
- result.append((yield from f))
- return result
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def runner():
+ result = []
+ c = coro('ham')
+ for f in asyncio.as_completed([c, c, coro('spam')],
+ loop=self.loop):
+ result.append((yield from f))
+ return result
fut = self.new_task(self.loop, runner())
self.loop.run_until_complete(fut)
loop = self.new_test_loop(gen)
- @asyncio.coroutine
- def sleeper(dt, arg):
- yield from asyncio.sleep(dt/2)
- res = yield from asyncio.sleep(dt/2, arg)
+ async def sleeper(dt, arg):
+ await asyncio.sleep(dt/2)
+ res = await asyncio.sleep(dt/2, arg)
return res
t = self.new_task(loop, sleeper(0.1, 'yeah'))
loop = self.new_test_loop(gen)
- @asyncio.coroutine
- def sleep(dt):
- yield from asyncio.sleep(dt)
+ async def sleep(dt):
+ await asyncio.sleep(dt)
- @asyncio.coroutine
- def doit():
+ async def doit():
sleeper = self.new_task(loop, sleep(5000))
loop.call_later(0.1, sleeper.cancel)
try:
- yield from sleeper
+ await sleeper
except asyncio.CancelledError:
return 'cancelled'
else:
def test_task_cancel_waiter_future(self):
fut = self.new_future(self.loop)
- @asyncio.coroutine
- def coro():
- yield from fut
+ async def coro():
+ await fut
task = self.new_task(self.loop, coro())
test_utils.run_briefly(self.loop)
self.assertTrue(fut.cancelled())
def test_task_set_methods(self):
- @asyncio.coroutine
- def notmuch():
+ async def notmuch():
return 'ko'
gen = notmuch()
'ko')
def test_step_result(self):
- @asyncio.coroutine
- def notmuch():
- yield None
- yield 1
- return 'ko'
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def notmuch():
+ yield None
+ yield 1
+ return 'ko'
self.assertRaises(
RuntimeError, self.loop.run_until_complete, notmuch())
fut = Fut(loop=self.loop)
result = None
- @asyncio.coroutine
- def wait_for_future():
+ async def wait_for_future():
nonlocal result
- result = yield from fut
+ result = await fut
t = self.new_task(self.loop, wait_for_future())
test_utils.run_briefly(self.loop)
loop = self.new_test_loop(gen)
- @asyncio.coroutine
- def sleeper():
- yield from asyncio.sleep(10)
+ async def sleeper():
+ await asyncio.sleep(10)
base_exc = BaseException()
- @asyncio.coroutine
- def notmutch():
+ async def notmutch():
try:
- yield from sleeper()
+ await sleeper()
except asyncio.CancelledError:
raise base_exc
yield
self.assertFalse(asyncio.iscoroutinefunction(fn1))
- @asyncio.coroutine
- def fn2():
- yield
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def fn2():
+ yield
self.assertTrue(asyncio.iscoroutinefunction(fn2))
self.assertFalse(asyncio.iscoroutinefunction(mock.Mock()))
def test_yield_vs_yield_from(self):
fut = self.new_future(self.loop)
- @asyncio.coroutine
- def wait_for_future():
- yield fut
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def wait_for_future():
+ yield fut
task = wait_for_future()
with self.assertRaises(RuntimeError):
self.assertFalse(fut.done())
def test_yield_vs_yield_from_generator(self):
- @asyncio.coroutine
- def coro():
- yield
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro():
+ yield
- @asyncio.coroutine
- def wait_for_future():
- gen = coro()
- try:
- yield gen
- finally:
- gen.close()
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def wait_for_future():
+ gen = coro()
+ try:
+ yield gen
+ finally:
+ gen.close()
task = wait_for_future()
self.assertRaises(
self.loop.run_until_complete, task)
def test_coroutine_non_gen_function(self):
- @asyncio.coroutine
- def func():
- return 'test'
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def func():
+ return 'test'
self.assertTrue(asyncio.iscoroutinefunction(func))
def test_coroutine_non_gen_function_return_future(self):
fut = self.new_future(self.loop)
- @asyncio.coroutine
- def func():
- return fut
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def func():
+ return fut
- @asyncio.coroutine
- def coro():
+ async def coro():
fut.set_result('test')
t1 = self.new_task(self.loop, func())
# 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
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro():
+ # The actual coroutine.
+ self.assertTrue(gen.gi_running)
+ yield from fut
# A completed Future used to run the coroutine.
fut = self.new_future(self.loop)
def test_yield_from_corowrapper(self):
with set_coroutine_debug(True):
- @asyncio.coroutine
- def t1():
- return (yield from t2())
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def t1():
+ return (yield from t2())
- @asyncio.coroutine
- def t2():
- f = self.new_future(self.loop)
- self.new_task(self.loop, t3(f))
- return (yield from f)
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def t2():
+ f = self.new_future(self.loop)
+ self.new_task(self.loop, t3(f))
+ return (yield from f)
- @asyncio.coroutine
- def t3(f):
- f.set_result((1, 2, 3))
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def t3(f):
+ f.set_result((1, 2, 3))
task = self.new_task(self.loop, t1())
val = self.loop.run_until_complete(task)
def test_log_destroyed_pending_task(self):
Task = self.__class__.Task
- @asyncio.coroutine
- def kill_me(loop):
- future = self.new_future(loop)
- yield from future
- # at this point, the only reference to kill_me() task is
- # the Task._wakeup() method in future._callbacks
- raise Exception("code never reached")
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def kill_me(loop):
+ future = self.new_future(loop)
+ yield from future
+ # at this point, the only reference to kill_me() task is
+ # the Task._wakeup() method in future._callbacks
+ raise Exception("code never reached")
mock_handler = mock.Mock()
self.loop.set_debug(True)
loop = asyncio.new_event_loop()
self.set_event_loop(loop)
- @asyncio.coroutine
- def coro():
+ async def coro():
raise TypeError
- @asyncio.coroutine
- def runner():
+ async def runner():
task = self.new_task(loop, coro())
- yield from asyncio.sleep(0.05)
+ await asyncio.sleep(0.05)
task.cancel()
task = None
@mock.patch('asyncio.coroutines.logger')
def test_coroutine_never_yielded(self, m_log):
with set_coroutine_debug(True):
- @asyncio.coroutine
- def coro_noop():
- pass
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro_noop():
+ pass
tb_filename = __file__
tb_lineno = sys._getframe().f_lineno + 2
from @asyncio.coroutine()-wrapped function should have same effect as
returning generator object or Future."""
def check():
- @asyncio.coroutine
- def outer_coro():
+ with self.assertWarns(DeprecationWarning):
@asyncio.coroutine
- def inner_coro():
- return 1
+ def outer_coro():
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def inner_coro():
+ return 1
- return inner_coro()
+ return inner_coro()
result = self.loop.run_until_complete(outer_coro())
self.assertEqual(result, 1)
loop = asyncio.new_event_loop()
self.addCleanup(loop.close)
- @asyncio.coroutine
- def blocking_coroutine():
+ async def blocking_coroutine():
fut = self.new_future(loop)
# Block: fut result is never set
- yield from fut
+ await fut
task = loop.create_task(blocking_coroutine())
def test_exception_traceback(self):
# See http://bugs.python.org/issue28843
- @asyncio.coroutine
- def foo():
+ async def foo():
1 / 0
- @asyncio.coroutine
- def main():
+ async def main():
task = self.new_task(self.loop, foo())
- yield # skip one loop iteration
+ await asyncio.sleep(0) # skip one loop iteration
self.assertIsNotNone(task.exception().__traceback__)
self.loop.run_until_complete(main())
raise ValueError
self.loop.call_soon = call_soon
- @asyncio.coroutine
- def coro():
- pass
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro():
+ pass
self.assertFalse(m_log.error.called)
def test_create_task_with_oldstyle_coroutine(self):
- @asyncio.coroutine
- def coro():
- pass
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro():
+ pass
task = self.new_task(self.loop, coro())
self.assertIsInstance(task, self.Task)
@support.refcount_test
def test_refleaks_in_task___init__(self):
gettotalrefcount = support.get_attribute(sys, 'gettotalrefcount')
- @asyncio.coroutine
- def coro():
+ async def coro():
pass
task = self.new_task(self.loop, coro())
self.loop.run_until_complete(task)
self.assertAlmostEqual(gettotalrefcount() - refs_before, 0, delta=10)
def test_del__log_destroy_pending_segfault(self):
- @asyncio.coroutine
- def coro():
+ async def coro():
pass
task = self.new_task(self.loop, coro())
self.loop.run_until_complete(task)
def wrap_futures(self, *futures):
coros = []
for fut in futures:
- @asyncio.coroutine
- def coro(fut=fut):
- return (yield from fut)
+ async def coro(fut=fut):
+ return await fut
coros.append(coro())
return coros
def test_constructor_loop_selection(self):
- @asyncio.coroutine
- def coro():
+ async def coro():
return 'abc'
gen1 = coro()
gen2 = coro()
self.other_loop.run_until_complete(fut2)
def test_duplicate_coroutines(self):
- @asyncio.coroutine
- def coro(s):
- return s
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro(s):
+ return s
c = coro('abc')
fut = asyncio.gather(c, c, coro('def'), c, loop=self.one_loop)
self._run_loop(self.one_loop)
proof = 0
waiter = asyncio.Future(loop=self.one_loop)
- @asyncio.coroutine
- def inner():
+ async def inner():
nonlocal proof
- yield from waiter
+ await waiter
proof += 1
child1 = asyncio.ensure_future(inner(), loop=self.one_loop)
child2 = asyncio.ensure_future(inner(), loop=self.one_loop)
gatherer = None
- @asyncio.coroutine
- def outer():
+ async def outer():
nonlocal proof, gatherer
gatherer = asyncio.gather(child1, child2, loop=self.one_loop)
- yield from gatherer
+ await gatherer
proof += 100
f = asyncio.ensure_future(outer(), loop=self.one_loop)
def test_exception_marking(self):
# Test for the first line marked "Mark exception retrieved."
- @asyncio.coroutine
- def inner(f):
- yield from f
+ async def inner(f):
+ await f
raise RuntimeError('should not be ignored')
a = asyncio.Future(loop=self.one_loop)
b = asyncio.Future(loop=self.one_loop)
- @asyncio.coroutine
- def outer():
- yield from asyncio.gather(inner(a), inner(b), loop=self.one_loop)
+ async def outer():
+ await asyncio.gather(inner(a), inner(b), loop=self.one_loop)
f = asyncio.ensure_future(outer(), loop=self.one_loop)
test_utils.run_briefly(self.one_loop)
self.loop = asyncio.new_event_loop()
self.set_event_loop(self.loop) # Will cleanup properly
- @asyncio.coroutine
- def add(self, a, b, fail=False, cancel=False):
+ async def add(self, a, b, fail=False, cancel=False):
"""Wait 0.05 second and return a + b."""
- yield from asyncio.sleep(0.05)
+ await asyncio.sleep(0.05)
if fail:
raise RuntimeError("Fail!")
if cancel:
asyncio.current_task(self.loop).cancel()
- yield
+ await asyncio.sleep(0)
return a + b
def target(self, fail=False, cancel=False, timeout=None,
nonlocal result
result += num
- @asyncio.coroutine
- def coro():
+ async def coro():
self.loop.call_soon(inc_result, 1)
self.assertEqual(result, 0)
- num = yield from asyncio.sleep(0, result=10)
+ num = await asyncio.sleep(0, result=10)
self.assertEqual(result, 1) # inc'ed by call_soon
inc_result(num) # num should be 11
def test_yield_from_awaitable(self):
- @asyncio.coroutine
- def coro():
- yield from asyncio.sleep(0)
- return 'ok'
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro():
+ yield from asyncio.sleep(0)
+ return 'ok'
result = self.loop.run_until_complete(coro())
self.assertEqual('ok', result)
def test_await_old_style_coro(self):
- @asyncio.coroutine
- def coro1():
- return 'ok1'
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro1():
+ return 'ok1'
- @asyncio.coroutine
- def coro2():
- yield from asyncio.sleep(0)
- return 'ok2'
+ with self.assertWarns(DeprecationWarning):
+ @asyncio.coroutine
+ def coro2():
+ yield from asyncio.sleep(0)
+ return 'ok2'
async def inner():
return await asyncio.gather(coro1(), coro2(), loop=self.loop)
projects / python / commitdiff