如何testingPython 3.4 asyncio代码？

使用Python 3.4 asyncio库为代码编写unit testing的最佳方式是什么？ 假设我想testing一个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()) 

当使用默认testing运行器运行此testing用例时，testing将始终成功，因为该方法只执行直到第一次yield from指令产生，然后在执行任何断言之前返回。 这导致testing总是成功。

是否有预构build的testing运行器能够处理这样的asynchronous代码？

Answer 1

async_test ，由Marvin Killingbuild议，绝对可以帮助 – 以及直接调用loop.run_until_complete()

但我也强烈build议为每个testing重新创build一个新的事件循环，并将循环直接传递给API调用（至lessasyncio本身只接受每个需要它的调用的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()) 

它隔离testing用例中的testing，并防止奇怪的错误，如test_a创build的test_a协程，但仅在test_b执行时间结束。

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 

就像JF Sebastianbuild议的那样，这个装饰器将会阻塞，直到testing方法协程完成。 这使我可以写这样的testing用例：

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

这个解决scheme可能会错过一些边缘情况。

我认为像这样的设施应该添加到Python的标准库，以使asyncio和unittest交互更加方便。

Answer 3

pytest-asyncio看起来很有希望：

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

Answer 4

使用这个类而不是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) 

Answer 5

你也可以使用aiounittest Svetlov类似的方法，@Marvin杀死答案并将其包装在易于使用的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) 

正如你所看到的asynchronous情况是由AsyncTestCase处理的。 它也支持同步testing。 有可能提供自定义的事件循环，只是覆盖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) 

Answer 6

真的很喜欢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()) 

Answer 7

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

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