How to fix "attached to a different loop"?

[Archelyst]

I have a very simple app called "myapp". It uses the AsyncElasticsearch client:

from elasticsearch_async import AsyncElasticsearch

def create_app():
    app = dict()
    app['es_client'] = AsyncElasticsearch('http://index:9200/')
    app['stuff'] = Stuff(app['es_client'])
    return app

class Stuff:
    def __init__(self, es_client):
        self.es_client = es_client

    def do_async_stuff(self):
        return self.es_client.index(index='test',
                                    doc_type='test',
                                    body={'field': 'sample content'})

My question is not about AsyncElasticsearch, it just happens to be an async thing I want to work with, could be sth else like a Mongo driver or whatever.

I want to test do_async_stuff() and wrote the following conftest.py

import pytest
from myapp import create_app

@pytest.fixture(scope='session')
def app():
    return create_app()

... and test_stuff.py

import pytest

@pytest.mark.asyncio
async def test_stuff(app):
    await app['stuff'].do_async_stuff()
    assert True

When I execute the test I get an exception with the message "attached to a different loop". Digging into that matter I found that pytest-asyncio creates a new event_loop for each test case (right?). The Elasticsearch client however, takes the default loop on instantiation and sticks with it. So I tried to convince it to use the pytest-asyncio event_loop like so:

import pytest

@pytest.mark.asyncio
async def test_stuff(app, event_loop):
    app['es_client'].transport.loop = event_loop
    await app['stuff'].do_async_stuff()
    assert True

This however gives me another exception:

__________________________________ test_stuff __________________________________

app = {'es_client': <Elasticsearch([{'host': 'index', 'port': 9200, 'scheme': 'http'}])>, 'stuff': <myapp.Stuff object at 0x7ffbbaff1860>}

    @pytest.mark.asyncio
    async def test_stuff(app):
>       await app['stuff'].do_async_stuff()

test/test_stuff.py:6: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <Task pending coro=<AsyncTransport.main_loop() running at /usr/local/lib/python3.5/dist-packages/elasticsearch_async/transport.py:133>>

    def __iter__(self):
        if not self.done():
            self._blocking = True
>           yield self  # This tells Task to wait for completion.
E           RuntimeError: Task <Task pending coro=<test_stuff() running at /srv/app/backend/test/test_stuff.py:6> cb=[_run_until_complete_cb() at /usr/lib/python3.5/asyncio/base_events.py:164]> got Future <Task pending coro=<AsyncTransport.main_loop() running at /usr/local/lib/python3.5/dist-packages/elasticsearch_async/transport.py:133>> attached to a different loop

How am I supposed to test this scenario?

[Tinche]

Hi,

the first thing you can try is this:

@pytest.fixture
def app(event_loop):
    return create_app()

i.e. make your app fixture depend on the event loop fixture. This should make your client instance get the loop actually used in the test (even if you don't actually use the argument, the event loop will get installed as the default loop for the duration of the test). This will also make your client fixture function-scoped, but it's a good starting point.

[Archelyst]

Hi Tinche, thanks for the incredibly fast response. You suggestions seems to work. However "This will also make your client fixture function-scoped" is something I'd rather like to avoid as it doesn't exactly increase the speed of the tests. What's the rationale for pytest-asyncio to create a new loop for each test? Not to share resources between tests? Maybe your suggestion is the only viable solution then.

Any idea, why my attempt to inject the loop into the client does not work? Cannot make any sense of the error message.

[Tinche]

The AsyncElasticsearch instance probably grabs the loop and uses it somewhere before you can change it in the test.

Yeah, ideally tests should run in total isolation. The reason why your client fixture must be function scoped is that the event loop fixture is function scoped. You can override the event loop fixture to be session scoped though. I'm not sure this is tested but it should be possible :)

Make a conftest.py file in your tests directory, and put this in it:

@pytest.yield_fixture(scope='session')
def event_loop(request):
    """Create an instance of the default event loop for each test case."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

[Archelyst]

Ok, thanks for the valuable input.

[juandiegopalomino]

Hi there,

So following from this issue, I'm trying to see if we could use this library to run async tests dependent on i/o in parallel. I've been trying the following example but couldn't get it to work asynchronously (total test takes 20 seconds instead of the desired 10)

import asyncio
import pytest

@pytest.yield_fixture(scope='session')
def event_loop(request):
    """Create an instance of the default event loop for each test case."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


# All test coroutines will be treated as marked.
pytestmark = pytest.mark.asyncio

async def test_example1(event_loop):
    """No marker!"""
    await asyncio.sleep(10, loop=event_loop)


async def test_example2(event_loop):
    """No marker!"""
    await asyncio.sleep(10, loop=event_loop)

If anyone could give me input, I'd greatly appreciate it!

[david-shiko]

The AsyncElasticsearch instance probably grabs the loop and uses it somewhere before you can change it in the test.

Yeah, ideally tests should run in total isolation. The reason why your client fixture must be function scoped is that the event loop fixture is function scoped. You can override the event loop fixture to be session scoped though. I'm not sure this is tested but it should be possible :)

Make a conftest.py file in your tests directory, and put this in it:

@pytest.yield_fixture(scope='session')
def event_loop(request):
    """Create an instance of the default event loop for each test case."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

Hello!
Can you please update an aswer to modern usage?
I came from this link

[Tinche]

@david-shiko What exactly do you mean?

[david-shiko]

1. As I know, get.event_loop() and using loops directly is deprecated (pytest warning about it)
2. I have a such code (I'm novice in tests) and can't figure out how to apply provided solution.

import pytest
from httpx import AsyncClient
import main
from tests import conftest


@pytest.mark.asyncio()
async def test_root():
    async with AsyncClient(app=main.app, base_url="http://test") as ac:
        response = conftest.event_loop(await ac.post("/register", json={
            "phone_number": "+7 931 964 0000",
            "full_name": "Чехов Антон Павлович"
        }))
    assert response.status_code == 201

I got an error:

RuntimeError("Task <Task pending name='Task-5' coro=<test_root() running at /home/david/PycharmProjects/work/tests/main_test.py:12> cb=[_run_until_complete_cb() at /usr/lib/python3.8/asyncio/base_events.py:184]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop")

[Tinche]

Why do you need the conftest.event_loop?

What happens if you try this:

import pytest
from httpx import AsyncClient
import main
from tests import conftest


@pytest.mark.asyncio()
async def test_root():
    async with AsyncClient(app=main.app, base_url="http://test") as ac:
        response = await ac.post("/register", json={
            "phone_number": "+7 931 964 0000",
            "full_name": "Чехов Антон Павлович"
        })
    assert response.status_code == 201