为异步代码编写测试用例

提出问题

很方便使用 Python 中的 unittest 为代码编写测试用例。对于异步代码,则有 aiounittest 进行处理,但这种 import 又不得不花时间证明它是可靠的。因此,是否存在某一种简单的方法来编写异步代码的测试用例呢?

解决问题

解决问题的思路大概是让这段异步代码得以执行结束,后根据其返回值进行判断。

import aiohttp
import asyncio


async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(url="https://api.ip.sb/jsonip") as response:
            r = await response.json()
    return r


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(fetch())

上述代码是获取外网 IP 的,采用了异步的写法(在这里只是为了举个例子)。

import aiohttp
import asyncio
import unittest
import ipaddress


async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(url="https://api.ip.sb/jsonip") as response:
            r = await response.json()
    return r


class AsyncTest(unittest.TestCase):
    @async_test
    async def test_fetch(self):
        r = await fetch()
        ip = r.get("ip")
        assert type(ipaddress.ip_address(ip)) == ipaddress.IPv4Address


if __name__ == "__main__":
    unittest.main()

通过增加一个异步的装饰器使异步代码执行完毕获取结果。当然,除了这种方式亦有其他的方式。

unittest.TestCase 中有一个方法叫做 setUp()。这个方法会在进行测试用例前进行调用,其默认为实现而不操作的空方法。因此,对其进行合适的修改也可以实现所需要的功能。

import aiohttp
import asyncio
import unittest
import ipaddress


async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(url="https://api.ip.sb/jsonip") as response:
            r = await response.json()
    return r


class AsyncTest(unittest.TestCase):
    def setUp(self):
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(None)

    def test_fetch(self):
        async def go():
            r = await fetch()
            ip = r.get("ip")
            assert type(ipaddress.ip_address(ip)) == ipaddress.IPv4Address

        self.loop.run_until_complete(go())


if __name__ == "__main__":
    unittest.main()

如果使用的 Python 足够新(2020年05月08日,最新 release 版本为 Python3.8.2),在 Python3.8 中 unittest 提供了异步的方法—— IsolatedAsyncioTestCase

This class provides an API similar to TestCase and also accepts coroutines as test functions.
这个类提供类似于 TestCase 的 API,并且也接受协程作为测试函数。

import aiohttp
import asyncio
import unittest
import ipaddress


async def fetch():
    async with aiohttp.ClientSession() as session:
        async with session.get(url="https://api.ip.sb/jsonip") as response:
            r = await response.json()
    return r


class AsyncTest(unittest.IsolatedAsyncioTestCase):
    async def test_fetch(self):
        r = await fetch()
        ip = r.get("ip")
        assert type(ipaddress.ip_address(ip)) == ipaddress.IPv4Address


if __name__ == "__main__":
    unittest.main()

本文链接:

https://jamchoi.cc/archives/testcase-for-async.html
1 + 7 =
快来做第一个评论的人吧~