使用 asyncio 事件循环运行 tornado.testing.AsyncTestCase答案

【问题标题】：Run tornado.testing.AsyncTestCase using asyncio event loop使用 asyncio 事件循环运行 tornado.testing.AsyncTestCase
【发布时间】：2016-06-28 07:16:14
【问题描述】：

我有一个基于异步的类，我想对其进行单元测试。使用tornado.testing.AsyncTestCase 这很容易工作。但是，我班级的一个特定方法使用asyncio.ensure_future 来安排另一种方法的执行。这永远不会在 AsyncTestCase 中完成，因为默认测试运行器使用 tornado KQueueIOLoop 事件循环，而不是 asyncio 事件循环。

class TestSubject:
    def foo(self):
        asyncio.ensure_future(self.bar())

    async def bar(self):
        pass

class TestSubjectTest(AsyncTestCase):
    def test_foo(self):
        t = TestSubject()
        # here be somewhat involved setup with MagicMock and self.stop
        t.foo()
        self.wait()

$ python -m tornado.testing baz.testsubject_test
...
[E 160627 17:48:22 testing:731] FAIL
[E 160627 17:48:22 base_events:1090] Task was destroyed but it is pending!
    task: <Task pending coro=<TestSubject.bar() running at ...>>
.../asyncio/base_events.py:362: RuntimeWarning: coroutine 'TestSubject.bar' was never awaited

如何使用不同的事件循环来运行测试以确保我的任务会真正执行？或者，如何使我的实现事件循环独立和交叉兼容？

【问题讨论】：

标签： python unit-testing python-3.x tornado python-asyncio

【解决方案1】：

事实证明很简单......

class TestSubjectTest(AsyncTestCase):
    def get_new_ioloop(self):  # override this method
        return tornado.platform.asyncio.AsyncIOMainLoop()

我之前试过这个，但是直接返回asyncio.get_event_loop()，没用。返回 Tornado 的 asyncio 循环包装器就可以解决问题。

【讨论】：