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

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