如何在异步unittest中运行classmethod以共享服务器会话

问题场景 #

在使用unittest.IsolatedAsyncioTestCase执行集成测试时，需要让所有测试用例共用同一个服务器会话（频繁创建新会话会被服务端拦截），但常规实现方式存在障碍：

• 用@classmethod修饰的setUpClass不支持异步逻辑，直接写async方法无法生效：

class IntegrationTest(unittest.IsolatedAsyncioTestCase):
    @classmethod
    async def setUpClass(self):  # 无法正常执行
        self.session = await get_new_aio_session()

• 在asyncSetUp中通过hasattr判断会话是否存在也没用，因为每个测试用例对应独立的类实例，self.session无法跨实例共享：

async def asyncSetUp(self):
    if hasattr(self, 'session'): # 无法实现会话复用
        return
    self.session = await get_new_aio_session()

我们需要的是：会话仅初始化一次，所有测试用例直接复用该对象，效果类似：

class IntegrationTest(unittest.IsolatedAsyncioTestCase):
    async def fixture(self):
        # 该方法仅执行一次
        self.session = await get_new_aio_session()

    async def test_a(self):
        case = foo(session=self.session)
        self.assertEqual("foo", case.status)        

    async def test_b(self):
        case = bar(session=self.session)
        self.assertEqual("bar", case.status)        

可行解决方案 #

方案1：同步类方法+asyncio.run初始化类级会话 #

利用类变量存储会话，在同步的setUpClass中用asyncio.run执行异步的会话创建逻辑，所有实例都能访问这个类变量：

import unittest
import asyncio
from your_module import get_new_aio_session, foo, bar

class IntegrationTest(unittest.IsolatedAsyncioTestCase):
    # 类变量，所有实例共享
    session = None

    @classmethod
    def setUpClass(cls):
        # 在同步方法内执行异步初始化
        cls.session = asyncio.run(get_new_aio_session())

    @classmethod
    def tearDownClass(cls):
        # 同步方法内执行异步清理
        asyncio.run(cls.session.close())

    async def test_a(self):
        case = foo(session=self.session)
        self.assertEqual("foo", case.status)        

    async def test_b(self):
        case = bar(session=self.session)
        self.assertEqual("bar", case.status)        

方案2：重写类级异步初始化逻辑（Python 3.10+） #

针对Python 3.10及以上版本，可以借助IsolatedAsyncioTestCase的内部事件循环，手动执行异步的类级初始化：

import unittest
from your_module import get_new_aio_session, foo, bar

class IntegrationTest(unittest.IsolatedAsyncioTestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        # 用测试类的事件循环执行异步初始化
        cls._asyncioTestLoop.run_until_complete(cls._async_setUpClass())

    @classmethod
    async def _async_setUpClass(cls):
        cls.session = await get_new_aio_session()

    @classmethod
    def tearDownClass(cls):
        # 执行异步清理逻辑
        cls._asyncioTestLoop.run_until_complete(cls._async_tearDownClass())
        super().tearDownClass()

    @classmethod
    async def _async_tearDownClass(cls):
        await cls.session.close()

    async def test_a(self):
        case = foo(session=self.session)
        self.assertEqual("foo", case.status)        

    async def test_b(self):
        case = bar(session=self.session)
        self.assertEqual("bar", case.status)        

方案3：模块级共享会话 #

如果测试类数量不多，可以在模块初始化时创建会话，让所有测试类直接复用：

import unittest
import asyncio
from your_module import get_new_aio_session, foo, bar
import atexit

# 模块级会话，所有测试类共享
SESSION = asyncio.run(get_new_aio_session())

class IntegrationTest(unittest.IsolatedAsyncioTestCase):
    async def test_a(self):
        case = foo(session=SESSION)
        self.assertEqual("foo", case.status)        

    async def test_b(self):
        case = bar(session=SESSION)
        self.assertEqual("bar", case.status)        

# 模块退出时自动关闭会话
def cleanup_session():
    asyncio.run(SESSION.close())

atexit.register(cleanup_session)

内容的提问来源于stack exchange，提问作者Anders_K