如何测试 Python 3.4 异步代码？

How to test Python 3.4 asyncio code?

提问by Marvin Killing

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

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

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 frominstruction, after which it returns before executing any assertions. This causes tests to always succeed.

使用默认测试运行器运行此测试用例时，测试将始终成功，因为该方法仅执行到第一yield from条指令，之后在执行任何断言之前返回。这会导致测试总是成功。

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

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

回答by Marvin Killing

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

我使用受 Tornado 的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:

就像 JF Sebastian 建议的那样，这个装饰器将阻塞，直到测试方法协程完成。这允许我编写这样的测试用例：

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 asyncioand unittestinteraction more convenient out of the box.

我觉得像这样的设施应加入到了Python的标准库，使asyncio和unittest互动更方便开箱。

回答by Andrew Svetlov

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

async_test，由 Marvin Killing 建议，绝对可以提供帮助 - 以及直接调用 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 asyncioitself accepts loopkeyword-only parameter for every call that need it).

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

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_abut finished only on test_bexecution time.

隔离测试用例中的测试并防止奇怪的错误，例如已在其中创建test_a但仅在test_b执行时完成的长期协程。

回答by pylover

Use this class instead of unittest.TestCasebase class:

使用这个类而不是unittest.TestCase基类：

import asyncio
import unittest


class AioTestCase(unittest.TestCase):

    # noinspection PyPep8Naming
    def __init__(self, methodName='runTest', loop=None):
        self.loop = loop or asyncio.get_event_loop()
        self._function_cache = {}
        super(AioTestCase, self).__init__(methodName=methodName)

    def coroutine_function_decorator(self, func):
        def wrapper(*args, **kw):
            return self.loop.run_until_complete(func(*args, **kw))
        return wrapper

    def __getattribute__(self, item):
        attr = object.__getattribute__(self, item)
        if asyncio.iscoroutinefunction(attr):
            if item not in self._function_cache:
                self._function_cache[item] = self.coroutine_function_decorator(attr)
            return self._function_cache[item]
        return attr


class TestMyCase(AioTestCase):

    async def test_dispatch(self):
        self.assertEqual(1, 1)

EDIT 1:

编辑 1：

Please note the @Nitay answerabout nested tests.

请注意关于嵌套测试的@Nitay回答。

回答by ostrokach

pytest-asynciolooks promising:

pytest-asyncio看起来很有希望：

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

回答by jcazor

I usually define my async tests as coroutines and use a decorator for "syncing" them:

我通常将我的异步测试定义为协程并使用装饰器来“同步”它们：

import asyncio
import unittest

def sync(coro):
    def wrapper(*args, **kwargs):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(coro(*args, **kwargs))
    return wrapper

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

    @sync
    async def test_sends_handshake_after_connect(self):
        await self.socket_connection.connect()
        self.assertTrue(self.mock_server.received_handshake())

回答by kwarunek

You can also use aiounittestthat takes similar approach as @Andrew Svetlov, @Marvin Killing answers and wrap it in easy to use AsyncTestCaseclass:

您还可以使用aiounittest与@Andrew Svetlov 类似的方法，@Marvin Killing 回答并将其包装在易于使用的AsyncTestCase类中：

import asyncio
import aiounittest


async def add(x, y):
    await asyncio.sleep(0.1)
    return x + y

class MyTest(aiounittest.AsyncTestCase):

    async def test_async_add(self):
        ret = await add(5, 6)
        self.assertEqual(ret, 11)

    # or 3.4 way
    @asyncio.coroutine
    def test_sleep(self):
        ret = yield from add(5, 6)
        self.assertEqual(ret, 11)

    # some regular test code
    def test_something(self):
        self.assertTrue(true)

As you can see the async case is handled by AsyncTestCase. It supports also synchronous test. There is a possibility to provide custom event loop, just override AsyncTestCase.get_event_loop.

如您所见，异步情况由AsyncTestCase. 它还支持同步测试。有可能提供自定义事件循环，只需覆盖AsyncTestCase.get_event_loop.

If you prefer (for some reason) the other TestCase class (eg unittest.TestCase), you might use async_testdecorator:

