You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在单元测试setUpModule中启动asyncio后台任务?

在E2E测试中启动全局asyncio后台任务的实现方案

方案一:使用pytest + pytest-asyncio(推荐)

pytest对asyncio的支持更友好,通过模块级fixture可以轻松实现全局后台任务的复用,且无需线程/子进程,后台任务与测试用例共享同一事件循环。

步骤:

  1. 安装依赖:
pip install pytest pytest-asyncio
  1. 编写测试代码:
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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.22 22:45:11