如何测试Python3.4异步代码？

Question

使用Python 3.4asyncio库为代码编写单元测试的最佳方法是什么？假设我要测试TCP客户端（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())

当使用默认测试运行程序运行此测试用例时，测试将始终成功，因为该方法只执行到第一条yield from指令，然后在执行任何断言之前返回。这使得测试总是成功的。

是否有一个预构建的测试运行程序能够处理这样的异步代码？

Answer 1

我用一个装饰师临时解决了这个问题，他的灵感来自于龙卷风的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

正如J.F.Sebastian建议的那样，这个装饰器将阻塞，直到测试方法coroutine完成为止。这允许我编写这样的测试用例：

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

这个解决方案可能遗漏了一些边缘情况。

我认为这样的工具应该添加到Python的标准库中，使asyncio和unittest的交互更加方便。

Answer 2

由Marvin Killing建议的async_test绝对可以帮助——以及直接调用loop.run_until_complete()

但我也强烈建议为每个测试重新创建新的事件循环，并直接将循环传递给API调用（至少asyncio本身对每个需要它的调用都只接受loop关键字参数）。

就像

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

它隔离测试用例中的测试，并防止出现奇怪的错误，如在test_a中创建但仅在test_b执行时完成的长期协同路由。

Answer 3

pytest-asyncio看起来很有前途：

@pytest.mark.asyncio
async def test_some_asyncio_code():
    res = await library.do_something()
    assert b'expected result' == res