如何对 tornado + async def 进行单元测试?

Howto do unittest for tornado + async def?

环境:Python3,龙卷风4.4。不能使用正常的单元测试,因为方法是异步的。有 ttp://www.tornadoweb.org/en/stable/testing.html 解释了如何对异步代码进行单元测试。但这仅适用于龙卷风协程。我要测试的 classes 正在使用 async def 语句,并且无法通过这种方式进行测试。例如,这是一个使用 ASyncHTTPClient.fetch 及其回调参数的测试用例:

class MyTestCase2(AsyncTestCase):
    def test_http_fetch(self):
        client = AsyncHTTPClient(self.io_loop)
        client.fetch("http://www.tornadoweb.org/", self.stop)
        response = self.wait()
        # Test contents of response
        self.assertIn("FriendFeed", response.body)

但是我的方法是这样声明的:

class 连接: 异步定义 get_data(url, *args): # ....

并且没有回调。我如何 "await" 从测试用例中获取此方法?

更新: 根据 Jessie 的回答,我创建了这个 MWE:

import unittest

from tornado.httpclient import AsyncHTTPClient
from tornado.testing import AsyncTestCase, gen_test, main


class MyTestCase2(AsyncTestCase):
    @gen_test
    async def test_01(self):
        await self.do_more()

    async def do_more(self):
        self.assertEqual(1+1, 2)

main()

结果是这样的:

>py -3 -m test.py
E
======================================================================
ERROR: all (unittest.loader._FailedTest)
----------------------------------------------------------------------
AttributeError: module '__main__' has no attribute 'all'

----------------------------------------------------------------------
Ran 1 test in 0.000s

FAILED (errors=1)
[E 170205 10:05:43 testing:731] FAIL

没有回溯。但是如果我用 unittest.main() 替换 tornado.testing.main() 然后它突然开始工作。

但是为什么呢?我猜想对于 asnyc 单元测试,我需要使用 tornado.testing.main ( http://www.tornadoweb.org/en/stable/testing.html#tornado.testing.main )

我很困惑。

更新 2: 这是 tornado.testing 中的错误。解决方法:

all = MyTestCase2
main()

不使用 self.wait / self.stop 回调,您可以等待 "fetch" 完成,方法是在 "await" 表达式中使用它:

import unittest

from tornado.httpclient import AsyncHTTPClient
from tornado.testing import AsyncTestCase, gen_test


class MyTestCase2(AsyncTestCase):
    @gen_test
    async def test_http_fetch(self):
        client = AsyncHTTPClient(self.io_loop)
        response = await client.fetch("http://www.tornadoweb.org/")
        # Test contents of response
        self.assertIn("FriendFeed", response.body.decode())

unittest.main()

我必须在您的代码中进行的另一个更改是在主体上调用 "decode" 以便将主体(字节)与 "FriendFeed" 进行比较,后者是一个字符串。