如何解决Python unittest子类化IsolatedAsyncioTestCase时Event loop is closed错误?
解决IsolatedAsyncioTestCase后续测试用例事件循环已关闭的问题
问题根源
出现这种情况,大多是因为测试代码里手动关闭了事件循环,或者复用了全局事件循环实例,导致第一个测试用例执行完毕后循环被关闭,后续测试无法再启动新的循环。
具体解决方法
禁止手动关闭事件循环
IsolatedAsyncioTestCase框架会自动负责事件循环的创建、运行和销毁,每个测试用例都会分配独立的循环实例。如果你的代码中存在loop.close()这类语句,直接删除即可。使用测试类专属的事件循环
不要通过asyncio.get_event_loop()获取全局循环,改用测试类提供的self.loop属性,它是当前测试用例的专属循环:# 错误示例 import asyncio loop = asyncio.get_event_loop() loop.run_until_complete(my_coroutine()) # 正确示例 await self.loop.run_until_complete(my_coroutine()) # 更简单的方式:直接await协程,框架会自动用当前循环执行 await my_coroutine()规范异步测试方法的定义
测试方法必须以test_开头,且定义为异步函数(async def),框架会自动绑定当前测试用例的事件循环来执行:import unittest from unittest import IsolatedAsyncioTestCase import asyncio class MyAsyncTests(IsolatedAsyncioTestCase): async def test_first_case(self): res = await asyncio.sleep(0.1, result="success") self.assertEqual(res, "success") async def test_second_case(self): res = await asyncio.sleep(0.1, result="success2") self.assertEqual(res, "success2")清理异步资源避免干扰
如果测试中创建了未完成的异步任务、连接等资源,可在tearDownAsync方法中统一清理:async def tearDownAsync(self): # 取消所有未完成的任务 pending_tasks = [t for t in asyncio.all_tasks(self.loop) if t != asyncio.current_task()] for task in pending_tasks: task.cancel() # 等待任务取消完成 await asyncio.gather(*pending_tasks, return_exceptions=True)
内容的提问来源于stack exchange,提问作者Atharva
相关产品推荐
相关产品推荐

