diff options
| -rw-r--r-- | dbus/connection.py | 52 | ||||
| -rw-r--r-- | dbus/proxies.py | 10 | ||||
| -rwxr-xr-x | test/test-standalone.py | 62 |
3 files changed, 118 insertions, 6 deletions
diff --git a/dbus/connection.py b/dbus/connection.py index 621e54b..2bada4b 100644 --- a/dbus/connection.py +++ b/dbus/connection.py @@ -26,6 +26,7 @@ __all__ = ('Connection', 'SignalMatch') __docformat__ = 'reStructuredText' import logging +import re import threading import weakref @@ -39,7 +40,7 @@ from dbus.lowlevel import ( from dbus.proxies import ProxyObject from dbus._compat import is_py2, is_py3 -from _dbus_bindings import String +from _dbus_bindings import (ObjectPath, String) _logger = logging.getLogger('dbus.connection') @@ -56,7 +57,7 @@ class SignalMatch(object): '_destination_keyword', '_interface_keyword', '_message_keyword', '_member_keyword', '_sender_keyword', '_path_keyword', '_int_args_match', - '_arg0namespace' + '_arg0namespace', '_int_args_paths' ] __slots__ = tuple(_slots) @@ -103,21 +104,30 @@ class SignalMatch(object): self._args_match = kwargs if not kwargs: self._int_args_match = None + self._int_args_paths = None else: self._int_args_match = {} + self._int_args_paths = {} for kwarg in kwargs: - if not kwarg.startswith('arg'): + match = re.match("arg(\d+)(\w+)?$", kwarg) + if not match: raise TypeError('SignalMatch: unknown keyword argument %s' % kwarg) try: - index = int(kwarg[3:]) + index = int(match[1]) except ValueError: raise TypeError('SignalMatch: unknown keyword argument %s' % kwarg) if index < 0 or index > 63: raise TypeError('SignalMatch: arg match index must be in ' 'range(64), not %d' % index) - self._int_args_match[index] = kwargs[kwarg] + if match[2] is None: + self._int_args_match[index] = kwargs[kwarg] + elif match[2] == "path": + self._int_args_paths[index] = kwargs[kwarg] + else: + raise TypeError('SignalMatch: unknown keyword argument %s' + % kwarg) def __hash__(self): """SignalMatch objects are compared by identity.""" @@ -149,6 +159,9 @@ class SignalMatch(object): if self._int_args_match is not None: for index, value in self._int_args_match.items(): rule.append("arg%d='%s'" % (index, value)) + if self._int_args_paths is not None: + for index, value in self._int_args_paths.items(): + rule.append("arg%dpath='%s'" % (index, value)) self._rule = ','.join(rule) @@ -183,7 +196,7 @@ class SignalMatch(object): # these haven't been checked yet by the match tree if self._sender_name_owner not in (None, message.get_sender()): return False - if self._int_args_match is not None: + if self._int_args_match is not None or self._int_args_paths is not None: # extracting args with byte_arrays is less work kwargs = dict(byte_arrays=True) args = message.get_args_list(**kwargs) @@ -192,6 +205,23 @@ class SignalMatch(object): or not isinstance(args[index], String) or args[index] != value): return False + + if self._int_args_paths is not None: + for index, value in self._int_args_paths.items(): + if index >= len(args): + return False + + arg = args[index] + if not isinstance(arg, String) and not isinstance(arg, ObjectPath): + return False + + if not ( + value == arg or + value[-1:] == "/" and arg.startswith(value) or + arg[-1:] == "/" and value.startswith(arg) + ): + return False + if self._arg0namespace is not None: kwargs = dict(byte_arrays=True) args = message.get_args_list(**kwargs) @@ -402,6 +432,16 @@ class Connection(_Connection): argument is a string that either is equal to the keyword parameter, or starts with the keyword parameter followed by a dot (and optionally more text). + `arg...path`: str + If there are additional keyword parameters of the form + ``arg``\\ *n* ``path``, match only signals where the *n*\\ th + argument is either equal or matches in a path like manner. + A path-like comparison matches when either the keyword or the + argument ends with a '/' and is a prefix of the other. An + example argument path match is arg0path='/aa/bb/'. This would + match messages with first arguments of '/', '/aa/', '/aa/bb/', + '/aa/bb/cc/' and '/aa/bb/cc'. It would not match messages with + first arguments of '/aa/b', '/aa' or even '/aa/bb'. `named_service` : str A deprecated alias for `bus_name`. """ diff --git a/dbus/proxies.py b/dbus/proxies.py index a31b5ea..523399e 100644 --- a/dbus/proxies.py +++ b/dbus/proxies.py @@ -366,6 +366,16 @@ class ProxyObject(object): argument is a string that either is equal to the keyword parameter, or starts with the keyword parameter followed by a dot (and optionally more text). + `arg...path`: str + If there are additional keyword parameters of the form + ``arg``\\ *n* ``path``, match only signals where the *n*\\ th + argument is either equal or matches in a path-like manner. + A path-like comparison matches when either the keyword or the + argument ends with a '/' and is a prefix of the other. An + example argument path match is arg0path='/aa/bb/'. This would + match messages with first arguments of '/', '/aa/', '/aa/bb/', + '/aa/bb/cc/' and '/aa/bb/cc'. It would not match messages with + first arguments of '/aa/b', '/aa' or even '/aa/bb'. """ return \ self._bus.add_signal_receiver(handler_function, diff --git a/test/test-standalone.py b/test/test-standalone.py index 7cca6a0..c517a69 100755 --- a/test/test-standalone.py +++ b/test/test-standalone.py @@ -617,6 +617,68 @@ class TestArg0Namespace(unittest.TestCase): message.append(1, signature='i') self.assertFalse(match.maybe_handle_message(message)) +class TestPathMatches(unittest.TestCase): + def setUp(self): + self._match = dbus.connection.SignalMatch(object, None, '/', None, None, + None, arg0path='/aa/bb/') + + def test_empty_path(self): + match = dbus.connection.SignalMatch(object, None, '/', None, None, + None, arg0path='') + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('/', signature='s') + self.assertFalse(match.maybe_handle_message(message)) + + def test_empty_value(self): + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('', signature='s') + self.assertFalse(self._match.maybe_handle_message(message)) + + def test_root(self): + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('/', signature='s') + self.assertTrue(self._match.maybe_handle_message(message)) + + def test_aa(self): + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('/aa/', signature='s') + self.assertTrue(self._match.maybe_handle_message(message)) + + def test_aa_bb(self): + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('/aa/bb/', signature='s') + self.assertTrue(self._match.maybe_handle_message(message)) + + def test_aa_bb_cc(self): + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('/aa/bb/cc/', signature='s') + self.assertTrue(self._match.maybe_handle_message(message)) + + def test_aa_bb_cc2(self): + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('/aa/bb/cc', signature='s') + self.assertTrue(self._match.maybe_handle_message(message)) + + def test_no_match_aa_b(self): + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('/aa/b', signature='s') + self.assertFalse(self._match.maybe_handle_message(message)) + + def test_no_match_aa(self): + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('/aa', signature='s') + self.assertFalse(self._match.maybe_handle_message(message)) + + def test_no_match_aa_bb(self): + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('/aa/bb', signature='s') + self.assertFalse(self._match.maybe_handle_message(message)) + + def test_object_path(self): + message = _dbus_bindings.SignalMessage('/', 'a.b', 'c') + message.append('/aa/bb/cc', signature='o') + self.assertTrue(self._match.maybe_handle_message(message)) + class TestVersion(unittest.TestCase): if sys.version_info[:2] < (2, 7): def assertGreater(self, first, second): |