How do you assert ensure_future calls

[argaen]

I want to assert that an ensure_future call is being called with a specific coroutine. The fast way is:

        with patch('asyncio.ensure_future') as ensure_future:
            await whatever()  # it has the ensure_future call inside

            ensure_future.assert_called_once_with(my_coroutine)
            my_coroutine.assert_called_once_with('arg1', 'arg2')

problem is that this doesn't work because ensure_future is being called with a generator instance returned by the async function. For now, to solve this case I've done the following:

        my_coroutine = MagicMock(return_value="my_coroutine_generator") # the mocking here is wrong but its to simplify the example
        with patch('asyncio.ensure_future') as ensure_future:
            await whatever()  # it has the ensure_future call inside

            ensure_future.assert_called_once_with("my_coroutine_generator")
            my_coroutine.assert_called_once_with('arg1', 'arg2')

but doesn't feel correct at all. Do you have any example of how this could be done?

[Martiusweb]

ensure_future is a function which returns a task, you must mock it accordingly.

If you want to assert that ensure_future as been called but you don't want it's behavior to be mocked, you can set the original function as a side effect:

with patch('asyncio.ensure_future', side_effect=asyncio.ensure_future) as ensure_future_mock:
    await whatever()

if you want ensure_future to return a mock coroutine, you must tell it explicitly:

with patch('asyncio.ensure_future', side_effect=asynctest.CoroutineMock) as ensure_future_mock:
    await whatever()

[argaen]

my issue is more focused on how to ensure that ensure_future is being called with a specific coroutine generated by function X, not about mocking or spying the ensure_future behavior.

Let's say I have

async def fn():
   pass

await asyncio.ensure_future(fn())

My goal is to ensure that ensure_future is being called with a coroutine that comes from fn. This I'm already doing more or less with the example I wrote in the first post but just want to know if there is a better way

[Martiusweb]

You can make your assertions on the ensure_future_mock like you did (using assert_called_once_with). Your second approach seems the right one, but you don't need to set the return value, this will do the trick:

        ensure_future.assert_called_once_with(my_coroutine.return_value)

[argaen]

Ahh yeah good one. Thanks for the help! :)

[argaen]

Hey, so I just run into this issue that I wrote a while ago and seems that with the newest versions this is not possible anymore as the return_value doesn't equal what you get when you actually call your coroutine:

>>> from asynctest import CoroutineMock
>>> mock = CoroutineMock()
>>> mock.return_value
<MagicMock name='mock()' id='4514853944'>
>>> mock()
<generator object CoroutineMock._mock_call.<locals>.proxy at 0x10d1358b8>
>>>

version used is asynctest==0.12.2

Any ideas how to check this behavior in the newest versions?

[Kentzo]

@Martiusweb I think we can fix this behavior by:

1. Changing proxy to a class with attributes for result and _call
2. Instantiating it just once per mock

[Kentzo]

Alternatively a custom attribute can be added to each instance of proxy with a consistent value (e.g. id(self)) so it would be possible to distinguish it while checking ensure_future's arguments.

@Martiusweb What do you think?

(please re-open the issue)

[Martiusweb]

The first solution fixes the ensure_future.assert_called_with(coroutine_mock.return_value), but I think that replacing the proxy coroutine with a new class adds one more layer of complexity (and asynctest is already very complex).

I like the other solution because if we attach the original CoroutineMock object to the proxy somewhere, it can be used in other features (maybe).

I'm not sure how common it is to assert that ensure_future has been called on a coroutine function. Especially now that there's asyncio.create_task().

[argaen]

I'm not sure how common it is to assert that ensure_future has been called on a coroutine function. Especially now that there's asyncio.create_task().

would this check work with asyncio.create_task? Still for old versions not but for newer versions of Python happy to use itt

[Martiusweb]

The problem is the same with both implementations. I think asyncio.create_task() is a wrapper over asyncio.ensure_future()`` but setting the appropriate loop.

Can you explain me how/why you need to check this (ensure_future() called with a specific coroutine?). @Kentzo implemented assert_awaited*() methods for CoroutineMock (since 0.12), do you think it can work in your case?

I never needed to verify that a coroutine is indeed running in its own task. If this is actually useful, maybe there's a way to add this as a feature to asynctest.

