测试异步函数的方案

python异步函数需要在事件循环里跑才可以, 因此写单测会非常麻烦.
我在python源码里找到了一个简单的方案(https://github.com/python/cpython/blob/master/Lib/test/test_contextlib_async.py
), 很有意思, 推荐给大家:
_async_test装饰器, 该装饰器帮你把异步函数放到event loop里运行, 因此你只需要在单测里调用异步函数, 和同步用例一样写测试逻辑就行了. 不需要自己写eventloop.

"""测试异步函数的方案
来自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.
    这样就无须在测试用例里面手动触发事件循环了, 可以理解为自动把异步函数给触发运行"""

    @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的测试就更简单, 直接在异步函数上面加上装饰器即可:

@_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