(2023.04.02 Sun@HRB)
在Python的unittest中测试async对象，被mock的对象需要保持async对象的定义和调用方式。
unittest包中包含一个IsolatedAsyncioTestCase，该test case始于Python3.8版本。该test case和TestCase类似，差别在于前者用于对test case写入协程，并可调用其他协程和使用await关键字。在测试类中可混合使用同步和异步测试。
下面案例
import asyncio
import unittest
async def sleep():
await asyncio.sleep(0.5)
return True
class testClass(unittest.IsolatedAsyncioTestCase):
async def test_async(self):
res = await sleep()
self.assertTrue(res)
用pytest test_asyncio.py运行
(base) C:\>pytest test_asyncio.py
================================== test session starts ==================================
platform win32 -- Python 3.8.5, pytest-6.1.1, py-1.9.0, pluggy-0.13.1
rootdir: C:\
collected 1 item
test_asyncio.py . [100%]
异步test case支持TestCase中的setUp和tearDown方法，其名为asyncSetUp和asyncTearDown，接受协程。不过IsolatedAsyncioTestCase也支持setUpClass和tearDownClass类。
与MagicMock类似的AsyncMock类
(2023.04.03 Mon@HRB)
unittest.mock中包含一个异步版本的MagicMock，即AsyncMock，用于创建异步的mock。和MagicMock相似，AsyncMock也有side_effect和return_value两个属性，并且属性行为相似，await AsyncMock将返回return_value或side_effect，但注意调用时需要采用异步方式。如下面的案例，在调用时，my_mock()调用返回的是一个协程，必须使用await关键字才能返回结果。
import unittest
from unittest.mock import AsyncMock
class testCase(unittest.IsolatedAsyncioTestCase):
async def test_mocking_demo(self):
amock = AsyncMock()
amock.return_value = 298
r = amock()
print(type(r))
await_result = await r
print(await_result)
self.assertEqual(298, await amock())
Mock生成器函数和context manager
mock生成器函数和contxt manager仅需要修改该函数中的特定magic mthod，分别是__aiter和__aenter__。示例如下
import unittest
from unittest.mock import AsyncMock, Mock
class TestMockingDemo(unittest.IsolatedAsyncioTestCase):
async def test_mock_generator(self):
expected_values = ["foo", "bar", "baz"]
my_mock_generator = AsyncMock()
my_mock_generator.__aiter__.return_value = expected_values
actual_values = []
async for value in my_mock_generator:
actual_values.append(value)
self.assertListEqual(expected_values, actual_values)
async def test_mock_context_manager(self):
mock_cm = AsyncMock()
# Note that by default an AsyncMock returns more AsyncMocks - we have to replace it with a Mock if we want a
# synchronous function
mock_cm.get_state = Mock(return_value="Not entered")
# Get a context object as a result of entering the context manager. Alternatively, __aenter__ could return
# mock_cm, to emulate the behaviour of returning self upon entering the context manager
mock_ctx = AsyncMock()
mock_cm.__aenter__.return_value = mock_ctx
mock_ctx.get_state = Mock(return_value="Entered")
print(mock_cm.get_state())
self.assertEqual("Not entered", mock_cm.get_state())
async with mock_cm as entered_ctx:
print(entered_ctx.get_state())
self.assertEqual("Entered", entered_ctx.get_state())
async def test_mock_has_awaits(self):
my_mock = AsyncMock()
my_mock.assert_not_awaited()
await my_mock(27)
my_mock.assert_awaited_once_with(27)
Reference
1 bbc点github点io/cloudfit减号public减号docs/asyncio/testing.html
