如何测试Python 3.4异步代码？
如何测试Python 3.4异步代码？
提问于 2014-04-13 01:43:34
使用Python3.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指令之前执行，在此之后，它在执行任何断言之前返回。这会导致测试始终成功。
有没有一个预建的测试运行器能够处理这样的异步代码？
回答 9
Stack Overflow用户
发布于 2019-12-14 17:41:29
因为Python3.8 unittest附带了IsolatedAsyncioTestCase函数，就是为此目的而设计的。
from unittest import IsolatedAsyncioTestCase
class Test(IsolatedAsyncioTestCase):
async def test_functionality(self):
result = await functionality()
self.assertEqual(expected, result)Stack Overflow用户
发布于 2014-04-13 06:14:15
我使用了一个受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建议的那样，这个装饰器将被阻塞，直到测试方法协程结束。这使我可以像这样编写测试用例：
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的交互变得更加方便。
Stack Overflow用户
发布于 2014-05-14 05:55:33
Marvin Killing建议的async_test当然可以提供帮助，也可以直接调用loop.run_until_complete()
但我也强烈建议为每个测试重新创建新的事件循环，并将循环直接传递给应用编程接口调用(对于每个需要的调用，至少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执行时完成的长期协程。
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/23033939
复制相关文章
相似问题