Async pytest plugin

pytest_plugin.py

from functools import partial
from inspect import iscoroutinefunction

import pytest

import hyperio

try:
    from async_generator import isasyncgenfunction
except ImportError:
    from inspect import isasyncgenfunction


@pytest.hookimpl(hookwrapper=True)
def pytest_fixture_setup(fixturedef, request):
    def wrapper(*args, **kwargs):
        backend = kwargs['hyperio_backend']
        if strip_backend:
            del kwargs['hyperio_backend']

        if isasyncgenfunction(func):
            gen = func(*args, **kwargs)
            try:
                value = hyperio.run(gen.__anext__, backend=backend)
            except StopAsyncIteration:
                raise RuntimeError('Async generator did not yield')

            yield value

            try:
                hyperio.run(gen.__anext__, backend=backend)
            except StopAsyncIteration:
                pass
            else:
                hyperio.run(gen.aclose)
                raise RuntimeError('Async generator fixture did not stop')
        else:
            yield hyperio.run(partial(func, *args, **kwargs), backend=backend)

    func = fixturedef.func
    if hasattr(func, '__wrapped__'):
        func = func.__wrapped__

    if isasyncgenfunction(func) or iscoroutinefunction(func):
        strip_backend = False
        if 'hyperio_backend' not in fixturedef.argnames:
            fixturedef.argnames += ('hyperio_backend',)
            strip_backend = True

        fixturedef.func = wrapper

    yield


def pytest_generate_tests(metafunc):
    if metafunc.definition.get_closest_marker('hyperio'):
        metafunc.fixturenames.append('hyperio_backend')

        # If an explicit hyperio_backend was defined, skip parametrization
        for marker in metafunc.definition.iter_markers(name='parametrize'):
            if 'hyperio_backend' in marker.args:
                return

        metafunc.parametrize('hyperio_backend', ['asyncio', 'curio', 'trio'], scope='session')


@pytest.mark.tryfirst
def pytest_pyfunc_call(pyfuncitem):
    if pyfuncitem.get_closest_marker('hyperio'):
        funcargs = pyfuncitem.funcargs
        backend = funcargs.get('hyperio_backend', 'asyncio')
        testargs = {arg: funcargs[arg] for arg in pyfuncitem._fixtureinfo.argnames}
        hyperio.run(partial(pyfuncitem.obj, **testargs), backend=backend)
        return True

test_pytest_plugin.py

import pytest
from async_generator import async_generator, yield_

from hyperio import sleep


@pytest.fixture
async def async_fixture():
    await sleep(0)
    return 'foo'


@pytest.fixture
@async_generator
async def asyncgen_fixture():
    await sleep(0)
    await yield_('foo')
    await sleep(0)


@pytest.mark.hyperio
async def test_fixture(async_fixture):
    assert async_fixture == 'foo'


@pytest.mark.hyperio
async def test_asyncgen_fixture(asyncgen_fixture):
    assert asyncgen_fixture == 'foo'