diff --git a/doc/qubes-events.rst b/doc/qubes-events.rst index 8857d549..269072bf 100644 --- a/doc/qubes-events.rst +++ b/doc/qubes-events.rst @@ -62,6 +62,7 @@ Note that your handler will be called for all instances of this class. .. TODO: extensions .. TODO: add/remove_handler +.. TODO: wildcards (property-set:*) Handling events with variable signature diff --git a/qubes/events.py b/qubes/events.py index 6f950b06..34bcd9fd 100644 --- a/qubes/events.py +++ b/qubes/events.py @@ -25,6 +25,7 @@ etc. ''' import asyncio import collections +import fnmatch import itertools @@ -35,14 +36,15 @@ def handler(*events): To hook an event, decorate a method in your plugin class with this decorator. - Some event handlers may be defined as coroutine. In such a case, *async* - should be set to :py:obj:``True``. + Some event handlers may be defined as coroutine. In such a case + :py:func:`asyncio.coroutine` decorator should be used after this one, + i.e. you should decorate a coroutine. See appropriate event documentation for details. .. note:: For hooking events from extensions, see :py:func:`qubes.ext.handler`. - :param str events: events + :param str events: events names, can contain basic wildcards (`*`, `?`) ''' def decorator(func): @@ -155,9 +157,9 @@ class Emitter(object, metaclass=EmitterMeta): handlers_dict = i.__handlers__ except AttributeError: continue - handlers = handlers_dict.get(event, set()) - if '*' in handlers_dict: - handlers = handlers_dict['*'] | handlers + handlers = [h_func for h_name, h_func_set in handlers_dict.items() + for h_func in h_func_set + if fnmatch.fnmatch(event, h_name)] for func in sorted(handlers, key=(lambda handler: hasattr(handler, 'ha_bound')), reverse=True): diff --git a/qubes/tests/events.py b/qubes/tests/events.py index 82a7ce23..2264b894 100644 --- a/qubes/tests/events.py +++ b/qubes/tests/events.py @@ -165,3 +165,29 @@ class TC_00_Emitter(qubes.tests.QubesTestCase): self.assertCountEqual(effect, ('testvalue1', 'testvalue2', 'testvalue3', 'testvalue4')) + + def test_006_wildcard(self): + # need something mutable + testevent_fired = [0] + + def on_foobar(subject, event, *args, **kwargs): + # pylint: disable=unused-argument + testevent_fired[0] += 1 + + def on_foo(subject, event, *args, **kwargs): + # pylint: disable=unused-argument + testevent_fired[0] += 1 + + emitter = qubes.events.Emitter() + emitter.add_handler('foo:*', on_foo) + emitter.add_handler('foo:bar', on_foobar) + emitter.events_enabled = True + emitter.fire_event('foo:testevent') + self.assertEqual(testevent_fired[0], 1) + emitter.fire_event('foo:bar') + # now foo:bar and foo:* should be executed + self.assertEqual(testevent_fired[0], 3) + emitter.fire_event('foo:') + self.assertEqual(testevent_fired[0], 4) + emitter.fire_event('testevent') + self.assertEqual(testevent_fired[0], 4)