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的下一次迭代中。

# 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. 此处进行测试。

一个解决办法就是明确地给函数传入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()    

原文地址：https://www.cnblogs.com/ikct2017/p/9829055.html