我们从Python开源项目中,提取了以下6个代码示例,用于说明如何使用asyncio.coroutines()。
def test_yield_from_corowrapper_send(self): def foo(): a = yield return a def call(arg): cw = asyncio.coroutines.CoroWrapper(foo()) cw.send(None) try: cw.send(arg) except StopIteration as ex: return ex.args[0] else: raise AssertionError('StopIteration was expected') self.assertEqual(call((1, 2)), (1, 2)) self.assertEqual(call('spam'), 'spam')
def set_coroutine_debug(enabled): coroutines = asyncio.coroutines old_debug = coroutines._DEBUG try: coroutines._DEBUG = enabled yield finally: coroutines._DEBUG = old_debug
def test_current_task_with_interleaving_tasks(self): self.assertIsNone(asyncio.Task.current_task(loop=self.loop)) fut1 = asyncio.Future(loop=self.loop) fut2 = asyncio.Future(loop=self.loop) @asyncio.coroutine def coro1(loop): self.assertTrue(asyncio.Task.current_task(loop=loop) is task1) yield from fut1 self.assertTrue(asyncio.Task.current_task(loop=loop) is task1) fut2.set_result(True) @asyncio.coroutine def coro2(loop): self.assertTrue(asyncio.Task.current_task(loop=loop) is task2) fut1.set_result(True) yield from fut2 self.assertTrue(asyncio.Task.current_task(loop=loop) is task2) task1 = asyncio.Task(coro1(self.loop), loop=self.loop) task2 = asyncio.Task(coro2(self.loop), loop=self.loop) self.loop.run_until_complete(asyncio.wait((task1, task2), loop=self.loop)) self.assertIsNone(asyncio.Task.current_task(loop=self.loop)) # Some thorough tests for cancellation propagation through # coroutines, tasks and wait().
def test_corowrapper_weakref(self): wd = weakref.WeakValueDictionary() def foo(): yield from [] cw = asyncio.coroutines.CoroWrapper(foo()) wd['cw'] = cw # Would fail without __weakref__ slot. cw.gen = None # Suppress warning from __del__.
def run(self, *coros: CoroWrapper): """ Runs an arbitrary list of coroutines in order and then quits the loop, if not running as a context manager. """ if not self.running: raise RuntimeError("not running!") async def wrapper(): results = [] for coro in coros: try: if inspect.isawaitable(coro): results.append(await coro) elif inspect.isfunction(coro): res = coro() if inspect.isawaitable(res): results.append(await res) else: results.append(res) else: raise RuntimeError( "don't know how to run {}".format(coro)) except Exception as ex: logger.error("Error while running coroutine {}: {}" .format(coro.__name__, ex.__repr__())) raise ex if len(results) == 1: return results[0] return results if coros: what = wrapper() else: # if no coros supplied, then assume we run forever what = self.runFut return self.loop.run_until_complete(what)
def test_task_repr(self): self.loop.set_debug(False) @asyncio.coroutine def notmuch(): yield from [] return 'abc' # test coroutine function self.assertEqual(notmuch.__name__, 'notmuch') if PY35: self.assertEqual(notmuch.__qualname__, 'TaskTests.test_task_repr.<locals>.notmuch') self.assertEqual(notmuch.__module__, __name__) filename, lineno = test_utils.get_function_source(notmuch) src = "%s:%s" % (filename, lineno) # test coroutine object gen = notmuch() if coroutines._DEBUG or PY35: coro_qualname = 'TaskTests.test_task_repr.<locals>.notmuch' else: coro_qualname = 'notmuch' self.assertEqual(gen.__name__, 'notmuch') if PY35: self.assertEqual(gen.__qualname__, coro_qualname) # test pending Task t = asyncio.Task(gen, loop=self.loop) t.add_done_callback(Dummy()) coro = format_coroutine(coro_qualname, 'running', src, t._source_traceback, generator=True) self.assertEqual(repr(t), '<Task pending %s cb=[<Dummy>()]>' % coro) # test cancelling Task t.cancel() # Does not take immediate effect! self.assertEqual(repr(t), '<Task cancelling %s cb=[<Dummy>()]>' % coro) # test cancelled Task self.assertRaises(asyncio.CancelledError, self.loop.run_until_complete, t) coro = format_coroutine(coro_qualname, 'done', src, t._source_traceback) self.assertEqual(repr(t), '<Task cancelled %s>' % coro) # test finished Task t = asyncio.Task(notmuch(), loop=self.loop) self.loop.run_until_complete(t) coro = format_coroutine(coro_qualname, 'done', src, t._source_traceback) self.assertEqual(repr(t), "<Task finished %s result='abc'>" % coro)