python異步函數(shù)需要在事件循環(huán)里跑才可以, 因此寫(xiě)單測(cè)會(huì)非常麻煩.
我在python源碼里找到了一個(gè)簡(jiǎn)單的方案(https://github.com/python/cpython/blob/master/Lib/test/test_contextlib_async.py
), 很有意思, 推薦給大家:
_async_test裝飾器, 該裝飾器幫你把異步函數(shù)放到event loop里運(yùn)行, 因此你只需要在單測(cè)里調(diào)用異步函數(shù), 和同步用例一樣寫(xiě)測(cè)試邏輯就行了. 不需要自己寫(xiě)eventloop.

"""測(cè)試異步函數(shù)的方案
來(lái)自python源碼: https://github.com/python/cpython/blob/master/Lib/test/test_contextlib_async.py"""
import asyncio
import functools
import unittest
from contextlib import asynccontextmanager


def _async_test(func):
    """Decorator to turn an async function into a test case.
    這樣就無(wú)須在測(cè)試用例里面手動(dòng)觸發(fā)事件循環(huán)了, 可以理解為自動(dòng)把異步函數(shù)給觸發(fā)運(yùn)行"""

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        coro = func(*args, **kwargs)
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            return loop.run_until_complete(coro)
        finally:
            loop.close()
            asyncio.set_event_loop(None)

    return wrapper


class AsyncContextManagerTestCase(unittest.TestCase):

    @_async_test
    async def test_contextmanager_plain(self):
        state = []

        @asynccontextmanager
        async def woohoo():
            state.append(1)
            yield 42
            state.append(999)

        async with woohoo() as x:
            self.assertEqual(state, [1])
            self.assertEqual(x, 42)
            state.append(x)
        self.assertEqual(state, [1, 42, 999])

    @_async_test
    async def test_xx(self):
        assert 1 == 5

pytest的測(cè)試就更簡(jiǎn)單, 直接在異步函數(shù)上面加上裝飾器即可:

@_async_test
async def test_contextmanager_plain():
    state = []

    @asynccontextmanager
    async def woohoo():
        state.append(1)
        yield 42
        state.append(999)

    async with woohoo() as x:
        assert state == [1]
        assert x == 42
        state.append(x)
    assert state == [1, 42, 999]


@_async_test
async def test_xx():
    assert 1 == 5