Title: Allow waiting on a mock
Type: enhancement Stage: patch review
Components: Library (Lib) Versions: Python 3.8
Status: open Resolution:
Dependencies: Superseder:
Assigned To: Nosy List: cjw296, ezio.melotti, jcea, lkollar, mariocj89, michael.foord, pitrou, rbcollins, xtreak
Priority: normal Keywords: patch

Created on 2013-01-22 12:15 by pitrou, last changed 2019-04-14 08:42 by lkollar.

Pull Requests
URL Status Linked Edit
PR 12818 open xtreak, 2019-04-13 15:36
Messages (7)
msg180377 - (view) Author: Antoine Pitrou (pitrou) * (Python committer) Date: 2013-01-22 12:15
In non-trivial tests, you may want to wait for a method to be called in another thread. This is a case where unittest.mock currently doesn't help. It would be nice to be able to write:

  myobj.some_method = Mock(side_effect=myobj.some_method)
  # launch some thread

And perhaps


(with an optional timeout?)

If we don't want every Mock object to embed a threading.Event, perhaps there could be a ThreadedMock subclass?
Or perhaps even:

  WaitableMock(..., event_class=threading.Event)

so that people can pass multiprocessing.Event if they want to wait on the mock from another process?
msg181398 - (view) Author: Michael Foord (michael.foord) * (Python committer) Date: 2013-02-05 00:14
There is a similar feature request on the mock issue tracker:

I prefer this proposal to the other one though. (Although technically allowing a wait for multiple calls is more flexible.)
msg247777 - (view) Author: Robert Collins (rbcollins) * (Python committer) Date: 2015-07-31 21:59
Now at
msg340146 - (view) Author: Karthikeyan Singaravelan (xtreak) * (Python triager) Date: 2019-04-13 07:03
Is there still sufficient interest in this? I gave an initial try at this based on my limited knowledge of mock and Antoine's idea. I created a new class WaitableMock that inherits from Mock and accepts an event_class with threading.Event as default and stores an event object. When call is made then via CallableMixin _mock_call is called which I have overridden to set the event object. I have wait_until_called that waits on this event object for a given timeout to return whether it's called or not. 

I am not sure of implementing wait_until_called_with since I set the event object for any call in _mock_call irrespective of argument and since the call params for which the event has to be set are passed to wait_until_called_with. Perhaps allow passing a call object to wait_until_called_with and and during _mock_call set event object only if the call is the same as one passed to wait_until_called_with ?

See also issue26467 to support asyncio with mock

Initial implementation 

class WaitableMock(Mock):

    def __init__(self, *args, **kwargs):
        event_class = kwargs.pop('event_class', threading.Event)
        _safe_super(WaitableMock, self).__init__(*args, **kwargs)
        self.event = event_class()

    def _mock_call(self, *args, **kwargs):
        _safe_super(WaitableMock, self)._mock_call(*args, **kwargs)

    def wait_until_called(self, timeout=1):
        return self.event.wait(timeout=timeout)

Sample program : 

import multiprocessing
import threading
import time
from unittest.mock import WaitableMock, patch

def call_after_sometime(func, delay=1):

def foo():

def bar():

with patch('', WaitableMock(event_class=threading.Event)):
    with patch('', WaitableMock(event_class=threading.Event)):
        threading.Thread(target=call_after_sometime, args=(foo, 1)).start()
        threading.Thread(target=call_after_sometime, args=(bar, 5)).start()
        print("foo called ", foo.wait_until_called(timeout=2)) # successful
        print("bar called ", bar.wait_until_called(timeout=2)) # times out
        # Wait for the bar's thread to complete to verify call is registered

# foo is called but waiting for call to bar times out and hence no calls to bar are registered though bar is eventually called in the end and the call is registered at the end of the program.

➜  cpython git:(master) ✗ time ./python.exe ../backups/
foo called  True
bar called  False
./python.exe ../backups/  0.40s user 0.05s system 7% cpu 5.765 total
msg340153 - (view) Author: Mario Corchero (mariocj89) * (Python triager) Date: 2019-04-13 11:27
I think this is REALLY interesting!, there are many situations where this has been useful, it would greatly improve multithreading testing in Python.

Q? I see you inherit from Mock, should it inherit from MagicMock?
I'd say send the PR and the discussion can happen there about the implementation, seems there is consensus in this thread.

I would find it OK to start with `wait_until_called` if you want and can be discussed in another issue/PR if you don't have the time/think it might not be worth.
msg340158 - (view) Author: Karthikeyan Singaravelan (xtreak) * (Python triager) Date: 2019-04-13 13:35
Thanks Mario for the feedback.

> I see you inherit from Mock, should it inherit from MagicMock?

yes, it can inherit from MagicMock to mock magic methods and to wait on their call.

