How to fix "attached to a different loop"?

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?

Issue Analytics

State:
Created 7 years ago
Reactions:7
Comments:13 (5 by maintainers)

Top GitHub Comments

50reactions

Tinchecommented, Dec 2, 2016

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()

9reactions

Tinchecommented, Dec 2, 2016

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.

Top Results From Across the Web

RuntimeError: Task attached to a different loop - Stack Overflow

Hi I'm using AsyncIOMotorClient for asynchronous db calls to mongoDb. Below is my code. xyz.py async def insertMany(self,collection_name, ...

Developing with asyncio — Python 3.11.1 documentation

Asynchronous programming is different from classic “sequential” programming. This page lists common mistakes and traps and explains how to avoid them.

ASGI Event Loop Gotcha - Rob Blackbourn

3 min read. ·. Listen. Save. ASGI Event Loop Gotcha. I've shed many tears with the following exception: got Future <Future pending> attached...

asyncio.Semaphore RuntimeError: Task got Future attached to ...

Coding example for the question asyncio.Semaphore RuntimeError: Task got Future attached to a different loop.

18.5.1. Base Event Loop - Python 3.7.0a2 documentation

It provides multiple facilities, including: Registering, executing and cancelling delayed calls (timeouts). Creating client and server transports for various ...