使用Pytest异步测试：测试前后置处理异常问题排查

你的异步fixture中yield后的清理逻辑提前执行，主要是因为pytest对异步代码的处理需要特定配置，以及测试用例的异步写法规范问题，具体原因和解决步骤如下：

核心原因 #

• pytest本身不原生支持异步fixture和异步测试用例，必须依赖pytest-asyncio插件才能正确处理异步代码的等待逻辑。如果缺少该插件或配置不当，pytest会直接跳过异步操作的等待，导致测试还未完成就执行yield后的清理步骤。
• 如果你的测试用例是用普通def定义的同步函数，而非async def的异步函数，pytest不会等待测试内部的异步操作完成，就会触发fixture的清理逻辑。

解决步骤 #

1. 安装并配置pytest-asyncio
   先确保安装了该插件：

   pip install pytest-asyncio
   

   可以在项目根目录的pytest.ini中添加配置，确保pytest默认使用asyncio运行器：

   [pytest]
   asyncio_mode = auto
   

2. 确保测试用例是异步函数
   所有依赖该异步fixture的测试用例都必须用async def定义，并且内部的异步操作要加上await：

   async def test_catalog_query(elastic_api, catalog):
       # 测试逻辑，所有异步调用都要加await
       result = await catalog.get_document(index='catalog_test', doc_id='1')
       assert result is not None
   

3. 修复fixture中的文件操作
   你的代码中直接用open打开文件但未关闭，可能导致资源泄漏，建议改用with语句自动管理文件：

   @pytest.fixture(autouse=True)
   async def run_around_tests(elastic_api, catalog):
       with open('test/resources/es_index_mapping.json') as es_index_mapping:
           es_index_mapping_dict = json.load(es_index_mapping)
       with open('test/resources/es_cars.json') as es_cars:
           es_cars_dict = json.load(es_cars)
       
       await elastic_api.create_index(index='catalog_test', payload=es_index_mapping_dict)
       await catalog.upsert_documents(index='catalog_test', payload=es_cars_dict)
       yield
       await elastic_api.delete_index(index='catalog_test')
       await catalog.close()
   

4. 检查fixture作用域（可选）
   如果你希望该fixture在所有测试执行前只初始化一次，所有测试结束后再执行清理，可以设置scope='session'，但要注意依赖的elastic_api和catalog fixture的作用域也需要匹配为session，否则会触发pytest的作用域冲突错误：

   @pytest.fixture(autouse=True, scope='session')
   async def run_around_tests(elastic_api, catalog):
       # 初始化逻辑...
       yield
       # 清理逻辑...
   

内容的提问来源于stack exchange，提问作者Guilherme Maia E Silva