Teardown error: "attached to a different loop" with even_loop module scope fixture

[alblasco]

I'm working on python/pytest/asyncio since beginning of this year, and when updating pytest-asyncio (from 0.10.0 to 0.12.0) then some previous working tests started to fail.

I have created following minimal example (isolated from a more complex solution) that always fails at teardown.
Can you please let me know if you can reproduce it, and your kind feedback about this potential issue.
Don´t hesitate to ask for any additional information or test on my side.

BR

Angel Luis

Additional info

Versions used to reproduce are:

• pytest-asyncio 0.12.0. (it fails also with 0.11.0)
• pytest 5.4.2
• python 3.8.2
• setuptools 46.2.0 & pip 20.1
• windows 10 Pro [Versión 10.0.18362.175]
  Note: Minimal Example works with pytest-asyncio 0.10.0.

Minimal example is the following

import asyncio
import contextlib
import functools
import pytest


@pytest.mark.asyncio
async def test_simple(port):
    assert True

@pytest.fixture(scope="module")
def event_loop():
    """Change event_loop fixture to module level."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="module")
async def port(request, event_loop):
    def port_finalizer(finalizer):
        async def port_afinalizer():
            await finalizer(None, None, None)
        event_loop.run_until_complete(port_afinalizer())

    context_manager = port_map()
    port = await context_manager.__aenter__()
    request.addfinalizer(functools.partial(port_finalizer, context_manager.__aexit__))
    return port

@contextlib.asynccontextmanager
async def port_map():
    worker = asyncio.create_task(background_worker())
    yield
    try:
        worker.cancel()
        await worker
    except asyncio.CancelledError:
        pass

async def background_worker():
    while True:
        await asyncio.sleep(10.0)

When I run the test I get the following console output:

(venv) C:\git\ATB2\ejemplo_issue>pytest
=================== test session starts =============================
platform win32 -- Python 3.8.2, pytest-5.4.2, py-1.8.1, pluggy-0.13.1
rootdir: C:\git\ATB2\ejemplo_issue
plugins: asyncio-0.12.0
collected 1 item                                                                                                                                                                                                                                                           

test_example.py .E                                                                                                                                                                                                                                                   [100%]

====================== ERRORS =======================================
______________ ERROR at teardown of test_simple _____________________

finalizer = <bound method _AsyncGeneratorContextManager.__aexit__ of <contextlib._AsyncGeneratorContextManager object at 0x0423AC40>>

    def port_finalizer(finalizer):
        async def port_afinalizer():
            await finalizer(None, None, None)
>       event_loop.run_until_complete(port_afinalizer())

test_example.py:24:
 _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
C:\Users\ablasco\AppData\Local\Programs\Python\Python38-32\lib\asyncio\base_events.py:616: in run_until_complete
    return future.result()
test_example.py:23: in port_afinalizer
    await finalizer(None, None, None)
C:\Users\ablasco\AppData\Local\Programs\Python\Python38-32\lib\contextlib.py:178: in __aexit__
    await self.gen.__anext__()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

    @contextlib.asynccontextmanager
    async def port_map():
        worker = asyncio.create_task(background_worker())
        yield
        try:
            worker.cancel()
>           await worker
E           RuntimeError: Task <Task pending name='Task-4' coro=<port.<locals>.port_finalizer.<locals>.port_afinalizer() running at C:\git\ATB2\ejemplo_issue\test_example.py:23> cb=[_run_until_complete_cb() at C:\Users\ablasco\AppData\Local\Programs\Python\Python38-32
\lib\asyncio\base_events.py:184]> got Future <Task pending name='Task-2' coro=<background_worker() running at C:\git\ATB2\ejemplo_issue\test_example.py:43> wait_for=<Future cancelled>> attached to a different loop

test_example.py:37: RuntimeError
============================= short test summary info ==============================
ERROR test_example.py::test_simple - RuntimeError: Task <Task pending name='Task-4' coro=<port.<locals>.port_finalizer.<locals>.port_afinalizer() running at C:\git\ATB2\ejemplo_issue\test_example.py:23> cb=[_run_until_complete_cb() at C:\Users\ablasco\AppData\Local...

=========================== 1 passed, 1 error in 0.13s =============================

Unexpectedly either of the following changes, made the test to pass:

a) Change fixture to a lower level scope (class or function):

@pytest.fixture(scope="class")
def event_loop():

@pytest.fixture(scope="class")
async def port(request, event_loop):

b) Embed the test inside a class:

class TestClass:
    @pytest.mark.asyncio
    async def test_simple(port):
        assert True

c) Or change finalizer to call get_event_loop (instead of using event_loop from fixture)

    def port_finalizer(finalizer):
        async def port_afinalizer():
            await finalizer(None, None, None)
        # event_loop.run_until_complete(port_afinalizer())
        loop = asyncio.get_event_loop()
        loop.run_until_complete(port_afinalizer())

[yanlend]

I have a similar problem that async fixtures are somehow attached to a different loop than the pytest event_loop since 0.11.0
For me, it was the case on windows and ubuntu.

[alblasco]

After some debugging, side_effect of get_event_loop is causing a change of loop. It seems safer the previous method used in 0.10.0 that recovers the loop created in event_loop fixture. See details below

• When event_loop fixture is scoped as module, then request.keywords doesn´t contains "asyncio" (as module is previous to any specific test). So next lines 0.6.0 version causing test failures with: There is no current event loop in thread 'MainThread' #54-66 are not executed, and so there is no event loop set yet

    if fixturedef.argname == "event_loop" and 'asyncio' in request.keywords:
        outcome = yield
        loop = outcome.get_result()
        policy = asyncio.get_event_loop_policy()
        try:
            old_loop = policy.get_event_loop()
        except RuntimeError as exc:
            if 'no current event loop' not in str(exc):
                raise
            old_loop = None
        policy.set_event_loop(loop)
        fixturedef.addfinalizer(lambda: policy.set_event_loop(old_loop))
        return

• And when following async fixture (port in the previous minimal example) is executed in lines Remove transfer_markers #105-115:

    elif inspect.iscoroutinefunction(fixturedef.func):
        coro = fixturedef.func

        def wrapper(*args, **kwargs):
            async def setup():
                res = await coro(*args, **kwargs)
                return res

            return asyncio.get_event_loop().run_until_complete(setup())

        fixturedef.func = wrapper

then there is a problem with an implicit side effect of asyncio.get_event_loop. See extracted below:

def get_event_loop():
    """Return an asyncio event loop.

    When called from a coroutine or a callback (e.g. scheduled with call_soon
    or similar API), this function will always return the running event loop.

    If there is no running event loop set, the function will return
    the result of `get_event_loop_policy().get_event_loop()` call.
    """
    # NOTE: this function is implemented in C (see _asynciomodule.c)
    current_loop = _get_running_loop()
    if current_loop is not None:
        return current_loop
    return get_event_loop_policy().get_event_loop()
	
class BaseDefaultEventLoopPolicy(AbstractEventLoopPolicy):	
    def get_event_loop(self):
        """Get the event loop for the current context.

        Returns an instance of EventLoop or raises an exception.
        """
        if (self._local._loop is None and
                not self._local._set_called and
                isinstance(threading.current_thread(), threading._MainThread)):
            self.set_event_loop(self.new_event_loop())       <<<<< SIDE EFFECT

        if self._local._loop is None:
            raise RuntimeError('There is no current event loop in thread %r.'
                               % threading.current_thread().name)

        return self._local._loop

As _get_running_loop() is still None, then self.set_event_loop(self.new_event_loop()), is creating and setting a 2º event_loop (different from the one created at the event_loop fixture)

Hope this can help. If anything else is needed please let me know.

[Tinche]

Thanks for the analysis, I guess I need to dive deeper into this to really understand it. Can you think of a way of solving this?

[alblasco]

I can try to do my best to find a solution and prepare a pull request, if you agree on it.

Just one question I don´t understand block under if on line 54:

if fixturedef.argname == "event_loop" and 'asyncio' in request.keywords:
        outcome = yield
        loop = outcome.get_result()
        policy = asyncio.get_event_loop_policy()
        try:
            old_loop = policy.get_event_loop()
        except RuntimeError as exc:
            if 'no current event loop' not in str(exc):
                raise
            old_loop = None
        policy.set_event_loop(loop)
        fixturedef.addfinalizer(lambda: policy.set_event_loop(old_loop))
        return

This block is under the condition of being asyncio in request.keyworks:

• If event_loop is defined with scope = module this will not be true (as it is right, as in a module it has not marks of every test yet)
• if event_loop is defined with scope = function this is true, and it is executed. So event_loop is set and a finalizer is registered to come back to previous value.

Several options:

• To execute it for event_loop fixture, on all scopes: so second condition ('asyncio' in request.keywords:) should be removed
• Or this block should never be executed (on any scope),
• ...

I would appreciate if anyone can clarify on the right expected behaviour

Thanks in advance.

[alblasco]

Hi
Just proposed a solution, that fixes this issue, and pass all available tests on windows & python 3.7 & 3.8 (sorry not tested on previous python versions).
Note: A new test case is included.
Please let me know your kind feedback.

Don´t hesitate to contact me for any additional support.
BR

[seifertm]

I can no longer reproduce the error of the OP as of pytest-asyncio 0.21.1. Hence, I'll mark this issue as resolved.

[poppyi-domain]

I'm getting the same error message which occurs on teardown ("..attached to a different loop..") for version 0.23.2. When downgrading to version 0.21.1 everything is fine.

[seifertm]

@poppyi-domain Pytest-asyncio v0.23 supports running tests in different event loops. There's a loop for each level of the test suite (session, package, module, class, function). However, pytest-asyncio wrongly assumes that the scope of an async fixture is the same as the scope of the event loop in which the fixture runs. This essentially makes it impossible to have a session-scoped fixture that runs in the same loop as a function-scoped test. See #706.

If v0.23 causes trouble for you, I suggest sticking to pytest-asyncio v0.21, until the issue is resolved.