为异步代码编写测试用例
提出问题
很方便使用 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()