V2EX = way to explore
V2EX 是一个关于分享和探索的地方
现在注册
已注册用户请  登录
推荐学习书目
Learn Python the Hard Way
Python Sites
PyPI - Python Package Index
http://diveintopython.org/toc/index.html
Pocoo
值得关注的项目
PyPy
Celery
Jinja2
Read the Docs
gevent
pyenv
virtualenv
Stackless Python
Beautiful Soup
结巴中文分词
Green Unicorn
Sentry
Shovel
Pyflakes
pytest
Python 编程
pep8 Checker
Styles
PEP 8
Google Python Style Guide
Code Style from The Hitchhiker's Guide
aiqier
V2EX  ›  Python

如何给 tornado 做一个 mock 类?

  •  
  •   aiqier · 2016-04-24 12:55:16 +08:00 · 3728 次点击
    这是一个创建于 2929 天前的主题,其中的信息可能已经有所发展或是发生改变。

    我所在的环境是, tornado4.3,python2.7 目前在给自己的服务写单元测试, 我想要让我的单元测试完全独立于第三方环境(比如一个三方的 url 接口)。 首先,我尝试封装了这样两个函数:

    @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 的单测要做特殊处理?

    8 条回复    2016-04-25 12:35:50 +08:00
    calease
        1
    calease  
       2016-04-24 13:09:37 +08:00   ❤️ 1
    一般的做法是 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
        2
    jmp2x  
       2016-04-24 13:17:52 +08:00
    1. 单测没有必要异步吧
    2. tornado 要想把异步封装到函数模块里面有点蛋疼,因为函数执行到一半返回了父函数,父函数没有拿到结果就继续执行了。这里有点小 trick 的实现方式,不知道适不适合。 http://jmpews.com/posts/tornado-yield-module-design.html
    aiqier
        3
    aiqier  
    OP
       2016-04-24 13:21:55 +08:00
    @jmp2x 你的代码里面,一个函数又有 yield ,又有 return ,连“编译”都过不去吧?
    jmp2x
        4
    jmp2x  
       2016-04-24 13:27:15 +08:00
    @aiqier 可以的,一个 yield 是单纯生成器,一 yield 是 tornado 的异步,你仔细看下就区分开了。
    neoblackcap
        5
    neoblackcap  
       2016-04-24 15:16:39 +08:00
    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
        6
    jadecoder  
       2016-04-24 16:56:26 +08:00   ❤️ 1
    tornado 有异步的单元测试类 tornado.testing.AsyncTestCase

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

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

    就可以调用异步函数了

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

    这么说不知道你明不明白
    fengchang
        7
    fengchang  
       2016-04-24 17:00:46 +08:00
    1L 的思路是对的,贴一个自己的 stub client

    https://gist.github.com/fengchang/1292819deaf5f60dae8d927ab6b873e7
    keakon
        8
    keakon  
       2016-04-25 12:35:50 +08:00
    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)
    ```

    如果能控制对象的生成过程,那就不需要写成类方法了。
    关于   ·   帮助文档   ·   博客   ·   API   ·   FAQ   ·   我们的愿景   ·   实用小工具   ·   2252 人在线   最高记录 6543   ·     Select Language
    创意工作者们的社区
    World is powered by solitude
    VERSION: 3.9.8.5 · 28ms · UTC 05:47 · PVG 13:47 · LAX 22:47 · JFK 01:47
    Developed with CodeLauncher
    ♥ Do have faith in what you're doing.