深入AsyncioAsyncio与单元测试

Posted 2021-01-14 ikct2017

Testing with asyncio

之前有说过应用开发者不需要将loop当作参数在函数间传递，只需要调用asyncio.get_event_loop()即可获得。但是在写单元测试时，可能会需要用多个loop（每个测试用一个单独的loop），问题来了：是否为了支持单元测试而要将loop作为函数参数传入呢？

先看个例子。

import asyncio
from typing import Callable

async def f(notify: Callable[[str], None]):    # 1
    # < ... some code ... >
    loop = asyncio.get_event_loop()    # 2
    loop.call_soon(notify, ‘Alert!‘)    # 3
    # < ... some code ... >

1. 想象一个coroutine内部需要通过call_soon调用另一个函数，这个函数可能是logging，发聊天信息，短线股票操作或其它任何操作；

2. 仍然不通过函数参数来获取loop，但要记住一点，这个方法调用始终获取的是当前线程的loop；

3. 将回调函数及其参数添加到loop的下一次迭代中。

---

最佳方式是通过fixture来为异步代码提供loop，Pytest将fixtures中定义的函数返回值作为参数传入测试函数中，描述起来有些复杂，用代码展示一下。

# conftest.py    # 1
import pytest

@pytest.fixture(scope=‘function‘)   # 2
def loop():
    loop = asyncio.new_event_loop()    # 3
    try:
        yield loop
    finally:
        loop.close()    # 在结束时关闭loop

1. Pytest将会自动导入名称为“conftest.py”的文件并使其中的配置生效；

2. 这里创建了一个fixture，scope参数告诉Pytest这个fixture的作用范围，用function限制将会使得每个函数都获得新的loop；

3. 创建一个全新的loop，但不会立刻让其开始运行。

---

上述代码有个错误，不要直接使用它，错误很微妙，但也是本章的全部要点，下面开始讨论它，先给一个测试用例。

from somewhere import f    # 1

def test_f(loop):    # 2
    collection = []    # 3
    def f_notify(msg):    # 4
        collection.append(msg)

    loop.create_task(f(f_notify))   # 5
    loop.call_later(1, loop.stop)   # 6
    loop.run_forever()

    assert collection[0] == ‘Alert!‘    # 7

1. 这里当作伪代码，表示f是一个在其它模块中定义的coroutine；

2. Pytest会识别loop函数名并从fixtures中找到这个函数并传入它的调用返回值；

3. 用一个容器收集notify的信息；

4. 这是notify函数；

5. 安排一个coroutine调用notify作为task；

6. 因为loop是run_forever的，用call_later确保loop会停止；

7. 此处进行测试。

---

上面提到有个错误，在这个导入的coroutine函数f中，loop是通过get_event_loop()获得的，而非fixture中提供的，所以这个测试会失败，因为通过get_event_loop()获得的loop压根不会运行。

一个解决办法就是明确地给函数传入loop作为参数，这样就能保证正确的loop被使用，然而这样写代码十分痛苦，因为这样一来大量的函数都要传入loop参数。

有个更好的办法就是，当一个新的loop运行时，将这个loop设置为当前线程的loop，这样get_event_loop()返回的总是最新的loop，这对单元测试十分有用。

# conftest.py
import pytest

@pytest.fixture(scope=‘function‘)  
def loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)    # 这个方法执行后，所有后续的get_event_loop获得的都是fixture中的loop，不需要显式地将loop作为参数传入了
    try:
        yield loop
    finally:
        loop.close()