異步編程 101：如何測試 async 代碼

視頻原文：Strategies for testing Async code - PyCon 2019

同時參考了：

• Testing Asyncio Python Code with Pytest

前面幾篇關於異步編程的文章：

• 異步編程 101: 是什麼、小試Python asyncio
• 異步編程 101：Python async await發展簡史
• 異步編程 101：寫一個事件循環
• 異步編程 101：asyncio中的 for 循環
• 異步編程 101：asyncio 進階上篇

異步編程，本質上是經過合做（cooperation）來達成併發效果，也即：須要 wait 的時候，也就是發生 IO 的時候，把控制權交給主事件循環。 (yield control when 'awaiting' asynchronous results.) 這個過程有點像事件循環完成了操做系統的工做，能夠將事件循環看做是操做系統，而後把協程看做是線程這麼來理解。整個事件循環是在一個線程裏面的，意味着任務切換更加高效，無需上下文轉換。

異步代碼很高效，可是也有很蛋疼的地方，那就是測試。

0x01 : async 測試實例

來經過一個簡單的例子看一下吧：一個Cat類，有一個 move 方法，這個方法是異步的。

而後用 unittest 寫一個測試類，你能發現下面代碼的問題嗎？

herd(grafield, 'forward') 返回的是一個協程對象（coroutine object），若是你不await他，什麼也不會發生。而coroutine object是 truthy 的，因此assertTrue() 是可以經過的。若是你運行一下 test，會看到coroutine herd was nerver awaited的 warning。

下圖這樣調用await仍是不對的，由於await關鍵字只能出如今async函數裏面。

一個解決方案是加入事件循環：

這能work，可是估計你也看出來了，這很麻煩。若是我有多個方法，難道我須要每一個 test方法都加一個事件循環嗎？更重要的是，我只是想作一下單元測試，事件循環在這個時候其實是一個底層細節，我不須要關心。

在 Python3.7 中，asyncio新增了一個方法：asyncio.run()，爲你隱藏了事件循環的細節，因此可以讓代碼更加簡潔：

0x02 pytest-asyncio

安裝：pip install pytest-asyncio，這其實是pytest的一個插件。

用法很簡單，重要的是咱們得知道工做原理。以前的代碼問題在於，pytest 默認的的 runner 會將全部的函數看成普通函數處理，而對於 async 函數， 調用的時候返回的是一個 coroutine object。因此咱們得想辦法告訴 pytest 使用一個 eventloop 來運行測試方法。

一種方法是，實例化一個eventloop 而後注入到 tests裏面，好比：

import asyncio
import pytest

async def say(what, when):
    await asyncio.sleep(when)
    return what

@pytest.fixture
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()


def test_say(event_loop):
    expected = 'This should fail!'
    assert expected == event_loop.run_until_complete(say('Hello!', 0))
複製代碼

這種方法不方便之處在於每次都須要手動注入 eventloop，更優雅的方法是調整 test runner，讓它識別 async 函數，看成 asyncio tasks 來執行。

pytest-asyncio完成的功能就是這樣的，它的 API 很是簡單，你只須要爲 async function 添加一個 @pytest.mark.asyncio 修飾器便可：

import pytest
from say import say

@pytest.mark.asyncio
async def test_say():
    assert 'Hello!' == await say('Hello!', 0)
複製代碼