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()    

深入Asyncio（十二）Asyncio與單元測試