Howto do unittest for tornado + async def?

Environment: Python 3, tornado 4.4. The normal unittests cannot be used because methods are asynchronous. There is ttp://www.tornadoweb.org/en/stable/testing.html that explains how to do unit testing for asynchronous code. But that works with tornado coroutines ONLY. The classes I want to test are using the async def statements, and they cannot be tested this way. For example, here is a test case that uses ASyncHTTPClient.fetch and its callback parameter:

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)

But my methods are declared like this:

class Connection: async def get_data(url, *args): # ....

And there is no callback. How can I "await" for this method from a test case?

UPDATE: based on Jessie's answer, I created this 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()

The result is this:

>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

There is no traceback. But if I replace tornado.testing.main() with unittest.main() then it suddenly starts working.

But why? I guessed that for asnyc unit tests, I need to use tornado.testing.main ( http://www.tornadoweb.org/en/stable/testing.html#tornado.testing.main )

I'm confused.

UPDATE 2: It is a bug in tornado.testing. Workaround:

all = MyTestCase2
main()

---

Instead of using the self.wait / self.stop callbacks, you can wait for "fetch" to complete by using it in an "await" expression:

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()

The other change I had to make in your code is to call "decode" on the body in order to compare the body (which is bytes) to "FriendFeed" which is a string.

链接地址: http://www.djcxy.com/p/27858.html

上一篇: 如何申请！使用.css（）重要？

下一篇: 如何做单元测试龙卷风+异步def？