Message340146
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 |
|
Date |
User |
Action |
Args |
2019-04-13 07:03:54 | xtreak | set | recipients:
+ xtreak, jcea, pitrou, rbcollins, cjw296, ezio.melotti, michael.foord, mariocj89 |
2019-04-13 07:03:54 | xtreak | set | messageid: <1555139034.9.0.598192064042.issue17013@roundup.psfhosted.org> |
2019-04-13 07:03:54 | xtreak | link | issue17013 messages |
2019-04-13 07:03:54 | xtreak | create | |
|