如何给 tornado 做一个 mock 类?

@tornado.gen.coroutine def put_get_request_into_ioloop(context, url, origin=False): """ 把一个异步 get 请求扔到 tornado ioloop 中 :param url: 请求 url :param handler_id: 处理 id :param origin: 是否日志请求和响应原串 :return: True, 返回结果 False, None """ if origin: logger.info("Request:%s\t%s" % (context["handler_id"], url)) try: client = tornado.httpclient.AsyncHTTPClient() respOnse= yield client.fetch(url) if origin: logger.info("Response:%s\t%s" % (context["handler_id"], response.body)) except: logger.error("ErrorRequest:%s\t%s" % (context["handler_id"], url), exc_info=True) raise tornado.gen.Return((False, None)) else: raise tornado.gen.Return((True, response.body)) @tornado.gen.coroutine def put_post_request_into_ioloop(context, url, body, origin=False): """ 把一个异步 post 请求扔到 tornado ioloop 中 :param url: 请求 url :param body: 请求体 :param handler_id: 处理 id :param origin: 是否日志请求和响应原串 :return: True, 返回结果 False, None """ if origin: logger.info("Request:%s\t%s\t%s" % (context["handler_id"], url, body)) try: client = tornado.httpclient.AsyncHTTPClient() respOnse= yield client.fetch(url, method="POST", body=body) if origin: logger.info("Response:%s\t%s" % (context["handler_id"], response.body)) except: logger.error("ErrorRequest:%s\t%s\t%s" % (context["handler_id"], url, body), exc_info=True) raise tornado.gen.Return((False, None)) else: raise tornado.gen.Return((True, response.body))

在我的系统中，我在调用第三方接口的时候，会调用这两个函数。

我想要用 mock 的方法，替换这种函数的调用返回结果，使我的系统独立，便于单元测试。但是有 yield 和 coroutine 的“干扰”，我不知道应该如何实现，替换这个结果，并能运行我的单元测试。我尝试过写一个函数返回自己 raise tornaod.gen.Return,然后用 MethodType 的方式，对其进行替换。但是没能跑起来。我想知道： 1.我的这种思路是否对？ 2.是不是因为 ioloop 的启动方式有误， tornado 的单测要做特殊处理？

URL

handler_id

context

Origin

8 replies 2016-04-25 12:35:50 +08:00

calease

Apr 24, 2016

一般的做法是 initialize 一个 future object ， set_result 然后 mock yield 。 initialize+set_result 可以用 maybe_future(deprecated)代替。
比如 client.fetch 返回的是一个 future object ，你自己新建一个 future object,用 set_result()设置你想要的 return ，然后 mock client.fetch 的 return_value 。
我不知道 tornado 有没有自带的 mocked httpclient ，如果有的话也可以用。
最后提醒一点 unit test class 必须是 AsyncTestCase 的 child class 。

jmp2x

Apr 24, 2016

1. 单测没有必要异步吧
2. tornado 要想把异步封装到函数模块里面有点蛋疼，因为函数执行到一半返回了父函数，父函数没有拿到结果就继续执行了。这里有点小 trick 的实现方式，不知道适不适合。 http://jmpews.com/posts/tornado-yield-module-design.html

aiqier

Apr 24, 2016

@jmp2x 你的代码里面，一个函数又有 yield ，又有 return ，连“编译”都过不去吧？

jmp2x

Apr 24, 2016

@aiqier 可以的，一个 yield 是单纯生成器，一 yield 是 tornado 的异步，你仔细看下就区分开了。

neoblackcap

Apr 24, 2016

1. 直接按普通的 mock 返回结果就可以了，比如像一楼说的。 tornado 测试的时候会堵塞 IOLoop 直到返回结果，你当同步代码来用就可以了。比如你将 gen.coroutine 换成 tornado.testing.gen_test ，那么你执行的代码的时候就是堵塞。

2. 之所以要 raise tornado.gen.Return 是一个 hack ，是 python 2.7 时代不允许 generator 里面有 return 的 hack ，因此就通过 raise 一个异常来解决这个问题。你若是使用 python 3.5 就可以直接 return 。而且你还可以使用 async 跟 await 两个关键字来达到更好的语义。

jadecoder

Apr 24, 2016

tornado 有异步的单元测试类 tornado.testing.AsyncTestCase

假设你的异步函数是 def process(self, callback)

object.process(self.stop)
ret = self.wait()

就可以调用异步函数了

如果你要 mock tornado 的 httpclient 的话，就写一个类，实现 fetch 方法，从你指定的数据源取了数据然后 callback 就可以了。

这么说不知道你明不明白

fengchang

Apr 24, 2016

1L 的思路是对的，贴一个自己的 stub client

https://gist.github.com/fengchang/1292819deaf5f60dae8d927ab6b873e7

keakon

Apr 25, 2016

mock client 就行了，最简单的大概这样：

```
class MockedAsyncHTTPClient(AsyncHTTPClient):
_respOnses= []

@classmethod
def set_responses(cls, responses): # 需要是逆序的
cls._respOnses= responses

def fetch_impl(self, request, callback):
respOnse= self._responses.pop()
response.request = request
callback(response)
```

如果能控制对象的生成过程，那就不需要写成类方法了。