>

自由软件精神——“自由、开放、分享”。自由软件自诞生之日起,就秉承了学术自由的思想,信奉科学无国界,知识应该全人类共享。

ubuntu精神——人道待人,天下共享连接人人的信念。具有 ubuntu 精神的人心胸开阔,乐于助人,见贤思齐而不忌妒贤能......

tornado gen_test含义

下面这段代码来自tornado-redis的测试,类似gen_test的实现:


def async_test_ex(timeout=5):
    def _inner(func):
        def _runner(self, *args, **kwargs):
            try:
                func(self, *args, **kwargs)
            except:
                self.stop()
                raise
            return self.wait(timeout=timeout)
        return _runner
    return _inner

def async_test(func):
    _inner = async_test_ex()
    return _inner(func)

 

http://stackoverflow.com/questions/23033939/how-to-test-python-3-4-asyncio-code

What's the best way to write unit tests for code using the Python 3.4 asyncio library? Assume I want to test a TCP client (SocketConnection):

import asyncio
import unittest

class TestSocketConnection(unittest.TestCase):
    def setUp(self):
        self.mock_server = MockServer("localhost", 1337)
        self.socket_connection = SocketConnection("localhost", 1337)

    @asyncio.coroutine
    def test_sends_handshake_after_connect(self):
        yield from self.socket_connection.connect()
        self.assertTrue(self.mock_server.received_handshake())

When running this test case with the default test runner, the test will always succeed as the method executes only up until the first yield from instruction, after which it returns before executing any assertions. This causes tests to always succeed.

Is there a prebuilt test runner that is able to handle asynchronous code like this?

 

I temporarily solved the problem using a decorator inspired by Tornado's gen_test:

def async_test(f):
    def wrapper(*args, **kwargs):
        coro = asyncio.coroutine(f)
        future = coro(*args, **kwargs)
        loop = asyncio.get_event_loop()
        loop.run_until_complete(future)
    return wrapper

Like J.F. Sebastian suggested, this decorator will block until the test method coroutine has finished. This allows me to write test cases like this:

class TestSocketConnection(unittest.TestCase):
    def setUp(self):
        self.mock_server = MockServer("localhost", 1337)
        self.socket_connection = SocketConnection("localhost", 1337)

    @async_test
    def test_sends_handshake_after_connect(self):
        yield from self.socket_connection.connect()
        self.assertTrue(self.mock_server.received_handshake())

This solution probably misses some edge cases.

I think a facility like this should added to Python's standard library to make asyncio and unittest interaction more convenient out of the box.

 

每次测试new一个ioloop是好的做法,见下!

async_test, suggested by Marvin Killing, definitely can help -- as well as direct calling loop.run_until_complete()

But I also strongly recommend to recreate new event loop for every test and directly pass loop to API calls (at least asyncio itself accepts loop keyword-only parameter for every call that need it).

Like

class Test(unittest.TestCase):
    def setUp(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(None)

    def test_xxx(self):
        @asyncio.coroutine
        def go():
            reader, writer = yield from asyncio.open_connection('127.0.0.1', 8888, loop=self.loop)
            yield from asyncio.sleep(0.01, loop=self.loop)
        self.loop.run_until_complete(go())

that isolates tests in test case and prevents strange errors like longstanding coroutine that has been created in test_a but finished only on test_b execution time.

Meta

发布: 二月 27, 2015

作者: Bone Lee 李智华

评论:  

字数统计: 467

下一篇: torando单元测试

上一篇: 二叉树的前序中序后序遍历的非递归实现

Bookmark and Share

Tags

code python test tornado

文章链接

  1. unit testing - How to test Python 3.4 asyncio code? ...
  2. tornado.testing — Unit testing support for asynchronous code — Tornado ...
comments powered by Disqus

返回顶部