py.test 픽스쳐와 asyncio coroutine을 섞기

Question

py.test를 사용하여 python3 코드에 대한 테스트를하고 있습니다. 이 코드는 aiopg (Postgres에 대한 Asyncio 기반 인터페이스)를 사용하여 PostgreSQL 데이터베이스에 액세스합니다.

내 주요 기대 :

• 모든 테스트 케이스는 새로운 asyncio 이벤트 루프에 액세스 할 수 있어야합니다.

• 너무 오래 실행되는 테스트는 시간 초과 예외로 중지됩니다.

• 모든 테스트 케이스는 데이터베이스 연결에 대한 액세스 권한을 가져야합니다.

• 테스트 사례 작성시 자신을 반복하고 싶지 않습니다.

py.test fixtures를 사용하면 원하는 결과에 가깝지만 모든 비동기 테스트 케이스에서 약간의 반복을해야합니다. 내가 지금까지 가지고있는 솔루션을 살 수 있지만 내부 코 루틴을 정의하는 성가신 얻을 수있는 모든 비동기 테스트의 run_timeout 라인을 추가

@pytest.fixture(scope='function') 
def tloop(request): 
    # This fixture is responsible for getting a new event loop 
    # for every test, and close it when the test ends. 
    ... 

def run_timeout(cor,loop,timeout=ASYNC_TEST_TIMEOUT): 
    """ 
    Run a given coroutine with timeout. 
    """ 
    task_with_timeout = asyncio.wait_for(cor,timeout) 
    try: 
     loop.run_until_complete(task_with_timeout) 
    except futures.TimeoutError: 
     # Timeout: 
     raise ExceptAsyncTestTimeout() 


@pytest.fixture(scope='module') 
def clean_test_db(request): 
    # Empty the test database. 
    ... 

@pytest.fixture(scope='function') 
def udb(request,clean_test_db,tloop): 
    # Obtain a connection to the database using aiopg 
    # (That's why we need tloop here). 
    ... 


# An example for a test: 
def test_insert_user(tloop,udb): 
    @asyncio.coroutine 
    def insert_user(): 
     # Do user insertion here ... 
     yield from udb.insert_new_user(... 
     ... 

    run_timeout(insert_user(),tloop) 

:

이처럼 내 코드는 모습입니다 내가 쓰는 것. 나는 섬세한 방법으로 이러한 장식을 만들기 위해 시도했지만 실패

@some_magic_decorator 
def test_insert_user(udb): 
    # Do user insertion here ... 
    yield from udb.insert_new_user(... 
    ... 

:

나는 나의 테스트는 다음과 같이 다소보고 싶어요. 내 테스트가 보이는 경우보다 일반적으로, 같은 :

는

@some_magic_decorator 
def my_test(arg1,arg2,...,arg_n): 
    ... 

그런 다음 생성 기능 (데코레이터가 적용된 후)해야한다 :

def my_test_wrapper(tloop,arg1,arg2,...,arg_n): 
    run_timeout(my_test(),tloop) 

주 내 시험의 일부가 UDB 외에 (다른기구를 사용하는 것이 예제), 그 조명기는 생성 된 함수의 인자로 나타나야한다. 그렇지 않으면 py.test가 그것들을 호출하지 않을 것이다.

wrapt과 decorator 파이썬 모듈을 사용하여 이러한 마법 장식자를 만들려고했지만 두 모듈 모두 my_test와 동일한 서명을 가진 함수를 만드는 데 도움이되는 것처럼 보입니다.이 경우에는 좋은 해결책이 아닙니다.

이것은 아마도 eval 또는 유사한 해킹을 사용하여 해결할 수 있지만 여기에 누락 된 우아한 것이 있는지 궁금합니다.

Answer 1

저는 현재 비슷한 문제를 해결하려고합니다. 지금까지 제가 지금까지 생각해 냈습니다. 작동하는 것처럼 보이지만 정리가 필요합니다.

# tests/test_foo.py 
import asyncio 

@asyncio.coroutine 
def test_coro(loop): 
    yield from asyncio.sleep(0.1) 
    assert 0 

# tests/conftest.py 
import asyncio 


@pytest.yield_fixture 
def loop(): 
    loop = asyncio.new_event_loop() 
    asyncio.set_event_loop(loop) 
    yield loop 
    loop.close() 


def pytest_pycollect_makeitem(collector, name, obj): 
    """Collect asyncio coroutines as normal functions, not as generators.""" 
    if asyncio.iscoroutinefunction(obj): 
     return list(collector._genfunctions(name, obj)) 


def pytest_pyfunc_call(pyfuncitem): 
    """If ``pyfuncitem.obj`` is an asyncio coroutinefunction, execute it via 
    the event loop instead of calling it directly.""" 
    testfunction = pyfuncitem.obj 

    if not asyncio.iscoroutinefunction(testfunction): 
     return 

    # Copied from _pytest/python.py:pytest_pyfunc_call() 
    funcargs = pyfuncitem.funcargs 
    testargs = {} 
    for arg in pyfuncitem._fixtureinfo.argnames: 
     testargs[arg] = funcargs[arg] 
    coro = testfunction(**testargs) # Will no execute the test yet! 

    # Run the coro in the event loop 
    loop = testargs.get('loop', asyncio.get_event_loop()) 
    loop.run_until_complete(coro) 

    return True # TODO: What to return here? 

그래서 기본적으로 pytest는 정상적인 기능처럼 asyncio coroutines를 수집합니다. 나는 또한 함수에 대한 텍스트 exectuion을 가로 채기도한다. 테스트 할 함수가 코 루틴 인 경우 이벤트 루프에서 실행합니다. 테스트마다 새로운 이벤트 루프 인스턴스를 생성하는 조명기와 함께 또는 조명기없이 작동합니다.

편집 : Ronny Pfannschmidt에 따르면, 2.7 출시 이후 pytest에 이와 비슷한 내용이 추가됩니다. :-)

Answer 2

모든 테스트 케이스는 새로운 asyncio 이벤트 루프에 액세스 할 수 있어야합니다.

테스트 세트 asyncio는 unittest.TestCase를 사용합니다. setUp() 메서드를 사용하여 새 이벤트 루프를 만듭니다. addCleanup (loop.close)은 오류 발생시에도 자동으로 이벤트 루프를 닫습니다.

죄송합니다, 당신이 TestCase를 사용하고 싶지 않으면 py.test로 이것을 쓰는 법을 모르겠습니다. 하지만 올바르게 기억하면 py.test는 unittest.TestCase를 지원합니다.

너무 오래 실행되는 테스트는 시간 초과 예외로 중지됩니다.

loop.call_later()는 BaseException을 감시견으로 발생시키는 함수와 함께 사용할 수 있습니다.