Python unit testing nested "async with", how to mock/patch?

Question

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)

Ytsen de Boer · Accepted Answer · 2019-12-14 13:13:59Z

2

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.)

answered Dec 14, 2019 at 13:13

Ytsen de Boer

3,1272 gold badges29 silver badges41 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

Python unit testing nested "async with", how to mock/patch?

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related