Mocking async call in python 3.5

How do I mock async call from one native coroutine to other one using unittest.mock.patch?

I currently have quite an awkward solution:

class CoroutineMock(MagicMock):
    def __await__(self, *args, **kwargs):
        future = Future()
        future.set_result(self)
        result = yield from future
        return result

Then

class TestCoroutines(TestCase):
    @patch('some.path', new_callable=CoroutineMock)
    def test(self, mock):
        some_action()
        mock.assert_called_with(1,2,3)

This works but looks ugly. Is there more pythonic way to do this?

The solution was actually quite simple: I just needed to convert __call__ method of mock into coroutine:

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

This works perfectly, when mock is called, code receives native coroutine

Example usage:

@mock.patch('my.path.asyncio.sleep', new_callable=AsyncMock)
def test_stuff(sleep):
    # code

Everyone's missing what's probably the simplest and clearest solution:

@patch('some.path')
def test(self, mock):
    f = asyncio.Future()
    f.set_result('whatever result you want')
    process_smtp_message.return_value = f
    mock.assert_called_with(1, 2, 3)

remember a coroutine can be thought of as just a function which is guaranteed to return a future which can, in turn be awaited.

Based on @scolvin answer I created this (imo) cleaner way:

import asyncio

def async_return(result):
    f = asyncio.Future()
    f.set_result(result)
    return f

That's it, just use it around whatever return you want to be async, as in

mock = MagicMock(return_value=async_return("Example return"))
await mock()

Subclassing MagicMock will propagate your custom class for all the mocks generated from your coroutine mock. For instance, AsyncMock().__str__ will also become an AsyncMock which is probably not what you're looking for.

Instead, you might want to define a factory that creates a Mock (or a MagicMock) with custom arguments, for instance side_effect=coroutine(coro). Also, it might be a good idea to separate the coroutine function from the coroutine (as explained in the documentation).

Here is what I came up with:

from asyncio import coroutine

def CoroMock():
    coro = Mock(name="CoroutineResult")
    corofunc = Mock(name="CoroutineFunction", side_effect=coroutine(coro))
    corofunc.coro = coro
    return corofunc

An explanation of the different objects:

• corofunc: the coroutine function mock
• corofunc.side_effect(): the coroutine, generated for each call
• corofunc.coro: the mock used by the coroutine to get the result
• corofunc.coro.return_value: the value returned by the coroutine
• corofunc.coro.side_effect: might be used to raise an exception

Example:

async def coro(a, b):
    return await sleep(1, result=a+b)

def some_action(a, b):
    return get_event_loop().run_until_complete(coro(a, b))

@patch('__main__.coro', new_callable=CoroMock)
def test(corofunc):
    a, b, c = 1, 2, 3
    corofunc.coro.return_value = c
    result = some_action(a, b)
    corofunc.assert_called_with(a, b)
    assert result == c

Another way of mocking coroutine is to make coroutine, that returns mock. This way you can mock coroutines that will be passed into asyncio.wait or asyncio.wait_for.

This makes more universal coroutines though makes setup of tests more cumbersome:

def make_coroutine(mock)
    async def coroutine(*args, **kwargs):
        return mock(*args, **kwargs)
    return coroutine


class Test(TestCase):
    def setUp(self):
        self.coroutine_mock = Mock()
        self.patcher = patch('some.coroutine',
                             new=make_coroutine(self.coroutine_mock))
        self.patcher.start()
    
    def tearDown(self):
        self.patcher.stop()

One more variant of "simplest" solution to mock a async object, which is just a one liner.

In source:

class Yo:
    async def foo(self):
        await self.bar()
    async def bar(self):
        # Some code

In test:

from asyncio import coroutine

yo = Yo()
# Here bounded method bar is mocked and will return a customised result.
yo.bar = Mock(side_effect=coroutine(lambda:'the awaitable should return this'))
event_loop.run_until_complete(yo.foo())

You can set the return_value of an async method like so:

mock = unittest.mock.MagicMock()
mock.your_async_method.return_value = task_from_result(your_return_value)

async def task_from_result(result):
    return result

The caller will have to do await your_async_method(..) just like as if the method wasn't mocked.

I don't know why nobody mentioned the default option available. python provides a Async version of MagicMock.

You can read more about this here. https://docs.python.org/3/library/unittest.mock.html#unittest.mock.AsyncMock

In case you are using patch then also you don't need to make any other change. It will automatically replace it with a async mock function if required. Read more here https://docs.python.org/3/library/unittest.mock.html#patch

I like this approach, which also makes AsyncMock to behave exactly like Mock:

class AsyncMock:
    def __init__(self, *args, **kwargs):
        self.mock = Mock(*args, **kwargs)

    async def __call__(self, *args, **kwargs):
        return self.mock(*args, **kwargs)

    def __getattr__(self, item):
        return getattr(self.mock, item)

Then you can work with it in the same way like with Mock, i.e:

@pytest.mark.asyncio
async def test_async_mock_example(monkeypatch):
    fn = AsyncMock(side_effect=ValueError)
    with pytest.raises(ValueError):
        await fn()
    assert fn.call_count == 1