I thought some more about waiting for function call with arguments. One idea would be to have a dictionary with args to function as key mapping to an event object and to set and wait on that event object. To have wait_until_called that is always listening on a per mock object and to have wait_until_called_with listening on event object specific to that argument. Below is a sample implementation and an example to show it working with wait_until_called and wait_until_called_with. wait_until_called_with is something difficult to model since it needs per call event object and also need to store relevant key mapping to event object that is currently args and doesn't support keyword arguments. But wait_until_called is little simpler waiting on a per mock event object.

I will open up a PR with some tests.

class WaitableMock(MagicMock):

    def __init__(self, *args, **kwargs):
        _safe_super(WaitableMock, self).__init__(*args, **kwargs)
        self.event_class = kwargs.pop('event_class', threading.Event)
        self.event = self.event_class()
        self.expected_calls = {}

    def _mock_call(self, *args, **kwargs):
        ret_value  = _safe_super(WaitableMock, self)._mock_call(*args, **kwargs)

        for call in self._mock_mock_calls:
            event = self.expected_calls.get(call.args)
            if event and not event.is_set():

        # Always set per mock event object to ensure the function is called for wait_until_called.

        return ret_value

    def wait_until_called(self, timeout=1):
        return self.event.wait(timeout=timeout)

    def wait_until_called_with(self, *args, timeout=1):
        # If there are args create a per argument list event object and if not wait for per mock event object.
        if args:
            if args not in self.expected_calls:
                event = self.event_class()
                self.expected_calls[args] = event
                event = self.expected_calls[args]
            event = self.event

        return event.is_set() or event.wait(timeout=timeout)

# Sample program to wait on arguments, magic methods and validating wraps

import multiprocessing
import threading
import time
from unittest.mock import WaitableMock, patch, call

def call_after_sometime(func, *args, delay=1):

def wraps(*args):
    return 1

def foo(*args):

def bar(*args):

with patch('', WaitableMock(event_class=threading.Event, wraps=wraps)):
    with patch('', WaitableMock(event_class=threading.Event)):
        # Test normal call
        threading.Thread(target=call_after_sometime, args=(foo, 1), kwargs={'delay': 1}).start()
        threading.Thread(target=call_after_sometime, args=(bar, 1), kwargs={'delay': 5}).start()
        print("foo called ", foo.wait_until_called(timeout=2))
        print("bar called ", bar.wait_until_called(timeout=2))

        # Test wraps works
        assert foo() == 1

        # Test magic method
        threading.Thread(target=call_after_sometime, args=(foo.__str__, ), kwargs={'delay': 1}).start()
        print("foo.__str__ called ", foo.__str__.wait_until_called(timeout=2))
        print("bar.__str__ called ", bar.__str__.wait_until_called(timeout=2))


        # Test waiting for arguments
        threading.Thread(target=call_after_sometime, args=(bar, 1), kwargs={'delay': 1}).start()
        threading.Thread(target=call_after_sometime, args=(bar, 2), kwargs={'delay': 5}).start()
        print("bar called with 1 ", bar.wait_until_called_with(1, timeout=2))
        print("bar called with 2 ", bar.wait_until_called_with(2, timeout=2))
        print("bar called with 2 ", bar.wait_until_called_with(2))

$ ./python.exe ../backups/
foo called  True
bar called  False
foo.__str__ called  True
bar.__str__ called  False
bar called with 1  True
bar called with 2  False
bar called with 2  True
msg340164 - (view) Author: Mario Corchero (mariocj89) * (Python triager) Date: 2019-04-13 15:30
Kooning great! I would Add a test for the following (I think both fails with the proposed impl):

- The mock is called BEFORE calling wait_until_called_with
- I call the mock with arguments and then I call wait for call without arguments.
- using keyword arguments 

Also, I don’t have a great solution for it but it might be worth prefixing time-out with something in the wait_untill_called_with. In situations where the mocked object has a timeout parameter (which is a common argument for multithreaded apps).
Date User Action Args
2019-04-14 08:42:59lkollarsetnosy: + lkollar
2019-04-13 15:36:00xtreaksetkeywords: + patch
stage: needs patch -> patch review
pull_requests: + pull_request12743
2019-04-13 15:30:52mariocj89setmessages: + msg340164
2019-04-13 13:35:30xtreaksetmessages: + msg340158
2019-04-13 11:27:33mariocj89setmessages: + msg340153
2019-04-13 07:03:54xtreaksetnosy: + cjw296, xtreak, mariocj89
messages: + msg340146
2017-12-12 01:52:44cheryl.sabellasetstage: needs patch
versions: + Python 3.8, - Python 3.4
2015-07-31 21:59:15rbcollinssetnosy: + rbcollins
messages: + msg247777
2013-02-05 00:14:25michael.foordsetmessages: + msg181398
2013-01-22 19:07:55jceasetnosy: + jcea
2013-01-22 12:16:34ezio.melottisetnosy: + ezio.melotti
2013-01-22 12:15:47pitroucreate