klen/pytest-aio

Is a simple pytest plugin for testing async python code

pytest-aio

pytest-aio -- Is a simple pytest plugin for testing any async python code

Features

• Supports all most popular async python libraries: Asyncio, Trio and Curio
• Automatically run your async tests
• Works with contextvars correctly (supports it for async/sync fixtures)
• Supports trio-asyncio

Contents

• pytest-aio
  • Features
• Requirements
• Installation
• Usage
  • Async fixtures for sync tests
  • Customize async libraries
  • Helpers
• Bug tracker
• Contributing
• License

Requirements

• python >= 3.7

Installation

pytest-aio should be installed using pip:

pip install pytest-aio

optionally extras are available:

pip install pytest-aio[curio,trio]

Usage

When installed the plugin runs all your async test functions/fixtures.

async def test_async():
    assert True

No need to mark your async tests. Just run pytest as is.

Async fixtures for sync tests

If you plan use async fixtures for sync tests, please ensure you have to include aiolib fixture:

# It's important to add aiolib fixture here
def test_with_async_fixtures(async_fixture, aiolib):
    assert async_fixture == 'value from async fixture'

As an alternative, If you are doing the async fixtures yourself, you can add aiolib inside them:

@pytest.fixture
async def async_fixture(aiolib):
    return 'value from async fixture'

# So for the test we don't need to implicity use `aiolib` anymore
def test_with_async_fixtures(async_fixture):
    assert async_fixture == 'value from async fixture'

Customize async libraries

By default each test function will be run with asyncio, trio, curio backends consistently (only if trio/curio are installed). But you can customise the libraries for all your tests creating the global fixture:

# Run all tests with Asyncio/Trio only
@pytest.fixture(params=['asyncio', 'trio'])
def aiolib(request):
    assert request.param

If you want to specify different options for the selected backend, you can do so by passing a tuple of (backend name, options dict):

@pytest.fixture(params=[
    pytest.param(('asyncio', {'use_uvloop': False}), id='asyncio'),
    pytest.param(('asyncio', {'use_uvloop': True}), id='asyncio+uvloop'),
    pytest.param(('trio', {'trio_asyncio': True}), id='trio+asyncio'),
    pytest.param(('curio', {'debug': True}), id='curio'),
])
def aiolib(request):
    assert request.param

To set a specific backends for a single test only:

@pytest.mark.parametrize('aiolib', ['asyncio'])
async def only_with_asyncio():
    await asyncio.sleep(1)
    assert True

Helpers

The plugin contains aiosleep fixture. It's an equivalent to asyncio.sleep, trio.sleep, curio.sleep for current running library.

Bug tracker

If you have any suggestions, bug reports or annoyances please report them to the issue tracker at https://github.com/klen/asgi-tools/issues

Contributing

Development of the project happens at: https://github.com/klen/pytest-aio

License

Licensed under a MIT license.

pytest-aio's Issues

uvloop support

Hello @klen !

Cool stuff, like always! :)

I've noticed that you didn't mentioned uvloop in the README, is it because it doesn't support?

I didn't test it yet, I was just wondering before hand.

Inconsistent "Runtime Error: Event loop is closed"

Hi, thanks for this library. I am hoping to migrate away from pytest-asyncio because of recent changes there.

I am using the aiolib fixture like this:

@pytest.fixture(scope="session")
def aiolib() -> tuple[str, dict[str, Any]]:
    return "asyncio", {"use_uvloop": False}

However, I get the following exception.

platform darwin -- Python 3.11.7, pytest-8.0.2, pluggy-1.4.0 --~/.pyenv/versions/3.11.7/envs/bikeshed/bin/python
cachedir: .pytest_cache
Using --randomly-seed=657316342
configfile: pyproject.toml
plugins: aio-1.5.0, unordered-0.5.2, anyio-4.3.0, env-1.1.3, randomly-3.15.0, cov-4.1.0, icdiff-0.9, ddtrace-2.7.0, xdist-3.5.0
collected 562 items      

# many tests

--- Logging error ---
Traceback (most recent call last):
  File "~/.pyenv/versions/3.11.7/lib/python3.11/logging/__init__.py", line 1113, in emit
    stream.write(msg + self.terminator)
