如何测试 Python 3.4异步代码？

使用 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指令，之后在执行任何断言之前返回。这使得测试总是成功的。

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

50453 次浏览
小开

我暂时解决了这个问题，使用了一个设计师，灵感来自龙卷风的 基因测试:

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

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

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

小开

由马文杀手建议的 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())

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.

小开

使用此类代替 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 answer。

小开

Pytest-syncio 看起来很有希望:

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

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

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

你也可以使用与@Andrew Svetlov 类似的 aiounittest,@Marvin Kill 回答这个问题，然后用简单易用的 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)

正如您所看到的，异步案例由 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)
小开

Really like the async_test wrapper mentioned in https://stackoverflow.com/a/23036785/350195, here is an updated version for 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())
小开

因为 Python 3.8 团结一致附带了 分离的 AsyncioTestCase函数，就是为此而设计的。

from unittest import IsolatedAsyncioTestCase


class Test(IsolatedAsyncioTestCase):


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

除了 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)

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

小开

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

为了支持嵌套的异步测试，我要做一些小小的改变:

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
小开

I found python test file have a similar 'async_test' function like Marvin Killing's answer. Because "@coroutine" decorator is deprecated since Python 3.8. When I use python3.8 or above. I got a "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

Test example:

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)

提交答案