Skip to content

Commit

Permalink
Accept async fixtures in pytest plugin (aio-libs#2223)
Browse files Browse the repository at this point in the history
  • Loading branch information
k4nar committed Aug 25, 2017
1 parent d951059 commit ee02dc1
Show file tree
Hide file tree
Showing 2 changed files with 55 additions and 0 deletions.
7 changes: 7 additions & 0 deletions aiohttp/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import cgi
import datetime
import functools
import inspect
import os
import re
import sys
Expand Down Expand Up @@ -821,3 +822,9 @@ def update_cookies(self, cookies, response_url=None):

def filter_cookies(self, request_url):
return None


def isasyncgenfunction(obj):
if hasattr(inspect, 'isasyncgenfunction'):
return inspect.isasyncgenfunction(obj)
return False
48 changes: 48 additions & 0 deletions aiohttp/pytest_plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import pytest
from py import path

from aiohttp.helpers import isasyncgenfunction
from aiohttp.web import Application

from .test_utils import unused_port as _unused_port
Expand Down Expand Up @@ -37,6 +38,53 @@ def pytest_addoption(parser):
help='enable event loop debug mode')


def pytest_fixture_setup(fixturedef, request):
"""
Allow fixtures to be coroutines. Run coroutine fixtures in an event loop.
"""
func = fixturedef.func

if isasyncgenfunction(func):
# async generator fixture
is_async_gen = True
elif asyncio.iscoroutinefunction(func):
# regular async fixture
is_async_gen = False
else:
# not an async fixture, nothing to do
return

# if neither the fixture nor the test use the 'loop' fixture,
# 'getfixturevalue' will fail because the test is not parameterized
# (this can be removed someday if 'loop' is no longer parameterized)
if 'loop' not in request.fixturenames:
raise Exception(
"Asynchronous fixtures must depend on the 'loop' fixture or be "
"used in tests depending from it."
)

def wrapper(*args, **kwargs):
_loop = request.getfixturevalue('loop')

if is_async_gen:
# for async generators, we need to advance the generator once,
# then advance it again in a finalizer
gen = func(*args, **kwargs)

def finalizer():
try:
return _loop.run_until_complete(gen.__anext__())
except StopAsyncIteration:
pass

request.addfinalizer(finalizer)
return _loop.run_until_complete(gen.__anext__())
else:
return _loop.run_until_complete(func(*args, **kwargs))

fixturedef.func = wrapper


@pytest.fixture
def fast(request):
""" --fast config option """
Expand Down

0 comments on commit ee02dc1

Please sign in to comment.