python-3.x - Python单元测试嵌套 "async with"，如何模拟/修补？

Question

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

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)

Answer 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');

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