如果您更喜欢（出于某种原因）其他 TestCase 类（例如unittest.TestCase），您可以使用async_test装饰器：

import asyncio
import unittest
from aiounittest import async_test


async def add(x, y):
    await asyncio.sleep(0.1)
    return x + y

class MyTest(unittest.TestCase):

    @async_test
    async def test_async_add(self):
        ret = await add(5, 6)
        self.assertEqual(ret, 11)

回答by peralmq

Really like the async_testwrapper mentioned in https://stackoverflow.com/a/23036785/350195, here is an updated version for Python 3.5+

真的很像https://stackoverflow.com/a/23036785/350195 中async_test提到的包装器，这里是 Python 3.5+ 的更新版本

def async_test(coro):
    def wrapper(*args, **kwargs):
        loop = asyncio.new_event_loop()
        return loop.run_until_complete(coro(*args, **kwargs))
    return wrapper



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

    @async_test
    async def test_sends_handshake_after_connect(self):
        await self.socket_connection.connect()
        self.assertTrue(self.mock_server.received_handshake())

回答by Nico Rikken

Since Python 3.8 unittestcomes with the IsolatedAsyncioTestCasefunction, designed for this purpose.

由于 Python 3.8 unittest附带了IsolatedAsyncioTestCase函数，专为此目的而设计。

from unittest import IsolatedAsyncioTestCase

class Test(IsolatedAsyncioTestCase):

    async def test_functionality(self):
        result = await functionality()
        self.assertEqual(expected, result)

回答by shimon

In addition to pylover's answer, if you intend to use some other asynchronous method from the test class itself, the following implementation will work better -

除了 pylover 的答案，如果您打算使用测试类本身的其他一些异步方法，以下实现会更好地工作 -

import asyncio
import unittest

class AioTestCase(unittest.TestCase):

    # noinspection PyPep8Naming
    def __init__(self, methodName='runTest', loop=None):
        self.loop = loop or asyncio.get_event_loop()
        self._function_cache = {}
        super(AioTestCase, self).__init__(methodName=methodName)

    def coroutine_function_decorator(self, func):
        def wrapper(*args, **kw):
            return self.loop.run_until_complete(func(*args, **kw))
        return wrapper

    def __getattribute__(self, item):
        attr = object.__getattribute__(self, item)
        if asyncio.iscoroutinefunction(attr) and item.startswith('test_'):
            if item not in self._function_cache:
                self._function_cache[item] = 
                    self.coroutine_function_decorator(attr)
            return self._function_cache[item]
        return attr


class TestMyCase(AioTestCase):

    async def multiplier(self, n):
        await asyncio.sleep(1)  # just to show the difference
        return n*2

    async def test_dispatch(self):
        m = await self.multiplier(2)
        self.assertEqual(m, 4)

the only change was - and item.startswith('test_')in the __getattribute__method.

唯一的变化是 -and item.startswith('test_')在__getattribute__方法上。

回答by Nitay

pylover answer is correct and is something that should be added to unittest IMO.

pylover 答案是正确的，应该添加到 unittest IMO 中。

I would add in a slight change to support nested async tests:

我会添加一个小改动来支持嵌套的异步测试：

class TestCaseBase(unittest.TestCase):
    # noinspection PyPep8Naming
    def __init__(self, methodName='runTest', loop=None):
        self.loop = loop or asyncio.get_event_loop()
        self._function_cache = {}
        super(BasicRequests, self).__init__(methodName=methodName)

    def coroutine_function_decorator(self, func):
        def wrapper(*args, **kw):
            # Is the io loop is already running? (i.e. nested async tests)
            if self.loop.is_running():
                t = func(*args, **kw)
            else:
                # Nope, we are the first
                t = self.loop.run_until_complete(func(*args, **kw))
            return t

        return wrapper

    def __getattribute__(self, item):
        attr = object.__getattribute__(self, item)
        if asyncio.iscoroutinefunction(attr):
            if item not in self._function_cache:
                self._function_cache[item] = self.coroutine_function_decorator(attr)
            return self._function_cache[item]
        return attr