异步协程的单元测试

Question

我想使用nosetests来测试类中的协程。我的第一个想法是用asyncio.run()调用每个协程。不幸的是，在第一次测试之后，我开始收到运行时异常。这是一个显示问题的最小示例：

import asyncio

class MyClass:

    def __init__(self):
        self.event = asyncio.Event()

    async def hello(self):
        print("Hello from MyClass")

class TestMyClass:

    def setup(self):
        self.my_class = MyClass()

    def test1(self):
        asyncio.run(self.my_class.hello())

    def test2(self):
        asyncio.run(self.my_class.hello())

当使用nosetests运行此脚本时，将为test2引发一个RuntimeError: There is no current event loop in thread 'MainThread'.。令人不快的一行是在MyClass中创建Event。我不完全理解这个问题。documentation for asyncio.run()声明该函数“应该用作异步程序的主要入口点，并且理想情况下应该只调用一次。”对我来说，这听起来似乎可以在一个程序中多次调用asyncio.run()，尽管不推荐这样做。

无论如何，当错误指向一个不存在的事件循环时，我决定手动管理事件循环，并提出了以下解决方法：

import asyncio

class MyClass:

    def __init__(self):
        self.event = asyncio.Event()

    async def hello(self):
        print("Hello from MyClass")

class TestMyClass:

    def __init__(self):
        try:
            self.loop = asyncio.get_event_loop()
        except RuntimeError:
            self.loop = asyncio.new_event_loop()

    def setup(self):
        asyncio.set_event_loop(self.loop)
        self.my_class = MyClass()

    def test1(self):
        asyncio.run(self.my_class.hello())

    def test2(self):
        asyncio.run(self.my_class.hello())

当多个测试脚本顺序运行时，初始化中的try...except是必需的。我使用的是Python 3.7.6。

我的解决方案在我看来不是很干净，我想知道是否有更好的方法。

Answer 1

由于我使用的是asyncio低级API，一个潜在的更干净的解决方法可能是：

import asyncio

class MyClass:

    def __init__(self):
        self.event = asyncio.Event()

    async def hello(self):
        print("Hello from MyClass")

class TestMyClass:

    def __init__(self):
        self.loop = asyncio.get_event_loop()

    def setup(self):
        self.my_class = MyClass()

    def test1(self):
        self.loop.run_until_complete(self.my_class.hello())

    def test2(self):
        self.loop.run_until_complete(self.my_class.hello())

这种方法是否存在任何可能的问题？