Python 3.4 asyncio 코드를 테스트하는 방법은 무엇입니까?
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
명령 까지만 실행되고 그 후에는 어설 션을 실행하기 전에 반환 되므로 테스트는 항상 성공 합니다. 이로 인해 테스트가 항상 성공합니다.
이와 같은 비동기 코드를 처리 할 수있는 미리 빌드 된 테스트 실행기가 있습니까?
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 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())
이 솔루션은 아마도 일부 엣지 케이스를 놓칠 수 있습니다.
나는이 같은 시설이 파이썬의 표준 수 있도록 라이브러리에 추가해야한다고 생각 asyncio
하고 unittest
상호 작용 상자보다 편리 밖으로.
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
실행 시간 에만 완료된 오랜 코 루틴과 같은 이상한 오류를 방지 합니다.
pytest-asyncio 는 유망 해 보입니다.
@pytest.mark.asyncio
async def test_some_asyncio_code():
res = await library.do_something()
assert b'expected result' == res
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())
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)
당신은 또한 사용할 수 있습니다 aiounittest
그 답을 죽이는 @Marvin, 앤드류 Svetlov와 유사한 접근 방식을 취하고 및 사용에 쉽게에서 포장 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)
저는 일반적으로 비동기 테스트를 코 루틴으로 정의하고 "동기화"를 위해 데코레이터를 사용합니다.
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())
참고URL : https://stackoverflow.com/questions/23033939/how-to-test-python-3-4-asyncio-code
'Program Tip' 카테고리의 다른 글
100,000 개 이상의 개별 IP 주소를 차단하는 방법 (0) | 2020.11.14 |
---|---|
Entity Framework 6.0에서 마이그레이션을 비활성화하는 방법 (0) | 2020.11.14 |
RxJS에서`Scheduler` 란 무엇입니까? (0) | 2020.11.14 |
자바 스크립트에서 새 위치 객체 만들기 (0) | 2020.11.13 |
Visual Studio에서 실행하기 전에 F5가 프로젝트를 다시 빌드하지 않는 이유는 무엇입니까? (0) | 2020.11.13 |