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

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():
                event.set()

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

        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
            else:
                event = self.expected_calls[args]
        else:
            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):
    time.sleep(delay)
    func(*args)

def wraps(*args):
    return 1

def foo(*args):
    pass

def bar(*args):
    pass

with patch('__main__.foo', WaitableMock(event_class=threading.Event, 
wraps=wraps)):
    with patch('__main__.bar', 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))
        foo.assert_called_once()
        bar.assert_not_called()

        # 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))

        foo.reset_mock()
        bar.reset_mock()

        # 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))
        time.sleep(5)
        print("bar called with 2 ", bar.wait_until_called_with(2))


$ ./python.exe ../backups/bpo17013_mock.py
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

----------

_______________________________________
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