`Event loop is closed` in second test with aioredis connection

[crypto-only]

Requirements

pip install aioredis==2.0.1 There is redis server running at 127.0.0.1.

Code

test_get2() will raise Event loop is closed while test_get() will success. I guess event loop is closed after first test so saved _redis_conn can't connect to previous closed event loop. I don't want to update connection logic in Client to create connection each time to just pass the test because that will be slower and requires more code. I prefer to set pytest or pytest-asyncio to solve the problem.

import aioredis


class Client:

    def __init__(self):
        self._redis_conn = None

    async def _set_redis(self):
        if self._redis_conn is None:
            self._redis_conn = await aioredis.from_url('redis://127.0.0.1')
            
    async def get(self):
        await self._set_redis()
        return await self._redis_conn.get(b'key')
    

client = Client()


async def test_get():
    await client.get()


async def test_get2():
    await client.get()

[crypto-only]

Temporally solve this by adding function fixture, is there a way to just set pytest or pytest-asyncio to solve the problem?

import aioredis
import pytest


class Client:

    def __init__(self):
        self._redis_conn = None

    async def _set_redis(self):
        if self._redis_conn is None:
            self._redis_conn = await aioredis.from_url('redis://127.0.0.1')
            
    async def get(self):
        await self._set_redis()
        return await self._redis_conn.get(b'key')
    

@pytest.fixture(scope="function")
def client():
    return Client()


async def test_get(client):
    await client.get()


async def test_get2(client):
    await client.get()

[seifertm]

pytest-asyncio creates a separate event loop for each async def test. The loop is closed at the end of every test case. As you suggested, the Client._redis_conn receives a reference to the event loop of test_get during the test run. When test_get2 is run, Client._redis_conn still references the event loop of the previous test, which has been closed in the meantime.

If you have multiple test cases that need to run in the same event loop, you have the option to broaden the scope of the event loop fixutre [0]. The event loop fixture uses function scope by default. You can define a module-level fixture, for example, to run multiple tests with the same event loop:

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

[0] https://github.com/pytest-dev/pytest-asyncio#event_loop

[seifertm]

@crypto-only Have you had a chance to try extending the scope of the event loop fixture as I described in my previous post?

[seifertm]

Closing this due to inactivity of the OP.

@crypto-only if you have time to try the above fix, I'd very much appreciate your feedback. If the issue is not resolved for you, please reopen this issue.