RuntimeError: Cannot run the event loop while another loop is running using pytest.main

[PeterStolz]

To give you a minimal proof of concept in python3.10:
runner.py

import pytest                                                                                                   
import asyncio                                                                                                  
                 
# using nest_asyncio mitigates the problem                                                                                               
#import nest_asyncio                                                                                            
#nest_asyncio.apply()                                                                                                      
                                                                                                                
async def runner():                                                                                             
    pytest.main(['-k', 'test_asdf', 'test_file.py'])                                                            
                                                                                                                
loop = asyncio.get_event_loop()                                                                                 
loop.run_until_complete(runner())                                                                               
           
# does not raise an exception                                                                                                                                                                                                                     
# pytest.main(['-k', 'test_asdf', 'test_file.py'])     

test_file.py

import pytest                                                                                                   
                                                                                                                
@pytest.mark.asyncio                                                                                            
async def test_asdf():                                                                                          
    assert True     

/home/peter/Documents/pytestbug/test_runner.py:14: DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()
============================================= test session starts ==============================================
platform linux -- Python 3.10.4, pytest-7.1.2, pluggy-1.0.0
rootdir: /home/peter/Documents/pytestbug
plugins: asyncio-0.18.4.dev36+gc021932.d20220609
asyncio: mode=legacy
collected 1 item                                                                                               

test_file.py F                                                                                           [100%]

=================================================== FAILURES ===================================================
__________________________________________________ test_asdf ___________________________________________________

args = (), kwargs = {}, coro = <coroutine object test_asdf at 0x7f7c47cb3300>
old_loop = <_UnixSelectorEventLoop running=False closed=False debug=False>
task = <Task pending name='Task-2' coro=<test_asdf() running at /home/peter/Documents/pytestbug/test_file.py:5>>

    @functools.wraps(func)
    def inner(*args, **kwargs):
        coro = func(*args, **kwargs)
        if not inspect.isawaitable(coro):
            pyfuncitem.warn(
                pytest.PytestWarning(
                    f"The test {pyfuncitem} is marked with '@pytest.mark.asyncio' "
                    "but it is not an async function. "
                    "Please remove asyncio marker. "
                    "If the test is not marked explicitly, "
                    "check for global markers applied via 'pytestmark'."
                )
            )
            return
        nonlocal _loop
        _loop.stop()
        old_loop = _loop
        # old_loop
        _loop = asyncio.new_event_loop()
        asyncio.set_event_loop(_loop)
        task = asyncio.ensure_future(coro, loop=_loop)
        try:
>           _loop.run_until_complete(task)

../../.virtualenvs/pytestbug/lib/python3.10/site-packages/pytest_asyncio/plugin.py:460: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/usr/lib/python3.10/asyncio/base_events.py:622: in run_until_complete
    self._check_running()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

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

    def _check_running(self):
        if self.is_running():
            raise RuntimeError('This event loop is already running')
        if events._get_running_loop() is not None:
>           raise RuntimeError(
                'Cannot run the event loop while another loop is running')
E           RuntimeError: Cannot run the event loop while another loop is running

/usr/lib/python3.10/asyncio/base_events.py:584: RuntimeError
=============================================== warnings summary ===============================================
../../.virtualenvs/pytestbug/lib/python3.10/site-packages/pytest_asyncio/plugin.py:191
  /home/peter/.virtualenvs/pytestbug/lib/python3.10/site-packages/pytest_asyncio/plugin.py:191: DeprecationWarning: The 'asyncio_mode' default value will change to 'strict' in future, please explicitly use 'asyncio_mode=strict' or 'asyncio_mode=auto' in pytest configuration file.
    config.issue_config_time_warning(LEGACY_MODE, stacklevel=2)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================================== short test summary info ============================================
