Karthikeyan Singaravelan <tir.kar...@gmail.com> added the comment:

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

----------
nosy: +cjw296, mariocj89, xtreak

_______________________________________
Python tracker <rep...@bugs.python.org>
<https://bugs.python.org/issue17013>
_______________________________________
_______________________________________________
Python-bugs-list mailing list
Unsubscribe: 
https://mail.python.org/mailman/options/python-bugs-list/archive%40mail-archive.com

Reply via email to