Tornado异步单元测试

1 前言

1.1 单元测试

1.2 unittest 包

python 内置了 unittest 包，包含了通用的单元测试所需要的基本功能， 使用方法也非常简单：

import unittest
class TestStringMethods(unittest.TestCase):
    def test_upper(self):
        self.assertEqual('foo'.upper(), 'FOO')
    def test_isupper(self):
        self.assertTrue('FOO'.isupper())
        self.assertFalse('Foo'.isupper())
    def test_split(self):
        s = 'hello world'
        self.assertEqual(s.split(), ['hello', 'world'])
        # check that s.split fails when the separator is not a string
        with self.assertRaises(TypeError):
            s.split(2)
if __name__ == '__main__':
    unittest.main()

每一个单元测试类继承 unittest.TestCase ，每一个测试方法以 test_ 开头， self.assert__ 方法为单元测试判断比较。 unittest.main() 启动全部测试。在子类中可以重载 setUp 和 tearDown 方法，在这里可以做一些测试初始化和清理功能，比如数据路连接和关闭等等。一旦 assert 相关方法失败，则抛出异常。

1.3 代码覆盖率

单元测试还有一个很重要的目的是测试代码的覆盖率， coverage 包提供了检查代码覆盖率的功能，而且能够输出网页版的查看工具。

import unittest
import coverage
COV = None
COV = coverage.coverage(branch=True, include="./*", omit=["ENV/*", "_run.py", "test/*", "pep8/*", "*/__init__.py"])
COV.start()
def _test():
    tests = unittest.TestLoader().discover("test")
    unittest.TextTestRunner(verbosity=2).run(tests)
    COV.stop()
    COV.save()
    COV.report()
    basedir = os.path.abspath(os.path.dirname(__file__))
    covdir = os.path.join(basedir, "tmp/coverage")
    COV.html_report(directory=covdir)
    COV.erase()

coverage 函数中 include 参数给出检查单元覆盖率的文件夹， omit 参数给出忽略的文件夹。 html_report 将单元测试覆盖率输出到文件到指定文件夹中。在启动 _test 方法后，可以查看每一行代码是否被单元测试覆盖到。

2 异步方法测试

一般的方法编写测试非常简单，但是如果使用 tornado 编写的异步方法和函数该如何编写单元测试呢？没关系， tornado 提供了 tornado.testing 包，包含了 AsyncTestCase 和 AsyncHTTPTestCase 类， 该类继承 unittest.TestCase , 因此也包含了之前的 assert 相关的方法。 gen_test 装饰器用来异步的单元测试中，将 yield 的异步操作变成同步操作，该装饰器 timeout 参数用来指明这个异步单元测试方法的时间界限，超出时间将会引发单元测试失败异常。

# func.py
from tornado.httpclient import HTTPRequest, AsyncHTTPClient
from tornado.gen import coroutine, Return
@coroutine
def run():
    request = HTTPRequest("https://gaufung.com")
    http_client = AsyncHTTPClient()
    response = yield http_client.fetch(request)
    raise Return(response.code)
    
# test_func.py
import unittest
from tornado.testing import AsyncTestCase, gen_test
from func import run
class TestFunc(AsyncTestCase):
    @gen_test(timeout=5)
    def test_run():
        code = yield run()
        self.assertEqual(code, 200)

每一个单元测试只能有一个 IOLoop 实例，如果在单元测试中使用 IOLoop ，则必须要重载 setUp 方法，否则将无法使用 self.io_loop 。

class TestFunc(AsyncTestCase):
    def setUp(self):
        super(TestFunc, self).setUp()
        pass

3 Tornado Web应用API测试

除了对函数和方法的单元测试，还需要对 Web 接口进行单元测试，单元测试类继承 AsyncHTTPTestCase , 并且重载 get_app 方法才可以进行单元测试。

# app.py
import tornado.ioloop
import tornado.web
from tornado.gen import coroutine, Return
from tornado.httpclient import AsyncHTTPClient, HTTPRequest
@coroutine
def blog():
    request = HTTPRequest("https://gaufung.com")
    http_client = AsyncHTTPClient(ioloop=IOLoop.current())
    response = yield http_client.fetch(request)
    raise Return(response.body)

class MainHandler(tornado.web.RequestHandler):
    @coroutine
    def get(self):
        body = yield blog()
        self.write(body)

def make_app():
    return tornado.web.Application([
        (r"/gaufung", MainHandler),
    ])
    
# app_test.py
import unittest
from tornado.testing import gen_test, AsyncHTTPTestCase
import app
class TestApp(AsyncHTTPTestCase):
    def get_app(self):
        return app.make_app()
    
    @gen_test(timeout=5)
    def test_gaufung(self):
        response = yield self.http_client.fetch(self.get_url("/gaufung"))
        self.assertNotNone(response)

make_app 方法返回 app.py 中创建的 Application 类， self.get_url 获取特定接口完整的 url 。

4 异步方法 Mock

在测试中往往需要依赖外部内容，比如向外部服务器发送HTTP请求来获取相应的状态。但是搭建外部服务器通常费时费力，而且不符合单元测试的依赖性要求。因此 Python 提供了 Mock 库，可以模拟外部请求并且返回内容。普通的请求 Mock 使用可以查看标准库内容，而且使用非常简单。但是对于异步的请求需要如下的 Mock 方法：

import tornado.ioloop
import tornado.web
from tornado.gen import coroutine, Return
from tornado.httpclient import AsyncHTTPClient, HTTPRequest

@coroutine
def blog():
    request = HTTPRequest("https://gaufung.com")
    http_client = AsyncHTTPClient(ioloop=IOLoop.current())
    response = yield http_client.fetch(request)
    raise Return(response.body)

class MainHandler(tornado.web.RequestHandler):
    @coroutine
    def get(self):
        body = yield blog()
        self.write(body)

def make_app():
    return tornado.web.Application([
        (r"/gaufung", MainHandler),
    ])

# app_test.py
import app
import mock
import unittest
from tornado.testing import gen_test, AsyncHTTPTestCase
from tornado.httpclient import AsyncHTTPClient
from tornado.concurrent import Future
class TestApp(AsyncHTTPTestCase):
    def get_app(self):
        return app.make_app()

    @mock.patch("app.blog")
    @gen_test
    def test_get(self, blog):
        blog_future = Future()
        blog_future.set_result("gaufung's blog")
        blog.return_value = blog_future
        response = yield self.http_client.fetch(self.get_url("/gaufung")
        self.assertEqual(response.code, 200)
        self.assertEqal(response.body, "gaufung's blog")