+2005-11-07 Robert McQueen <robot101@debian.org>
+
+ * python/decorators.py: Change emit_signal function to use the
+ signature annotation of the signal when marhsalling the arguments from
+ the service. Fix a bug where the code checking signature length
+ against argument length referenced the wrong variable.
+
+ * python/introspect_parser.py: Avoid adding the type signature of
+ signal arguments to any methods which occur after them in the
+ introspection data (!) by making the parser a little more careful
+ about its current state.
+
+ * python/service.py: Remove debug prints from last commit (again :D).
+
+ * test/python/test-client.py, test/python/test-service.py: Add test
+ signals with signature decorators to test the strict marshalling code
+ gives errors at the right time. Could do with checking the signals
+ actually get emitted too, given that the test does nothing with
+ signals at the moment...
+
2005-11-07 Robert McQueen <robot101@debian.org>
* python/_dbus.py: Add WeakReferenceDictionary cache of dbus.Bus
_util._validate_interface_or_name(dbus_interface)
def decorator(func):
def emit_signal(self, *args, **keywords):
- func(self, *args, **keywords)
+ func(self, *args, **keywords)
message = dbus_bindings.Signal(self._object_path, dbus_interface, func.__name__)
iter = message.get_iter(True)
-
- for arg in args:
- iter.append(arg)
-
+
+ if emit_signal._dbus_signature:
+ signature = tuple(dbus_bindings.Signature(emit_signal._dbus_signature))
+ for (arg, sig) in zip(args, signature):
+ iter.append_strict(arg, sig)
+ else:
+ for arg in args:
+ iter.append(arg)
+
self._connection.send(message)
args = inspect.getargspec(func)[0]
args.pop(0)
if signature:
- sig = tuple(dbus_bindings.Signature(func._dbus_signature))
+ sig = tuple(dbus_bindings.Signature(signature))
if len(sig) > len(args):
raise ValueError, 'signal signature is longer than the number of arguments provided'
reader = input_source.newTextReader("urn:introspect")
ret = reader.Read()
- current_iface=''
- current_method=''
+ current_iface = None
+ current_method = None
current_sigstr = ''
-
+
while ret == 1:
name = reader.LocalName()
if reader.NodeType() == XMLREADER_START_ELEMENT_NODE_TYPE:
- if name == 'interface':
+ if (not current_iface and not current_method and name == 'interface'):
current_iface = reader.GetAttribute('name')
- elif name == 'method':
+ elif (current_iface and not current_method and name == 'method'):
current_method = reader.GetAttribute('name')
if reader.IsEmptyElement():
- method_map[current_iface + '.' + current_method] = ''
- current_method = ''
+ method_map[current_iface + '.' + current_method] = ''
+ current_method = None
current_sigstr = ''
-
- elif name == 'arg':
+
+ elif (current_iface and current_method and name == 'arg'):
direction = reader.GetAttribute('direction')
if not direction or direction == 'in':
current_sigstr = current_sigstr + reader.GetAttribute('type')
elif reader.NodeType() == XMLREADER_END_ELEMENT_NODE_TYPE:
- if name == 'method':
- method_map[current_iface + '.' + current_method] = current_sigstr
- current_method = ''
+ if (current_iface and not current_method and name == 'interface'):
+ current_iface = None
+ if (current_iface and current_method and name == 'method'):
+ method_map[current_iface + '.' + current_method] = current_sigstr
+ current_method = None
current_sigstr = ''
-
ret = reader.Read()
if ret != 0:
# otherwise register the name
retval = dbus_bindings.bus_request_name(bus.get_connection(), name)
- print retval
+
# TODO: more intelligent tracking of bus name states?
if retval == dbus_bindings.REQUEST_NAME_REPLY_PRIMARY_OWNER:
pass
elif retval == dbus_bindings.REQUEST_NAME_REPLY_IN_QUEUE:
# you can't arrive at this state via the high-level bindings
# because you can't put flags in, but... who knows?
- print "joined queue for %s" % name
pass
elif retval == dbus_bindings.REQUEST_NAME_REPLY_EXISTS:
raise dbus_bindings.DBusException('requested name %s already exists' % name)
elif retval == dbus_bindings.REQUEST_NAME_REPLY_ALREADY_OWNER:
# if this is a shared bus which is being used by someone
# else in this process, this can happen legitimately
- print "already owner of %s" % name
pass
else:
raise dbus_bindings.DBusException('requesting name %s returned unexpected value %s' % (name, retval))
main_loop.run()
- def testReturnMarshalling(self):
- print "\n********* Testing return marshalling ***********"
+ def testStrictMarshalling(self):
+ print "\n********* Testing strict return & signal marshalling ***********"
# these values are the same as in the server, and the
# methods should only succeed when they are called with
# with a different number
values = ["", ("",""), ("","",""), [], {}, ["",""], ["","",""]]
methods = [
- (self.iface.ReturnOneString, set([0]), set([0])),
- (self.iface.ReturnTwoStrings, set([1, 5]), set([5])),
- (self.iface.ReturnStruct, set([1, 5]), set([1])),
+ (self.iface.ReturnOneString, 'SignalOneString', set([0]), set([0])),
+ (self.iface.ReturnTwoStrings, 'SignalTwoStrings', set([1, 5]), set([5])),
+ (self.iface.ReturnStruct, 'SignalStruct', set([1, 5]), set([1])),
# all of our test values are sequences so will marshall correctly into an array :P
- (self.iface.ReturnArray, set(range(len(values))), set([3, 5, 6])),
- (self.iface.ReturnDict, set([0, 3, 4]), set([4]))
+ (self.iface.ReturnArray, 'SignalArray', set(range(len(values))), set([3, 5, 6])),
+ (self.iface.ReturnDict, 'SignalDict', set([0, 3, 4]), set([4]))
]
- for (method, success_values, return_values) in methods:
+ for (method, signal, success_values, return_values) in methods:
print "\nTrying correct behaviour of", method._method_name
for value in range(len(values)):
try:
ret = method(value)
except Exception, e:
- print "%s(%s) raised %s" % (method._method_name, repr(values[value]), e.__class__)
+ print "%s(%r) raised %s" % (method._method_name, values[value], e.__class__)
# should fail if it tried to marshal the wrong type
- self.assert_(value not in success_values, "%s should succeed when we ask it to return %s\n%s" % (method._method_name, repr(values[value]), e))
+ self.assert_(value not in success_values, "%s should succeed when we ask it to return %r\n%s\n%s" % (method._method_name, values[value], e.__class__, e))
else:
- print "%s(%s) returned %s" % (method._method_name, repr(values[value]), repr(ret))
+ print "%s(%r) returned %r" % (method._method_name, values[value], ret)
# should only succeed if it's the right return type
- self.assert_(value in success_values, "%s should fail when we ask it to return %s" % (method._method_name, repr(values[value])))
+ self.assert_(value in success_values, "%s should fail when we ask it to return %r" % (method._method_name, values[value]))
# check the value is right too :D
returns = map(lambda n: values[n], return_values)
- self.assert_(ret in returns, "%s should return one of %s" % (method._method_name, repr(returns)))
+ self.assert_(ret in returns, "%s should return one of %r" % (method._method_name, returns))
+
+ print "\nTrying correct emission of", signal
+ for value in range(len(values)):
+ try:
+ self.iface.EmitSignal(signal, value)
+ except Exception, e:
+ print "EmitSignal(%s, %r) raised %s" % (signal, values[value], e.__class__)
+
+ # should fail if it tried to marshal the wrong type
+ self.assert_(value not in success_values, "EmitSignal(%s) should succeed when we ask it to return %r\n%s\n%s" % (signal, values[value], e.__class__, e))
+ else:
+ print "EmitSignal(%s, %r) appeared to succeed" % (signal, values[value])
+
+ # should only succeed if it's the right return type
+ self.assert_(value in success_values, "EmitSignal(%s) should fail when we ask it to return %r" % (signal, values[value]))
+
+ # FIXME: wait for the signal here
+
print
def testInheritance(self):
else:
names[name] = busname
+ print
+
class TestDBusPythonToGLibBindings(unittest.TestCase):
def setUp(self):
self.bus = dbus.SessionBus()
def ReturnDict(self, test):
return self.returnValue(test)
+ @dbus.service.signal("org.freedesktop.DBus.TestSuiteInterface", signature='s')
+ def SignalOneString(self, test):
+ pass
+
+ @dbus.service.signal("org.freedesktop.DBus.TestSuiteInterface", signature='ss')
+ def SignalTwoStrings(self, test, test2):
+ pass
+
+ @dbus.service.signal("org.freedesktop.DBus.TestSuiteInterface", signature='(ss)')
+ def SignalStruct(self, test):
+ pass
+
+ @dbus.service.signal("org.freedesktop.DBus.TestSuiteInterface", signature='as')
+ def SignalArray(self, test):
+ pass
+
+ @dbus.service.signal("org.freedesktop.DBus.TestSuiteInterface", signature='a{ss}')
+ def SignalDict(self, test):
+ pass
+
+ @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='su', out_signature='')
+ def EmitSignal(self, signal, value):
+ sig = getattr(self, signal, None)
+ assert(sig != None)
+
+ val = self.returnValue(value)
+ # make two string case work by passing arguments in by tuple
+ if (signal == 'SignalTwoStrings' and (value == 1 or value == 5)):
+ val = tuple(val)
+ else:
+ val = tuple([val])
+
+ sig(*val)
+
def CheckInheritance(self):
return True
projects / thirdparty / dbus.git / commitdiff
