KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use add_listener method in ATX

Best Python code snippet using ATX

event_manager_test.py

Source:event_manager_test.py

2from mock import Mock3from pokemongo_bot.event_manager import EventManager, Event4class EventManagerTest(unittest.TestCase):5    @staticmethod6    def test_add_listener():7        event_manager = EventManager()8        test_listener = Mock()9        event_manager.add_listener('test', test_listener)10        assert isinstance(event_manager.events['test'], Event)11        assert len(event_manager.events['test'].listeners[0]) == 112    @staticmethod13    def test_add_listener_priorities():14        event_manager = EventManager()15        test_listener = Mock()16        event_manager.add_listener('test', test_listener, priority=-100)17        event_manager.add_listener('test', test_listener, priority=100)18        assert isinstance(event_manager.events['test'], Event)19        assert len(event_manager.events['test'].listeners[-100]) == 120        assert len(event_manager.events['test'].listeners[100]) == 121    @staticmethod22    def test_remove_listener():23        event_manager = EventManager()24        test_listener = Mock()25        event_manager.add_listener('test', test_listener, priority=10)26        assert isinstance(event_manager.events['test'], Event)27        assert len(event_manager.events['test'].listeners) == 128        event_manager.remove_listener('test', test_listener)29        assert isinstance(event_manager.events['test'], Event)30        assert len(event_manager.events['test'].listeners[10]) == 031    @staticmethod32    def test_get_registered_events():33        event_manager = EventManager()34        test_listener = Mock()35        event_manager.add_listener('test', test_listener)36        events = event_manager.get_registered_events()37        assert len(events) == 138        assert 'test' in events39    @staticmethod40    def test_fire():41        event_manager = EventManager()42        # pylint: disable=unused-argument43        def test_listener(value=None):44            return {45                'value': 'first'46            }47        # pylint: disable=unused-argument48        def test_listener_late(value=None):49            return {50                'value': 'second'51            }52        event_manager.add_listener('test', test_listener_late, priority=100)53        event_manager.add_listener('test', test_listener, priority=-100)54        return_data = event_manager.fire('test', value=None)55        assert return_data['value'] == 'second'56    @staticmethod57    def test_fire_cancelled():58        event_manager = EventManager()59        def test_listener():60            return {61                'value': 'first'62            }63        def test_listener_late():64            return False65        event_manager.add_listener('test', test_listener_late, priority=100)66        event_manager.add_listener('test', test_listener, priority=-100)67        return_data = event_manager.fire('test', value=None)68        assert return_data is False69    @staticmethod70    def test_fire_no_listeners():71        import sys72        from io import StringIO73        out = StringIO()74        sys.stdout = out75        event_manager = EventManager()76        event_manager.events['test'] = Event('test')77        event_manager.fire('test', value=None)78        assert 'WARNING: No handler has registered to handle event "test"' in out.getvalue().strip()79    @staticmethod80    def test_fire_with_context():81        event_manager = EventManager()82        def test_listener(bot=None):83            bot()84        event_manager.add_listener('test', test_listener, priority=-100)85        bot = Mock()86        event_manager.fire_with_context('test', bot)87        bot.assert_called_once()88    @staticmethod89    def test_event_pipeline_empty_output():90        import sys91        from io import StringIO92        out = StringIO()93        sys.stdout = out94        event_manager = EventManager()95        event_manager.events['test'] = Event('test')96        event_manager.events['test'].listeners['test_listener'] = []97        event_manager.print_all_event_pipelines()98        assert 'Event pipeline for "test" is empty.' in out.getvalue().strip()99    @staticmethod100    def test_event_pipeline_no_listeners():101        import sys102        from io import StringIO103        out = StringIO()104        sys.stdout = out105        event_manager = EventManager()106        event_manager.events['test'] = Event('test')107        event_manager.events['test'].listeners['test_listener'] = []108        event_manager.print_all_event_pipelines()109        assert 'Event pipeline for "test" is empty.' in out.getvalue().strip()110    @staticmethod111    def test_event_pipeline_output():112        import sys113        from io import StringIO114        out = StringIO()115        sys.stdout = out116        event_manager = EventManager()117        def test_listener_1(bot=None):118            bot()119        def test_listener_2(bot=None):120            bot()121        def test_listener_3(bot=None):122            bot()123        def test_listener_4(bot=None):124            bot()125        event_manager.add_listener('test1', test_listener_1, priority=-100)126        event_manager.add_listener('test1', test_listener_2, priority=0)127        event_manager.add_listener('test1', test_listener_3, priority=0)128        event_manager.add_listener('test1', test_listener_4, priority=100)129        event_manager.add_listener('test2', test_listener_1, priority=100)130        event_manager.print_all_event_pipelines()131        assert 'Event pipeline for "test1":' in out.getvalue().strip()132        assert '-100 (test_listener_1) -> 0 (test_listener_2 -> test_listener_3) -> 100 (test_listener_4)' in out.getvalue().strip()133        assert 'Event pipeline for "test2":' in out.getvalue().strip()...

test_events.py

Source:test_events.py

