如何在单元测试setUpModule中启动asyncio后台任务?
在E2E测试中启动全局asyncio后台任务的实现方案
方案一:使用pytest + pytest-asyncio(推荐)
pytest对asyncio的支持更友好,通过模块级fixture可以轻松实现全局后台任务的复用,且无需线程/子进程,后台任务与测试用例共享同一事件循环。
步骤:
- 安装依赖:
pip install pytest pytest-asyncio
- 编写测试代码:
import asyncio import pytest # 模拟连接被测服务器模拟器的后台任务逻辑 async def simulator_connection_task(): while True: # 这里编写维持连接的逻辑(如心跳、消息监听等) await asyncio.sleep(1) # 模块级fixture,仅初始化一次,供所有测试用例复用 @pytest.fixture(scope="module") async def simulator_connection(): # 创建并启动后台任务 task = asyncio.create_task(simulator_connection_task()) yield # 此处开始执行模块内的所有测试用例 # 测试结束后清理任务 task.cancel() try: await task except asyncio.CancelledError: pass # 测试用例示例 async def test_case1(simulator_connection): # 直接使用已建立的模拟器连接环境执行测试 assert True async def test_case2(simulator_connection): # 第二个测试用例复用同一后台任务 assert True
方案二:基于unittest改造(不推荐,仅作兼容参考)
unittest原生不支持异步模块级初始化,需手动管理事件循环和后台任务,实现步骤如下:
import asyncio import unittest # 模块全局变量,存储事件循环与后台任务 _global_loop = None _simulator_task = None async def simulator_connection_task(): while True: await asyncio.sleep(1) # 维持模拟器连接的逻辑 def setUpModule(): """模块级初始化,启动后台任务""" global _global_loop, _simulator_task _global_loop = asyncio.new_event_loop() asyncio.set_event_loop(_global_loop) # 在当前事件循环中启动后台任务 _simulator_task = _global_loop.create_task(simulator_connection_task()) def tearDownModule(): """模块级清理,关闭后台任务与事件循环""" global _global_loop, _simulator_task _simulator_task.cancel() _global_loop.run_until_complete(_simulator_task) _global_loop.close() # 自定义异步测试基类,封装异步逻辑执行方法 class AsyncTestBase(unittest.TestCase): def run_async(self, coro): """复用全局事件循环执行异步测试逻辑""" return _global_loop.run_until_complete(coro) # 测试用例类 class TestSimulator(AsyncTestBase): def test_case1(self): async def test_logic(): # 测试逻辑,直接使用全局连接 assert True self.run_async(test_logic()) def test_case2(self): async def test_logic(): assert True self.run_async(test_logic()) if __name__ == '__main__': unittest.main()
说明:
两种方案均满足需求:后台任务与测试用例运行在同一asyncio事件循环环境中,无需依赖线程或子进程。优先推荐pytest方案,代码更简洁、维护成本更低。
内容的提问来源于stack exchange,提问作者Sampo
相关产品推荐
相关产品推荐

