I've been struggling a bit with unit testing a piece of asynchronous code, that uses nested "async with".
python version 3.6.3 aiohttp version 3.4.4
The bare knuckles version of the function that I want to unit test:
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()
And the stripped down unit test code is like so:
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()
Question: How to patch the nested calls, I want the "get" function to raise an exception. I suspect it should look something like this:
mock_session.__aenter__().get().__aenter__.side_effect = asyncio.TimeoutError()
But that gives me an error:
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)
However, if I run it without that line (just the code posted above), I get this error:
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)
Finally, I found a way around it, which may be worth posting here:
Use the module asynctest
in stead of unittest
.
That naturally handles the async contexts.
Code below shows the principle:
import asynctest
class Test(asynctest.TestCase):
@patch('aiohttp.ClientSession.get')
async def test_demo(self, mock_get):
mock_get.side_effect = ValueError('mock exception');
(Not accepting it as the "real" answer, because it is a work-around.)
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With