def test_yield_future_not_running(self):
result = []
def func(event, fut):
event.send('link')
value = aioeventlet.yield_future(fut)
result.append(value)
self.loop.stop()
event = eventlet.event.Event()
fut = asyncio.Future(loop=self.loop)
eventlet.spawn(func, event, fut)
event.wait()
self.loop.call_soon(fut.set_result, 21)
self.loop.run_forever()
self.assertEqual(result, [21])
评论列表
文章目录