我正在使用pytest-asyncio来测试我的基于asyncio的库。
我使用的是类级别的方法，在这种方法中，一个TestClass有几个TestMethods，这些TestMethods被测试框架一个接一个地调用。
setup方法是对ClassUnderTest方法的扩展。teardown方法目前什么也不做。然而，我在teardown中注解了预期的功能。
我想做的是，实现一个asyncteardown和/或setup，这样我就可以await的一些dec清理代码。这可能吗？
我在pytest-asyncio文档中没有找到关于这方面的内容，它非常简短。因此，我问这个问题。也许有人遇到了类似的问题，并找到了解决问题的方法。

import asyncio
import random

import pytest

class ClassUnderTest:
    def __init__(self):
        self._queue = asyncio.Queue()
        self._task1 = None
        self._task2 = None

    async def start(self):
        self._task1 = asyncio.create_task(self.producer())
        self._task2 = asyncio.create_task(self.consumer())

    async def stop(self):
        self._task1.cancel()
        self._task2.cancel()
        return await asyncio.gather(self._task1, self._task2, return_exceptions = True)

    @property
    def tasks(self):
        return self._task1, self._task2

    async def producer(self):
        try:
            while True:
                if self._queue.qsize() < 10:
                    self._queue.put_nowait(random.randint(0, 10))

                await asyncio.sleep(50)

        except asyncio.CancelledError:
            print("Finito!")
            raise

    async def consumer(self):
        try:
            while True:
                if self._queue.qsize() > 0:
                    elem = self._queue.get_nowait()
                    print(elem)

                await asyncio.sleep(100)

        except asyncio.CancelledError:
            print("Finito!")
            raise

@pytest.mark.asyncio
class TestClass:
    """ Tests my asynio code """

    def setup_method(self):
        self._my_class_under_test = ClassUnderTest()

    def teardown_method(self):
        """
        if not tasks[0].cancelled() or not tasks[1].cancelled():
            await self._my_class_under_test.stop()
        """

    async def test_start(self):
        await self._my_class_under_test.start()
        tasks = self._my_class_under_test.tasks
        assert not tasks[0].cancelled()
        assert not tasks[1].cancelled()
        await self._my_class_under_test.stop()

    async def test_stop(self):
        await self._my_class_under_test.start()
        tasks = self._my_class_under_test.tasks
        return_values = await self._my_class_under_test.stop()
        assert tasks[0].cancelled()
        assert tasks[1].cancelled()
        assert isinstance(return_values[0], asyncio.CancelledError)
        assert isinstance(return_values[1], asyncio.CancelledError)

    async def test_producer(self):
        pass

    async def test_consumer(self):
        pass

if __name__ == "__main__":
    pytest.main([__file__])

输出：

/home/user/.config/JetBrains/PyCharm2023.2/scratches/asyncio_test_setup_teardown.py 
============================= test session starts ==============================
platform linux -- Python 3.10.13, pytest-7.4.2, pluggy-1.3.0
rootdir: /home/user/.config/JetBrains/PyCharm2023.2/scratches
plugins: timeout-2.1.0, asyncio-0.21.1
asyncio: mode=strict
collected 2 items

asyncio_test_setup_teardown.py ..                                        [100%]

============================== 2 passed in 0.01s ===============================

Process finished with exit code 0