FAILED test_file.py::test_asdf - RuntimeError: Cannot run the event loop while another loop is running
========================================= 1 failed, 1 warning in 0.10s =========================================
/usr/lib/python3.10/asyncio/base_events.py:685: ResourceWarning: unclosed event loop <_UnixSelectorEventLoop running=False closed=False debug=False>
sys:1: RuntimeWarning: coroutine 'test_asdf' was never awaited
Task was destroyed but it is pending!
task: <Task pending name='Task-2' coro=<test_asdf() running at /home/peter/Documents/pytestbug/test_file.py:5>>

The cause for this exception is that my setup already has an existing running EventLoop and in that loop I invoke pytest.main.
pytest.asyncio uses wrap_in_sync in the async fixtures and tests, which calls

pytest-asyncio/pytest_asyncio/plugin.py

Line 454 in 860ff51

_loop.run_until_complete(task)

Therefore calling run_until_complete within a coroutine raises an exception.
Code executed inside an EventLoop can't stop the current EvenntLoop, create a new loop, run the code in that loop and restore the old loop. At least not without heavy modification. Python developers also stated that this behavior will not be changed.

This can be mitigated by using nest_asyncio, however there are different eventloop implementations that are not compatible with nest_asyncio. (My project used fastapi which uses a different loop implementation)

If there is an async entrypoiny like pytest.async_main into pytest that may be used as well. If we make all functions awaitable, we would not need to use run_until_complete, but could simply use await. However I don't think that this is feasible.

In the end I just ran pytest in a subprocess to call it outside of a coroutine.

As requested I further elaborated on my problem from #359 (comment)

[seifertm]

@PeterStolz thank you for the separate issue report. Sorry to cause you extra work, but it's much easier to keep on top of existing problems and to follow discussions.

The cause for this exception is that my setup already has an existing running EventLoop and in that loop I invoke pytest.main.
This was also my impression from your initial issue comment. That's why I asked about the specific use case for this. Thank you for elaborating on this.

Interestingly, I do not get an error when I run your example. Note that I did not uncomment any lines. The test finishes successfully:

$ pytest
===== test session starts =====
platform linux -- Python 3.10.5, pytest-7.1.2, pluggy-1.0.0
rootdir: /tmp/pa-test
plugins: asyncio-0.18.3
asyncio: mode=legacy
collected 1 item                                                                                                                    

test_file.py .                                                                                                                [100%]

===== warnings summary =====
venv/lib/python3.10/site-packages/pytest_asyncio/plugin.py:191
  /tmp/pa-test/venv/lib/python3.10/site-packages/pytest_asyncio/plugin.py:191: DeprecationWarning: The 'asyncio_mode' default value will change to 'strict' in future, please explicitly use 'asyncio_mode=strict' or 'asyncio_mode=auto' in pytest configuration file.
    config.issue_config_time_warning(LEGACY_MODE, stacklevel=2)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
===== 1 passed, 1 warning in 0.01s =====

I see your pytest-asyncio version is 0.18.4.dev36+gc021932.d20220609. Do you happen to run a patched/forked version of the plugin?

[PeterStolz]

I was using pytest-asyncio master to verify that the Issue still exists. I ran it again on the latest pypi version (asyncio-0.18.3 and python 3.10.4) and it still crashes. Did you really run the runner.py and not pytest directly?

[seifertm]

I was using pytest-asyncio master to verify that the Issue still exists. I ran it again on the latest pypi version (asyncio-0.18.3 and python 3.10.4) and it still crashes.

That is commendable, thank you :)

Did you really run the runner.py and not pytest directly?

No, I ran pytest. I can reproduce the crash with python runner.py.

Why does async def runner() need to be a coroutine? It does not require any await. Could it not be a regular synchronous function (e.g. def runner())?

[PeterStolz]

Why does async def runner() need to be a coroutine? It does not require any await. Could it not be a regular synchronous function (e.g. def runner())?

In this simple proof of concept there is no need to call pytest.main within a coroutine, however the code I am working with is highly asynchronous and if there is a coroutine further up in the call stack you have the same problem. Therefore I had to run it in a new subprocess.