阿狗工具

(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