How to mock asyncio coroutines?

Since mock library doesn't support coroutines I create mocked coroutines manually and assign those to mock object. A bit more verbose but it works.

Your example may look like this:

import asyncioimport unittestfrom unittest.mock import Mockclass ImGoingToBeMocked:    @asyncio.coroutine    def yeah_im_not_going_to_run(self):        yield from asyncio.sleep(1)        return "sup"class ImBeingTested:    def __init__(self, hidude):        self.hidude = hidude    @asyncio.coroutine    def i_call_other_coroutines(self):        return (yield from self.hidude.yeah_im_not_going_to_run())class TestImBeingTested(unittest.TestCase):    def test_i_call_other_coroutines(self):        mocked = Mock(ImGoingToBeMocked)        ibt = ImBeingTested(mocked)        @asyncio.coroutine        def mock_coro():            return "sup"        mocked.yeah_im_not_going_to_run = mock_coro        ret = asyncio.get_event_loop().run_until_complete(            ibt.i_call_other_coroutines())        self.assertEqual("sup", ret)if __name__ == '__main__':    unittest.main()

I am writting a wrapper to unittest which aims at cutting the boilerplate when writting tests for asyncio.

The code lives here: https://github.com/Martiusweb/asynctest

You can mock a coroutine with asynctest.CoroutineMock:

>>> mock = CoroutineMock(return_value='a result')>>> asyncio.iscoroutinefunction(mock)True>>> asyncio.iscoroutine(mock())True>>> asyncio.run_until_complete(mock())'a result'

It also works with the side_effect attribute, and an asynctest.Mock with a spec can return CoroutineMock:

>>> asyncio.iscoroutinefunction(Foo().coroutine)True>>> asyncio.iscoroutinefunction(Foo().function)False>>> asynctest.Mock(spec=Foo()).coroutine<class 'asynctest.mock.CoroutineMock'>>>> asynctest.Mock(spec=Foo()).function<class 'asynctest.mock.Mock'>

All the features of unittest.Mock are expected to work correctly (patch(), etc).

Springing off of Andrew Svetlov's answer, I just wanted to share this helper function:

def get_mock_coro(return_value):    @asyncio.coroutine    def mock_coro(*args, **kwargs):        return return_value    return Mock(wraps=mock_coro)

This lets you use the standard assert_called_with, call_count and other methods and attributes a regular unittest.Mock gives you.

You can use this with code in the question like:

class ImGoingToBeMocked:    @asyncio.coroutine    def yeah_im_not_going_to_run(self):        yield from asyncio.sleep(1)        return "sup"class ImBeingTested:    def __init__(self, hidude):        self.hidude = hidude    @asyncio.coroutine    def i_call_other_coroutines(self):        return (yield from self.hidude.yeah_im_not_going_to_run())class TestImBeingTested(unittest.TestCase):    def test_i_call_other_coroutines(self):        mocked = Mock(ImGoingToBeMocked)        mocked.yeah_im_not_going_to_run = get_mock_coro()        ibt = ImBeingTested(mocked)        ret = asyncio.get_event_loop().run_until_complete(ibt.i_call_other_coroutines())        self.assertEqual(mocked.yeah_im_not_going_to_run.call_count, 1)