Създавам някои тестове за код на python3, използвайки py.test. Кодът осъществява достъп до база данни на Postgresql с помощта на aiopg (базиран на Asyncio интерфейс към postgres).
Основните ми очаквания:
Всеки тестов случай трябва да има достъп до нов асинхронен цикъл на събития.
Тест, който работи твърде дълго, ще спре с изключение за изчакване.
Всеки тестов случай трябва да има достъп до връзка с база данни.
Не искам да се повтарям, когато пиша тестовите случаи.
Използвайки py.test fixtures, мога да се доближа до това, което искам, но все пак трябва да се повтарям малко във всеки асинхронен тестов случай.
Ето как изглежда моят код:
@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)
Мога да живея с решението, което имам досега, но може да стане тромаво да дефинирам вътрешна съпрограма и да добавям реда run_timeout за всеки асинхронен тест, който пиша.
Искам тестовете ми да изглеждат донякъде така:
@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 python модули за създаване на такъв магически декоратор, но изглежда, че и двата модула ми помагат да създам функция със сигнатура, идентична на my_test, което не е добро решение в такъв случай.
Това вероятно може да бъде разрешено с помощта на eval или подобен хак, но се чудех дали има нещо елегантно, което пропускам тук.
