如何测试 Python 3.4 的 asyncio 代码？

Question

如何为使用 Python 3.4 的 asyncio 库编写单元测试？假设我想测试一个 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指令执行之前执行，之后返回而未执行任何断言。这导致测试始终成功。

是否有预构建的测试运行程序可以处理此类异步代码？

- Marvin Killing

---

你可以使用loop.run_until_complete()代替yield from。另请参阅asyncio.test_utils。 - jfs

对于Python 3.5+的async def和await语法，请参见：https://dev59.com/Ep3ha4cB1Zd3GeqPV4zx - Udi

Answer 1

自Python 3.8起，unittest已经提供了IsolatedAsyncioTestCase函数，专门为此而设计。

from unittest import IsolatedAsyncioTestCase

class Test(IsolatedAsyncioTestCase):

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

- Nico Rikken

---

很遗憾，直到今天为止，至少需要经过5个变通才能看到这个答案。 - koks der drache

马文·基林可以接受这个答案，这可能会改变... - Malcolm

这是目前为止最好的解决方案。 - Andrey Nelubin

这个答案真的应该被标记为被采纳的解决方案！ - Victor Wong

非常好的答案。这拯救了我的生命！异步测试不应该成为Python开发人员永久的停滞和沮丧问题。 - mondayrris

Answer 2

我受到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

像J.F.Sebastian建议的那样，这个装饰器会一直阻塞，直到测试方法协程完成。这使我能够编写这样的测试用例：

与J.F. 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())

这个解决方案可能缺少一些边缘情况的考虑。

我认为像这样的功能应该被添加到Python标准库中，以便使asyncio和unittest之间的交互更加方便。

- Marvin Killing

---

有没有一种方法可以修改这个解决方案，使装饰器使用特定的循环，而不是线程默认的循环？ - Sebastian

是的，在Python中，函数注释可以带参数，因此您可以在那里传递事件循环。请注意，编写带有参数的注释起初可能会有点困惑：https://dev59.com/-2025IYBdhLWcg3wkG17#5929165 - Jack O'Connor

@JackO'Connor 我认为你的意思是函数_decorator_而不是函数_annotations_，因为函数_annotations_在Python中有特定的含义：https://docs.python.org/3/tutorial/controlflow.html#function-annotations - Dustin Wyatt

我在使用asyncio.get_event_loop()时遇到了问题，于是使用了asyncio.new_event_loop()。 - James

警告：asyncio.coroutine已被弃用，并将在py3.10中删除：https://docs.python.org/3/library/asyncio-task.html#generator-based-coroutines - metaperture

同时，使用"get_event_loop"来创建或获取循环已被弃用，它只会获取从3.11版本开始运行的当前循环。 - Eric Burel

Answer 3

async_test，Marvin Killing 建议的方法，肯定会有帮助 - 直接调用 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被执行时才完成的长时间协同程序等奇怪的错误。

- Andrew Svetlov

---

为什么您要执行 asyncio.set_event_loop(None)，然后稍后明确传递 self.loop 到 asyncio.open_connection()，而不是一开始就执行 asyncio.set_event_loop(self.loop) 呢？ - balu

这只是我的习惯而已。当我在使用asyncio和aio-based库时，我会使用asyncio.set_event_loop(None) 直接指定该库不应该依赖于全局事件循环的存在，并通过明确的循环传递来安全地工作。这是asyncio测试本身的代码风格，我也在我的库中使用它。 - Andrew Svetlov

这个例子也应该模拟asyncio.open_connection，不是吗？运行它会产生ConnectionRefusedError: [Errno 61] Connect call failed ('127.0.0.1', 8888)。 - terrycojones

@terrycojones mock并不总是必需的。例如，我使用本地地址，因此我可以在测试运行之前或在“setUp”方法中设置测试服务器的地址。具体实现取决于您的需求。 - Andrew Svetlov

增加更多的样板代码，但这绝对是使测试单元化和隔离化的方法。 - danius

记得在tearDown中关闭循环。 - math2001

Answer 4

非常喜欢在https://dev59.com/NmAg5IYBdhLWcg3w5ugd#23036785中提到的async_test装饰器，这里是适用于Python 3.5+的更新版本。

def async_test(coro):
    def wrapper(*args, **kwargs):
        loop = asyncio.new_event_loop()
        try:
            return loop.run_until_complete(coro(*args, **kwargs))
        finally:
            loop.close()
    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())

- peralmq

---

对于使用nosetests的任何人，您可能想要重命名装饰器，否则nose也会将其视为一个测试，并显示有关“async_test”缺少所需位置参数的神秘消息。我将它重新命名为asynctest并添加了另一个装饰器@nose.tools.istest以使测试用例能够自动发现。 - patricksurry

如果使用 nosetests，请使用 nose.tools.nottest 装饰器包装 async_test。 - millerdev

这很美丽。将其纳入我的项目中。谢谢！ - binarymason

在我的Odoo安装中，尽管我正在使用Python 3.10，但使用'IsolatedAsyncioTestCase'的最佳答案不起作用，但是这个方法可行；谢谢！ - LFLFM

Answer 5

pytest-asyncio 看起来很有前途:

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

- ostrokach

---

使用unittest.TestCase时，pytest的方法存在问题，对我来说非常有限。http://jacobbridges.github.io/post/unit-testing-with-asyncio/ - kwarunek

看起来他们在这里提交了一个问题。目前还没有解决方案。 https://github.com/pytest-dev/pytest-asyncio/issues/15 - James

同时，通过mock.patch来模拟类也无法正常工作。https://github.com/pytest-dev/pytest-asyncio/issues/42 - Deviacium

Answer 6

您还可以使用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)

正如您所看到的，async情况由AsyncTestCase处理。它也支持同步测试。可以提供自定义事件循环的可能性，只需覆盖AsyncTestCase.get_event_loop。

如果您更喜欢其他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)

- kwarunek

Answer 7

请使用这个类作为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)

编辑1：

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

- pylover

---

这是一个很棒的解决方案。在这里添加了一个小改动：https://dev59.com/NmAg5IYBdhLWcg3w5ugd#60986764 - Nitay

请为您的代码添加描述。仅提供代码不是答案。 - buhtz

Answer 8

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

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

- jcazor

Answer 9

我认为pylover的回答是正确的，应该被添加到单元测试中。我会加入一些修改来支持嵌套异步测试：

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

- Nitay

Answer 10

我发现Python测试文件中有一个类似于Marvin Killing答案的“async_test”函数。因为自Python 3.8起，“@coroutine”装饰器已被弃用，所以当我使用Python3.8或更高版本时，会收到“DeprecationWarning”的警告。

如果您使用的是Python 3.5+，那么这个答案可能是一个不错的选择。希望能够帮助到您。

import asyncio
import functools


def async_test(func):
    """Decorator to turn an async function into a test case."""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        coro = func(*args, **kwargs)
        asyncio.run(coro)
    return wrapper

测试示例：

import unittest


async def add_func(a, b):
    return a + b


class TestSomeCase(unittest.TestCase):
    @async_test
    async def test_add_func(self):
        self.assertEqual(await add_func(1, 2), 3)

- J.C