Python单元测试嵌套“异步”，如何模拟/修补？答案

【问题标题】：Python unit testing nested "async with", how to mock/patch?Python单元测试嵌套“异步”，如何模拟/修补？
【发布时间】：2018-11-27 21:24:35
【问题描述】：

我一直在努力对一段异步代码进行单元测试，该代码使用嵌套的“async with”。

python 版本 3.6.3 aiohttp 3.4.4版

我要单元测试的函数的简单版：

async def main():

    url = 'http://www.google.com'
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.read()

而精简后的单元测试代码是这样的：

class AsyncMock(MagicMock):
    async def __call__(self, *args, **kwargs):
        return super(AsyncMock, self).__call__(*args, **kwargs)


class TestAsyncTest(unittest.TestCase):

    @patch('aiohttp.ClientSession', new_callable=AsyncMock)
    def test_async_test(self, mock_session):

        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(main())
        print('result={}'.format(result))
        loop.close()

问题：如何修补嵌套调用，我希望“get”函数引发异常。我怀疑它应该看起来像这样：

mock_session.__aenter__().get().__aenter__.side_effect = asyncio.TimeoutError()

但这给了我一个错误：

E
======================================================================
ERROR: test_async_test (test_test_async_unittest.TestAsyncTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/lib/python3.6/unittest/mock.py", line 1179, in patched
    return func(*args, **keywargs)
  File "/ntfs/projects/gsmg/scratch/test_test_async_unittest.py", line 18, in test_async_test
    mock_session.__aenter__().get().__aenter__.side_effect = asyncio.TimeoutError()
  File "/usr/lib/python3.6/unittest/mock.py", line 584, in __getattr__
    raise AttributeError(name)
AttributeError: __aenter__

----------------------------------------------------------------------
Ran 1 test in 0.004s

FAILED (errors=1)

但是，如果我在没有该行的情况下运行它（只是上面发布的代码），我会收到以下错误：

E
======================================================================
ERROR: test_async_test (test_test_async_unittest.TestAsyncTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/lib/python3.6/unittest/mock.py", line 1179, in patched
    return func(*args, **keywargs)
  File "/ntfs/projects/gsmg/scratch/test_test_async_unittest.py", line 19, in test_async_test
    result = loop.run_until_complete(main())
  File "/usr/lib/python3.6/asyncio/base_events.py", line 473, in run_until_complete
    return future.result()
  File "/ntfs/projects/gsmg/scratch/test_async_unittest.py", line 8, in main
    async with aiohttp.ClientSession() as session:
AttributeError: __aexit__

----------------------------------------------------------------------
Ran 1 test in 0.007s

FAILED (errors=1)

【问题讨论】：

标签： python-3.x unit-testing python-asyncio

【解决方案1】：

最后，我找到了解决方法，可能值得在这里发布：

使用模块asynctest 代替unittest。这自然会处理异步上下文。

下面的代码说明了原理：

import asynctest

class Test(asynctest.TestCase):

    @patch('aiohttp.ClientSession.get')
    async def test_demo(self, mock_get):
        mock_get.side_effect = ValueError('mock exception');

（不接受它作为“真正的”答案，因为它是一种变通方法。）

【讨论】：