...25        self.manager = DOMEventFlowManager(self.doc)26    def get_node(self, expr):27        [node] = self.engine.evaluate(expr, self.doc).value28        return node29    def add_listener(self, target_expr, event_type, listener, capture=False):30        target = self.get_node(target_expr)31        self.manager.add_event_listener(target, event_type, listener, capture)32    def make_incr_handler(self, data, key):33        def handler(event):34            data.setdefault(key, 0)35            data[key] += 136        return handler37    def test_dispatch_events_no_listeners(self):38        self.manager.dispatch_event(self.get_node('/'), ToyEvent())39        self.manager.dispatch_event(self.get_node('//elem1'), ToyEvent())40        self.manager.dispatch_event(self.get_node('//elem2'), ToyEvent())41        self.manager.dispatch_event(self.get_node('//elem3'), ToyEvent())42    def test_dispatch_event_target_listeners(self):43        data = {}44        listener1 = self.make_incr_handler(data, 'listener1')45        listener2 = self.make_incr_handler(data, 'listener2')46        self.add_listener('//elem1', 'toy-event', listener1)47        self.add_listener('//elem2', 'toy-event', listener1)48        self.add_listener('//elem3', 'toy-event', listener1)49        self.add_listener('//elem2', 'toy-event', listener2)50        self.assertEqual(data, {})51        self.manager.dispatch_event(self.get_node('//elem2'), ToyEvent())52        self.assertEqual(data, {'listener1': 1, 'listener2': 1})53    def test_dispatch_event_target_and_bubble_listeners(self):54        data = {}55        listener1 = self.make_incr_handler(data, 'listener1')56        listener2 = self.make_incr_handler(data, 'listener2')57        self.add_listener('//elem1', 'bubble-event', listener1)58        self.add_listener('//elem2', 'bubble-event', listener1)59        self.add_listener('//elem3', 'bubble-event', listener1)60        self.add_listener('//elem2', 'bubble-event', listener2)61        self.assertEqual(data, {})62        self.manager.dispatch_event(self.get_node('//elem2'), BubbleEvent())63        self.assertEqual(data, {'listener1': 2, 'listener2': 1})64    def test_dispatch_event_bubble_listeners(self):65        data = {}66        listener1 = self.make_incr_handler(data, 'listener1')67        self.add_listener('//elem1', 'bubble-event', listener1)68        self.add_listener('//elem2', 'bubble-event', listener1)69        self.assertEqual(data, {})70        self.manager.dispatch_event(self.get_node('//elem3'), BubbleEvent())71        self.assertEqual(data, {'listener1': 2})72    def test_dispatch_event_target_capture_listeners(self):73        data = {}74        listener1 = self.make_incr_handler(data, 'listener1')75        listener2 = self.make_incr_handler(data, 'listener2')76        self.add_listener('//elem1', 'toy-event', listener1, True)77        self.add_listener('//elem2', 'toy-event', listener1, True)78        self.add_listener('//elem3', 'toy-event', listener1, True)79        self.add_listener('//elem2', 'toy-event', listener2)80        self.assertEqual(data, {})81        self.manager.dispatch_event(self.get_node('//elem2'), ToyEvent())82        self.assertEqual(data, {'listener1': 1, 'listener2': 1})83    def test_dispatch_event_target_capture_bubble_listeners(self):84        data = {}85        listener1 = self.make_incr_handler(data, 'listener1')86        listener2 = self.make_incr_handler(data, 'listener2')87        self.add_listener('//elem1', 'bubble-event', listener1, True)88        self.add_listener('//elem2', 'bubble-event', listener1, True)89        self.add_listener('//elem3', 'bubble-event', listener1, True)90        self.add_listener('//elem1', 'bubble-event', listener2)91        self.add_listener('//elem2', 'bubble-event', listener2)92        self.add_listener('//elem3', 'bubble-event', listener2)93        self.assertEqual(data, {})94        self.manager.dispatch_event(self.get_node('//elem2'), BubbleEvent())...

test_event_dispatcher.py

Source:test_event_dispatcher.py

