Pytest测试报错Event loop is closed，尝试pytest-asyncio仍未解决

运行测试用例test_post_train_restaurant_with_empty_products时，触发错误：qdrant_client.http.exceptions.ResponseHandlingException: Event loop is closed。测试代码包含自动生效的异步populate fixture，尝试给该fixture添加@pytest.mark.asyncio标记后问题仍未解决。

原测试代码 #

# 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

修改后的fixture（无效） #

@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'])

错误原因 #

1. 异步fixture需要配合异步测试用例或pytest-asyncio的事件循环管理，但你的测试用例是同步函数（用def定义），异步fixture的事件循环会在同步测试用例执行完毕后提前关闭，导致后续异步清理操作（deleteRestaurantById）报错。
2. @pytest.mark.asyncio是用来标记异步测试用例的，给fixture添加该标记无效，无法解决事件循环管理问题。

解决方案 #

有两种可行的修复方式：

方式一：改用AsyncClient编写异步测试用例

这是FastAPI异步接口测试的标准做法，确保事件循环统一管理：

# 替换TestClient为AsyncClient
from httpx import AsyncClient
import pytest

@pytest.fixture(scope='function')
async def async_client():
    async with AsyncClient(app=app, base_url=os.getenv('BASE_URL')) as client:
        yield client

@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'])

# 异步测试用例需添加@pytest.mark.asyncio标记
@pytest.mark.asyncio
async def test_post_train_restaurant_with_empty_products(async_client):
    response = await async_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

@pytest.mark.asyncio
async def test_post_train_restaurant_full(async_client):
    response = await async_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

方式二：在同步fixture中手动管理异步操作

如果不想修改测试用例为异步，可以用asyncio.run创建独立事件循环执行异步清理：

import asyncio
import pytest

@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
    # 用asyncio.run创建新事件循环执行异步清理
    asyncio.run(deleteRestaurantById(restaurant['_id']))

注意事项 #

• 确保已安装pytest-asyncio：pip install pytest-asyncio
• 尽量避免同步测试用例与异步fixture混用，否则容易出现事件循环管理混乱
• 优先选择AsyncClient方案，更符合FastAPI异步应用的测试逻辑

内容的提问来源于stack exchange，提问作者javi