Pytest - 事件循环已关闭

Question

我正在尝试运行测试，但我发现事件循环已关闭 我有这个测试用例：

# 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

测试用例

test_post_train_restaurant_with_empty_products

返回

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

你知道我做错了什么吗？

Answer 1

它通过

async def

链接到夹具定义。

@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-asyncio 允许您在测试中使用

async

函数。

也许你也可以像这样运行你的函数：

@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(deleteRestaurantById(restaurant['_id']))

PS：使用全局并不是一个好主意。你应该交出餐厅和产品。