...24def test_initial_state(dispatcher):25    assert {} == dispatcher.get_listeners()26    assert not dispatcher.has_listeners(PRE_FOO)27    assert not dispatcher.has_listeners(POST_FOO)28def test_add_listener(dispatcher, listener):29    dispatcher.add_listener(PRE_FOO, listener.pre_foo)30    dispatcher.add_listener(POST_FOO, listener.post_foo)31    assert dispatcher.has_listeners()32    assert dispatcher.has_listeners(PRE_FOO)33    assert dispatcher.has_listeners(POST_FOO)34    assert len(dispatcher.get_listeners(PRE_FOO)) == 135    assert len(dispatcher.get_listeners(POST_FOO)) == 136    assert len(dispatcher.get_listeners()) == 237def test_get_listeners_sorts_by_priority(dispatcher):38    listener1 = EventListener()39    listener2 = EventListener()40    listener3 = EventListener()41    dispatcher.add_listener(PRE_FOO, listener1.pre_foo, -10)42    dispatcher.add_listener(PRE_FOO, listener2.pre_foo, 10)43    dispatcher.add_listener(PRE_FOO, listener3.pre_foo)44    expected = [listener2.pre_foo, listener3.pre_foo, listener1.pre_foo]45    assert expected == dispatcher.get_listeners(PRE_FOO)46def test_get_all_listeners_sorts_by_priority(dispatcher):47    listener1 = EventListener()48    listener2 = EventListener()49    listener3 = EventListener()50    listener4 = EventListener()51    listener5 = EventListener()52    listener6 = EventListener()53    dispatcher.add_listener(PRE_FOO, listener1.pre_foo, -10)54    dispatcher.add_listener(PRE_FOO, listener2.pre_foo)55    dispatcher.add_listener(PRE_FOO, listener3.pre_foo, 10)56    dispatcher.add_listener(POST_FOO, listener4.pre_foo, -10)57    dispatcher.add_listener(POST_FOO, listener5.pre_foo)58    dispatcher.add_listener(POST_FOO, listener6.pre_foo, 10)59    expected = {60        PRE_FOO: [listener3.pre_foo, listener2.pre_foo, listener1.pre_foo],61        POST_FOO: [listener6.pre_foo, listener5.pre_foo, listener4.pre_foo],62    }63    assert dispatcher.get_listeners() == expected64def test_get_listener_priority(dispatcher):65    listener1 = EventListener()66    listener2 = EventListener()67    dispatcher.add_listener(PRE_FOO, listener1.pre_foo, -10)68    dispatcher.add_listener(PRE_FOO, listener2.pre_foo)69    assert dispatcher.get_listener_priority(PRE_FOO, listener1.pre_foo) == -1070    assert dispatcher.get_listener_priority(PRE_FOO, listener2.pre_foo) == 071    assert dispatcher.get_listener_priority(PRE_BAR, listener2.pre_foo) is None72def test_dispatch(dispatcher, listener):73    dispatcher.add_listener(PRE_FOO, listener.pre_foo)74    dispatcher.add_listener(POST_FOO, listener.post_foo)75    dispatcher.dispatch(Event(), PRE_FOO)76    assert listener.pre_foo_invoked77    assert not listener.post_foo_invoked78def test_stop_event_propagation(dispatcher, listener):79    other_listener = EventListener()80    dispatcher.add_listener(POST_FOO, listener.post_foo, 10)81    dispatcher.add_listener(POST_FOO, other_listener.post_foo)82    dispatcher.dispatch(Event(), POST_FOO)83    assert listener.post_foo_invoked...

bindable_variable_demo.py

Source:bindable_variable_demo.py

...14        print("Variable changed: name={}, old_val={}, new_val={}".format(var.name, old_val, new_val))15    # Basic manipulations16    print()17    v = Var(42, name="age")18    v.add_listener(_bindable_var_changed)19    print("::Setting age to 23 ...")20    v.value = 2321    print("::Incrementing age with v.value += 1 ...")22    v.value += 123    print("::Incrementing age with v += 1 ...")24    v += 125    # new_value-only notifications26    print()27    v = Var("Title")28    v.add_listener(print, value_only=True)29    print("::Notification can optionally have one argument, the new value ...")30    v.value = "New Title"31    # No argument notifications32    print()33    v = Var("Something")34    v.add_listener(print, no_args=True)35    print("::Notification will be called with no arguments ...")36    v.value = "Else"37    print("::Blank line above? Value is now", v.value)38    # Value is itself mutable39    print()40    v = Var()41    v.add_listener(print)42    v.value = ["cat", "dog", "elephant"]43    print("::No notification if list item is changed directly ...")44    v.value.append("lion")45    print("::List is now", v.value)46    print("::Notifications happen at end of 'with' construct ...")47    print("::Note that it is not possible to separate old value and new value.")48    with v:49        v.value[2] = "mouse"50        v.value[3] = "gerbil"51        v.value.append("fish")52    print("::An ugly way to abort notifications according to some test ...")53    with v as unchanged:54        if "some truth test":55            raise unchanged56        else:57            v.value.append("chimpanze")58    # Using the FormattableVar59    person_name = Var("Joe")60    person_age = Var(23)61    fv = FormattableVar("{} is {} years old", [person_name, person_age])62    fv.add_listener(print, value_only=True)63    person_age += 164    # Using tk65    import tkinter as tk66    root = tk.Tk()67    root.title("Bindable Variable")68    v = Var()69    v.add_listener(print, value_only=True)  # echo to console70    entry = tk.Entry(root, textvariable=v.tk_var())  # create a bound tk.Var71    entry.pack()72    v.value = "intial"73    tk.mainloop()74def demo_bindable_dictionary():75    def _bindable_dict_changed(d, key, old_val, new_val):76        print("Dictionary changed: key={}, old_val={}, new_val={}".format(key, old_val, new_val), flush=True)77    d = BindableDict()78    d.add_listener(_bindable_dict_changed)79    d.set("foo", "bar")80    d.set("cats", 4)81    with d:82        d.set("dogs", 0)83        print("before or after?", flush=True)84        d.set("cats", 5)85    a = {"pencil": "yellow", "cup": "full"}86    d.update(a)87if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.