Pytest - Event loop is closed

Question

I'm trying to run tests, but I got Event loop is closed. I have these test cases:

# other imports
from fastapi.testclient import TestClient

client = TestClient(app, base_url=os.getenv('BASE_URL'))

@pytest.fixture(scope='function', autouse=True)
async def populate():
  global restaurant, products
  restaurant = generateRestaurantData('Cafeteria', 
                                      'La cafeteria mas rica de la ciudad', 
                                      'San borja')
  products = [generateProductData('Cafe rico', 'Lo mas rico')]
  yield
  await deleteRestaurantById(restaurant['_id'])

# OTHER TESTS
def test_post_train_restaurant_with_empty_products():
  response = client.post('/train', auth=addBasicAuth(), json={
    'restaurant': generateRestaurantData('Cafeteria', 
                                         'La cafeteria mas rica de la ciudad', 
                                         'San borja'),
    'products': []
  })
  assert response.status_code == status.HTTP_200_OK

def test_post_train_restaurant_full():
  response = client.post('/train', auth=addBasicAuth(), json={
    'restaurant': generateRestaurantData('Cafeteria', 
                                         'La cafeteria mas rica de la ciudad', 
                                         'San borja'),
    'products': [generateProductData('Cafe rico', 'Lo mas rico')]
  })
  assert response.status_code == status.HTTP_200_OK

The test case test_post_train_restaurant_with_empty_products returns:

FAILED test/restaurants/test_post_trainRestaurant.py::test_post_train_restaurant_with_empty_products - qdrant_client.http.exceptions.ResponseHandlingException: Event loop is closed

Do you know what I'm doing wrong?

I tried using pytest-asyncio but isn't work for me

@pytest.fixture(scope='function', autouse=True)
@pytest.mark.asyncio
async def populate():
  global restaurant, products
  restaurant = generateRestaurantData('Cafeteria', 
                                      'La cafeteria mas rica de la ciudad', 
                                      'San borja')
  products = [generateProductData('Cafe rico', 'Lo mas rico')]
  yield
  await deleteRestaurantById(restaurant['_id'])

Answer 1

It's linked to the fixture definition with an async def which cannot be called outside an event loop.

@pytest.fixture(scope='function', autouse=True)
async def populate():
  global restaurant, products
  restaurant = generateRestaurantData('Cafeteria', 'La cafeteria mas rica de la ciudad', 'San borja')
  products = [generateProductData('Cafe rico', 'Lo mas rico')]
  yield
  await deleteRestaurantById(restaurant['_id'])

You could use pytest-asyncio allowing you to use async function in your test.

Maybe you could also run the your function like this:

@pytest.fixture(scope='function', autouse=True)
def populate():
  global restaurant, products
  restaurant = generateRestaurantData('Cafeteria', 'La cafeteria mas rica de la ciudad', 'San borja')
  products = [generateProductData('Cafe rico', 'Lo mas rico')]
  yield
  # call async function with asyncio
  asyncio.run(deleteRestaurantById(restaurant['_id']))

# or use async fixture:

@pytest_asyncio.fixture(scope='function', autouse=True)
async def populate():
  global restaurant, products
  restaurant = generateRestaurantData('Cafeteria', 'La cafeteria mas rica de la ciudad', 'San borja')
  products = [generateProductData('Cafe rico', 'Lo mas rico')]
  yield
  await deleteRestaurantById(restaurant['_id'])

PS: using global is not a good idea. You should yield the restaurant and the product.

---

edit:

I create a minimal reproductible example. The difficult point in your example is mixing synchronous and asynchronous function. As your function is synchronous, pytest is called synchronously and thus without any event loop (mentioned in your error message).

I suppose that deleteRestaurantById is a function called with the DELETE route (and thus need to be asynchronous).

Here are two ways for avoiding your matter. I installed the pytest-asyncio package. You either call the async function with the asyncio.run function or you mark your async fixture with the pytest_asyncio.fixture:

import asyncio

import pytest
import pytest_asyncio


async def async_action(a):
    await asyncio.sleep(0.1)
    print(f"I slept asynchronously with {a=}.")


@pytest.fixture()
def my_fixture():
    a ="a"
    b= "b"
    yield a, b
    asyncio.run(async_action(a))

@pytest_asyncio.fixture()
async def my_fixture_async():
    a ="async a"
    b= "async b"
    yield a, b
    await async_action(a)


def test_with_fixture(my_fixture):
    a, b = my_fixture
    print(f"{a=}, {b=}")


def test_with_async_fixture(my_fixture_async):
    a, b = my_fixture_async
    print(f"{a=}, {b=}")

You will have the following output:

============================= test session starts ==============================
collecting ... collected 2 items

77726761-pytest-async-fixture.py::test_with_fixture PASSED               [ 50%]
a='a', b='b'
I slept asynchronously with a='a'.

77726761-pytest-async-fixture.py::test_with_async_fixture PASSED         [100%]
a='async a', b='async b'
I slept asynchronously with a='async a'.