This issue tracker has been migrated to GitHub, and is currently read-only.
For more information, see the GitHub FAQs in the Python's Developer Guide.

Author xtreak
Recipients cjw296, ezio.melotti, jcea, mariocj89, michael.foord, pitrou, rbcollins, xtreak
Date 2019-04-13.07:03:54
SpamBayes Score -1.0
Marked as misclassified Yes
Message-id <1555139034.9.0.598192064042.issue17013@roundup.psfhosted.org>
In-reply-to
Content
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)
        self.event.set()

    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):
    time.sleep(delay)
    func()

def foo():
    pass

def bar():
    pass

with patch('__main__.foo', WaitableMock(event_class=threading.Event)):
    with patch('__main__.bar', 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
        foo.assert_called_once()
        bar.assert_not_called()
        # Wait for the bar's thread to complete to verify call is registered
        time.sleep(5)
        bar.assert_called_once()

# 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/bpo17013_mock.py
foo called  True
bar called  False
./python.exe ../backups/bpo17013_mock.py  0.40s user 0.05s system 7% cpu 5.765 total
History
Date User Action Args
2019-04-13 07:03:54xtreaksetrecipients: + xtreak, jcea, pitrou, rbcollins, cjw296, ezio.melotti, michael.foord, mariocj89
2019-04-13 07:03:54xtreaksetmessageid: <1555139034.9.0.598192064042.issue17013@roundup.psfhosted.org>
2019-04-13 07:03:54xtreaklinkissue17013 messages
2019-04-13 07:03:54xtreakcreate