可以将文章内容翻译成中文,广告屏蔽插件可能会导致该功能失效(如失效,请关闭广告屏蔽插件后再试):
问题:
What's the best way to write unit tests for code using the Python 3.4 asyncio
library? Assume I want to test a TCP client (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 from
instruction, after which it returns before executing any assertions. This causes tests to always succeed.
Is there a prebuilt test runner that is able to handle asynchronous code like this?
回答1:
async_test
, suggested by Marvin Killing, definitely can help -- as well as direct calling 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 asyncio
itself accepts loop
keyword-only parameter for every call that need it).
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_a
but finished only on test_b
execution time.
回答2:
I temporarily solved the problem using a decorator inspired by Tornado's 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:
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 asyncio
and unittest
interaction more convenient out of the box.
回答3:
pytest-asyncio looks promising:
@pytest.mark.asyncio
async def test_some_asyncio_code():
res = await library.do_something()
assert b'expected result' == res
回答4:
Use this class instead of unittest.TestCase
base class:
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)
回答5:
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()
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())
回答6:
You can also use aiounittest
that takes similar approach as @Andrew Svetlov, @Marvin Killing answers and wrap it in easy to use AsyncTestCase
class:
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
.
If you prefer (for some reason) the other TestCase class (eg unittest.TestCase
), you might use async_test
decorator:
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)
回答7:
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())