ValueError: I/O operation on closed file.
Call stack:
  File "~/.pyenv/versions/3.11.7/lib/python3.11/asyncio/tasks.py", line 141, in __del__
    self._loop.call_exception_handler(context)
  File "~/.pyenv/versions/3.11.7/lib/python3.11/asyncio/base_events.py", line 1797, in call_exception_handler
    self.default_exception_handler(context)
  File "~/.pyenv/versions/3.11.7/lib/python3.11/asyncio/base_events.py", line 1771, in default_exception_handler
    logger.error('\n'.join(log_lines), exc_info=exc_info)
Message: "Task was destroyed but it is pending!\ntask: <AsyncioContextTask pending name='Task-pytest-aio-367' coro=<<async_generator_asend without __name__>()>>"
Arguments: ()
--- Logging error ---
Traceback (most recent call last):
  File "~/.pyenv/versions/3.11.7/lib/python3.11/logging/__init__.py", line 1113, in emit
    stream.write(msg + self.terminator)
ValueError: I/O operation on closed file.
Call stack:
  File "~/.pyenv/versions/3.11.7/lib/python3.11/asyncio/tasks.py", line 141, in __del__
    self._loop.call_exception_handler(context)
  File "~/.pyenv/versions/3.11.7/lib/python3.11/asyncio/base_events.py", line 1797, in call_exception_handler
    self.default_exception_handler(context)
  File "~/.pyenv/versions/3.11.7/lib/python3.11/asyncio/base_events.py", line 1771, in default_exception_handler
    logger.error('\n'.join(log_lines), exc_info=exc_info)
Message: "Task was destroyed but it is pending!\ntask: <AsyncioContextTask pending name='Task-pytest-aio-471' coro=<<async_generator_asend without __name__>()>>"
Arguments: ()
__________________________________________________ ERROR at teardown of test_redacted ___________________________________________________
[gw3] darwin -- Python 3.11.7 ~/.pyenv/versions/3.11.7/envs/bikeshed/bin/python

aiolib = ('asyncio', {'use_uvloop': False}), args = (), kwargs = {}, lib = 'asyncio', params = {'use_uvloop': False}
runner = <pytest_aio.runners.AsyncioRunner object at 0x29e91a950>, gen = <async_generator object _admin_test_client at 0x2ab2fd460>

    def wrapper(*args, aiolib, **kwargs):
        lib, params = aiolib
        if "aiolib" in argnames:
            kwargs["aiolib"] = aiolib
    
        with get_runner(lib, **params) as runner:
            if iscoroutinefunction(func):
                yield runner.run(func, *args, **kwargs)
                return
    
            gen = func(*args, **kwargs)
            try:
                yield runner.run(gen.__anext__().__await__)
            except StopAsyncIteration:
                raise RuntimeError("Async generator did not yield")
    
            try:
>               runner.run(gen.__anext__().__await__)

~/.pyenv/versions/3.11.7/envs/bikeshed/lib/python3.11/site-packages/pytest_aio/plugin.py:116: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
~/.pyenv/versions/3.11.7/envs/bikeshed/lib/python3.11/site-packages/pytest_aio/runners.py:69: in run
    task = AsyncioContextTask(fn(*args, **kwargs), self.ctx, self._loop)
~/.pyenv/versions/3.11.7/envs/bikeshed/lib/python3.11/site-packages/pytest_aio/utils.py:46: in __init__
    self._loop.call_soon(self._Task__step, context=self._context)
~/.pyenv/versions/3.11.7/lib/python3.11/asyncio/base_events.py:761: in call_soon
    self._check_closed()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <_UnixSelectorEventLoop running=False closed=True debug=False>

    def _check_closed(self):
        if self._closed:
>           raise RuntimeError('Event loop is closed')
E           RuntimeError: Event loop is closed

~/.pyenv/versions/3.11.7/lib/python3.11/asyncio/base_events.py:519: RuntimeError

It happens on a different tests every time (I am using pytest-randomly), and always at the end, usually 2-3 tests. It doesn't seem to be related to pytest-xdist runners because it happens when I run it without multiple workers.

Do you have any insights?