Python 3.4 비동기 코드를 테스트하는 방법은 무엇입니까?

Python 3.4 비동기 코드를 테스트하는 방법은 무엇입니까?

파이썬 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명령을 실행하기 전에 명령이 반환됩니다.이로 인해 검정이 항상 성공합니다.

이와 같은 비동기 코드를 처리할 수 있는 사전 구축된 테스트 러너가 있습니까?

Python 3.8 유닛 테스트는 이러한 목적으로 설계된 IsolatedAsyncioTestCase 기능과 함께 제공됩니다.

from unittest import IsolatedAsyncioTestCase

class Test(IsolatedAsyncioTestCase):

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

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.처럼.세바스찬은 이 장식가가 테스트 방법 코루틴이 끝날 때까지 차단할 것이라고 제안했습니다.이를 통해 다음과 같은 테스트 사례를 작성할 수 있습니다.

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마빈 킬링이 제안한 것은 직통 전화뿐만 아니라 확실히 도움이 될 수 있습니다.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실행 시간

정말 좋아요.async_testhttps://stackoverflow.com/a/23036785/350195, 에 언급된 래퍼는 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())

pytest-florcio는 유망해 보입니다.

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

당신은 또한 @Andrew Svetlov, @Marvin Killing이 대답하고 그것을 사용하기 쉽게 포장하는 것과 유사한 접근법을 사용할 수 있습니다.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)

다음 대신 이 클래스 사용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 답변에 유의하십시오.

저는 보통 비동기 테스트를 코루틴으로 정의하고 "동기화"를 위해 데코레이터를 사용합니다.

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

파이러버 답변이 올바르고 유닛 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

저는 python 테스트 파일이 Marvin Killing의 답변과 유사한 'async_test' 기능을 가지고 있다는 것을 발견했습니다.왜냐하면 "@coroutine" decorator는 Python 3.8 이후로 더 이상 사용되지 않기 때문입니다.python 3.8 이상을 사용하는 경우.저는 "비권고 경고"를 받았습니다.

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

테스트 예:

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)

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)

- 유한변입화 - 다니는일▁- 입다▁the ▁was니▁change.and item.startswith('test_')에 시대에__getattribute__방법.

언급URL : https://stackoverflow.com/questions/23033939/how-to-test-python-3-4-asyncio-code