|
|
import unittest
|
|
|
from unittest.mock import Mock
|
|
|
|
|
|
from IPython.core import events
|
|
|
import IPython.testing.tools as tt
|
|
|
|
|
|
|
|
|
@events._define_event
|
|
|
def ping_received():
|
|
|
pass
|
|
|
|
|
|
|
|
|
@events._define_event
|
|
|
def event_with_argument(argument):
|
|
|
pass
|
|
|
|
|
|
|
|
|
class CallbackTests(unittest.TestCase):
|
|
|
def setUp(self):
|
|
|
self.em = events.EventManager(get_ipython(),
|
|
|
{'ping_received': ping_received,
|
|
|
'event_with_argument': event_with_argument})
|
|
|
|
|
|
def test_register_unregister(self):
|
|
|
cb = Mock()
|
|
|
|
|
|
self.em.register('ping_received', cb)
|
|
|
self.em.trigger('ping_received')
|
|
|
self.assertEqual(cb.call_count, 1)
|
|
|
|
|
|
self.em.unregister('ping_received', cb)
|
|
|
self.em.trigger('ping_received')
|
|
|
self.assertEqual(cb.call_count, 1)
|
|
|
|
|
|
def test_bare_function_missed_unregister(self):
|
|
|
def cb1():
|
|
|
...
|
|
|
|
|
|
def cb2():
|
|
|
...
|
|
|
|
|
|
self.em.register("ping_received", cb1)
|
|
|
self.assertRaises(ValueError, self.em.unregister, "ping_received", cb2)
|
|
|
self.em.unregister("ping_received", cb1)
|
|
|
|
|
|
def test_cb_error(self):
|
|
|
cb = Mock(side_effect=ValueError)
|
|
|
self.em.register('ping_received', cb)
|
|
|
with tt.AssertPrints("Error in callback"):
|
|
|
self.em.trigger('ping_received')
|
|
|
|
|
|
def test_cb_keyboard_interrupt(self):
|
|
|
cb = Mock(side_effect=KeyboardInterrupt)
|
|
|
self.em.register('ping_received', cb)
|
|
|
with tt.AssertPrints("Error in callback"):
|
|
|
self.em.trigger('ping_received')
|
|
|
|
|
|
def test_unregister_during_callback(self):
|
|
|
invoked = [False] * 3
|
|
|
|
|
|
def func1(*_):
|
|
|
invoked[0] = True
|
|
|
self.em.unregister('ping_received', func1)
|
|
|
self.em.register('ping_received', func3)
|
|
|
|
|
|
def func2(*_):
|
|
|
invoked[1] = True
|
|
|
self.em.unregister('ping_received', func2)
|
|
|
|
|
|
def func3(*_):
|
|
|
invoked[2] = True
|
|
|
|
|
|
self.em.register('ping_received', func1)
|
|
|
self.em.register('ping_received', func2)
|
|
|
|
|
|
self.em.trigger('ping_received')
|
|
|
self.assertEqual([True, True, False], invoked)
|
|
|
self.assertEqual([func3], self.em.callbacks['ping_received'])
|
|
|
|
|
|
def test_ignore_event_arguments_if_no_argument_required(self):
|
|
|
call_count = [0]
|
|
|
def event_with_no_argument():
|
|
|
call_count[0] += 1
|
|
|
|
|
|
self.em.register('event_with_argument', event_with_no_argument)
|
|
|
self.em.trigger('event_with_argument', 'the argument')
|
|
|
self.assertEqual(call_count[0], 1)
|
|
|
|
|
|
self.em.unregister('event_with_argument', event_with_no_argument)
|
|
|
self.em.trigger('ping_received')
|
|
|
self.assertEqual(call_count[0], 1)
|
|
|
|
