Cannot use with pytest

[aherok]

Hello! I've used the package although I cannot use it my pytest scenarios due to the infamous got Future <Future pending> attached to a different loop issue.

Consider the following test code:

import pytest
from aiodataloader import DataLoader


async def load_data(ids):
    return map(lambda x: x + 5, ids)


loader = DataLoader(batch_load_fn=load_data)


@pytest.mark.asyncio
async def test_loader_returns_data(event_loop):
    data = await loader.load(1)
    assert data == 6

When run, I get an error:

RuntimeError: Task <Task pending coro=<test_user_loader_succeedes() running at /app/loaders/tests/test_loader.py:14> cb=[_run_until_complete_cb() at /usr/local/lib/python3.7/asyncio/base_events.py:158]> got Future <Future pending> attached to a different loop

Is there an easy way to fix it?

[romikforest]

The loader should be created after the asyncio loop is started. But I still have Task was destroyed but it is pending! with aiodataloader under the heavy load.

[septatrix]

You can simply create the loader as a fixture:

import pytest
from aiodataloader import DataLoader

async def load_data(ids):
    return map(lambda x: x + 5, ids)

@pytest.fixture
def loader():
    return DataLoader(batch_load_fn=load_data)

@pytest.mark.asyncio
async def test_loader_returns_data(event_loop, loader):
    data = await loader.load(1)
    assert data == 6

This also ensures that tests do not influence each other