[argaen]

@Kentzo implemented assert_awaited*() methods for CoroutineMock (since 0.12), do you think it can work in your case?

and its great to verify a coroutine has been awaited.

In my case I want to verify the coroutine has been scheduled for execution. For example when I did https://github.com/argaen/aiocache/commit/f073f8362580485bcd1e2c60b5f4bd1ef1a172df#diff-478b98fad18969a6b3292f5922178797R111 there is one test that asserts the function has been awaited (https://github.com/argaen/aiocache/commit/f073f8362580485bcd1e2c60b5f4bd1ef1a172df#diff-3d469ab1daafc08395d1e4c16ad9f3a6R168) while there is another one that verifies the coroutine has been called (https://github.com/argaen/aiocache/commit/f073f8362580485bcd1e2c60b5f4bd1ef1a172df#diff-3d469ab1daafc08395d1e4c16ad9f3a6R176) but in this last case, its not ideal because the coroutine has been called AND scheduled with asyncio.ensure_future. And this last thing I can't check and what I would like to

Not sure if this makes sense or if there is a better way to do this, open to suggestions :)

[Kentzo]

How would you test against the following cases:

• Assuming you have a buggy code that schedules the awaitable twice instead of just once
• An "ensured" future that is later cancelled without a chance of being executed

[Martiusweb]

@argaen, I understand your point, you would like to assert that a coroutine has been scheduled asynchronously.

It's seems rather complicated to do well, and that it doesn't really guarantee that this coroutine running in the background will indeed be executed later.

In your test, you would like to test both that "the coroutine has been passed to ensure_future()" and that "the coroutine has not been awaited". In theory, there's a race condition and you can't really be sure that the 2nd assertion will be valid by the time it is checked.

I can't really think of situations when checking this really makes sense.

I've got some suggestions about the example you linked. Since it's out of the scope of the issue and that I don't know the big picture of your code, I'm sorry if this sounds incorrect to you.

I think that the simplest way to test this is to have 2 methods: a self.write_now() which is a coroutine calling set_in_cache(), an a plain function self.write_later() which only schedules the call. With this, you can easily ensure that you fall either in one case or the other.

I also think that the "one task per write" like you do can be problematic (that's maybe why it's so hard to test): if you don't keep track of these tasks (the result of ensure_future()), you can't be sure that it will succeed: it can raise an exception, be cancelled, etc. Since it's a cache, it might be OK, but a user of your lib might end up with logs filled by "Task exception was never retrieved" messages (and nothing to do about it).

The other issue is that scheduling writes asynchronously (especially to a database) in its own task introduces a race condition because you can't know which task will finish first. IMHO, a safer approach is to push all the writes to a queue, and depile them in a single background task.

[Kentzo]

We could track the lifetime of all coroutines and capture it in a digraph (e.g. via the networkx library). The nodes of such digraph would be:

• call objects (construction of a coroutine)
• await objects (execution of a coroutine)
• scheduling and unscheduling (cancelation)

The test authors could then traverse the graph to verify against expected patterns either explicitly or via some mini-lang that allows partial matching.

[argaen]

It's seems rather complicated to do well, and that it doesn't really guarantee that this coroutine running in the background will indeed be executed later.

Agreed, but my point is that I want to test that it has been scheduled. In there I really don't care if it has been executed or not. Its an asyncio problem once I send it to ensure_future :).

In your test, you would like to test both that "the coroutine has been passed to ensure_future()" and that "the coroutine has not been awaited". In theory, there's a race condition and you can't really be sure that the 2nd assertion will be valid by the time it is checked.

Nope, for the case of ensure_future I don't want to test if it has been awaited or not. I just pass responsibility to ensure_future and believe it will do its job.

if you don't keep track of these tasks (the result of ensure_future()), you can't be sure that it will succeed: it can raise an exception, be cancelled, etc.

In this case I'm happy with just logs.. could improve the messaging but imagine a function takes 50s, and then you fail with an exception because the write to the cache failed... That's bad :P. Agree there are things to improve though

IMHO, a safer approach is to push all the writes to a queue, and depile them in a single background task.

That's a good point, need to think if the effort is worth it regarding the ordering of writes. By default asyncio uses FIFO for executing but its an implicit behavior so shouldn't rely on it.