gio/tests/gdbus-testserver.py

   1 #!/usr/bin/env python
   2
   3 import gobject
   4 import time
   5
   6 import dbus
   7 import dbus.service
   8 import dbus.mainloop.glib
   9
  10 class TestException(dbus.DBusException):
  11     _dbus_error_name = 'com.example.TestException'
  12
  13
  14 class TestService(dbus.service.Object):
  15
  16     # ----------------------------------------------------------------------------------------------------
  17
  18     @dbus.service.method("com.example.Frob",
  19                          in_signature='s', out_signature='s')
  20     def HelloWorld(self, hello_message):
  21         if str(hello_message) == 'Yo':
  22             raise TestException('Yo is not a proper greeting')
  23         else:
  24             return "You greeted me with '%s'. Thanks!"%(str(hello_message))
  25
  26     @dbus.service.method("com.example.Frob",
  27                          in_signature='ss', out_signature='ss')
  28     def DoubleHelloWorld(self, hello1, hello2):
  29         return ("You greeted me with '%s'. Thanks!"%(str(hello1)), "Yo dawg, you uttered '%s'. Thanks!"%(str(hello2)))
  30
  31     @dbus.service.method("com.example.Frob",
  32                          in_signature='', out_signature='su')
  33     def PairReturn(self):
  34         return ("foo", 42)
  35
  36     # ----------------------------------------------------------------------------------------------------
  37
  38     @dbus.service.method("com.example.Frob",
  39                          in_signature='ybnqiuxtdsog', out_signature='ybnqiuxtdsog')
  40     def TestPrimitiveTypes(self, val_byte, val_boolean, val_int16, val_uint16, val_int32, val_uint32, val_int64, val_uint64, val_double, val_string, val_objpath, val_signature):
  41         return val_byte + 1, not val_boolean, val_int16 + 1, val_uint16 + 1, val_int32 + 1, val_uint32 + 1, val_int64 + 1, val_uint64 + 1, -val_double + 0.123, val_string * 2, val_objpath + "/modified", val_signature * 2
  42
  43     # ----------------------------------------------------------------------------------------------------
  44
  45     @dbus.service.method("com.example.Frob",
  46                          in_signature='ayabanaqaiauaxatad', out_signature='ayabanaqaiauaxatad')
  47     def TestArrayOfPrimitiveTypes(self, val_byte, val_boolean, val_int16, val_uint16, val_int32, val_uint32, val_int64, val_uint64, val_double):
  48         return val_byte*2, val_boolean*2, val_int16*2, val_uint16*2, val_int32*2, val_uint32*2, val_int64*2, val_uint64*2, val_double*2
  49
  50     # ----------------------------------------------------------------------------------------------------
  51
  52     @dbus.service.method("com.example.Frob",
  53                          in_signature='asaoag', out_signature='asaoag')
  54     def TestArrayOfStringTypes(self, val_string, val_objpath, val_signature):
  55         return val_string * 2, val_objpath * 2, val_signature * 2
  56
  57     # ----------------------------------------------------------------------------------------------------
  58
  59     @dbus.service.method("com.example.Frob",
  60                          in_signature  = 'a{yy}a{bb}a{nn}a{qq}a{ii}a{uu}a{xx}a{tt}a{dd}a{ss}a{oo}a{gg}',
  61                          out_signature = 'a{yy}a{bb}a{nn}a{qq}a{ii}a{uu}a{xx}a{tt}a{dd}a{ss}a{oo}a{gg}')
  62     def TestHashTables(self, hyy, hbb, hnn, hqq, hii, huu, hxx, htt, hdd, hss, hoo, hgg):
  63
  64         ret_hyy = {}
  65         for i in hyy:
  66             ret_hyy[i*2] = (hyy[i]*3) & 255
  67
  68         ret_hbb = {}
  69         for i in hbb:
  70             ret_hbb[i] = True
  71
  72         ret_hnn = {}
  73         for i in hnn:
  74             ret_hnn[i*2] = hnn[i]*3
  75
  76         ret_hqq = {}
  77         for i in hqq:
  78             ret_hqq[i*2] = hqq[i]*3
  79
  80         ret_hii = {}
  81         for i in hii:
  82             ret_hii[i*2] = hii[i]*3
  83
  84         ret_huu = {}
  85         for i in huu:
  86             ret_huu[i*2] = huu[i]*3
  87
  88         ret_hxx = {}
  89         for i in hxx:
  90             ret_hxx[i + 2] = hxx[i] + 1
  91
  92         ret_htt = {}
  93         for i in htt:
  94             ret_htt[i + 2] = htt[i] + 1
  95
  96         ret_hdd = {}
  97         for i in hdd:
  98             ret_hdd[i + 2.5] = hdd[i] + 5.0
  99
 100         ret_hss = {}
 101         for i in hss:
 102             ret_hss[i + "mod"] = hss[i]*2
 103
 104         ret_hoo = {}
 105         for i in hoo:
 106             ret_hoo[i + "/mod"] = hoo[i] + "/mod2"
 107
 108         ret_hgg = {}
 109         for i in hgg:
 110             ret_hgg[i + "assgit"] = hgg[i]*2
 111
 112         return ret_hyy, ret_hbb, ret_hnn, ret_hqq, ret_hii, ret_huu, ret_hxx, ret_htt, ret_hdd, ret_hss, ret_hoo, ret_hgg
 113
 114     # ----------------------------------------------------------------------------------------------------
 115
 116     @dbus.service.method("com.example.Frob",
 117                          in_signature='(ii)(s(ii)aya{ss})', out_signature='(ii)(s(ii)aya{ss})')
 118     def TestStructureTypes(self, s1, s2):
 119         (x, y) = s1;
 120         (desc, (x1, y1), ay, hss) = s2;
 121         ret_hss = {}
 122         for i in hss:
 123             ret_hss[i] = hss[i] + " ... in bed!"
 124         return (x + 1, y + 1), (desc + " ... in bed!", (x1 + 2, y1 + 2), ay * 2, ret_hss)
 125
 126     # ----------------------------------------------------------------------------------------------------
 127
 128     @dbus.service.method("com.example.Frob",
 129                          in_signature='vb', out_signature='v')
 130     def TestVariant(self, v, modify):
 131
 132         if modify:
 133             if type(v)==dbus.Boolean:
 134                 ret = False
 135             elif type(v)==dbus.Dictionary:
 136                 ret = {}
 137                 for i in v:
 138                     ret[i] = v[i] * 2
 139             elif type(v)==dbus.Struct:
 140                 ret = ["other struct", dbus.Int16(100)]
 141             else:
 142                 ret = v * 2
 143         else:
 144             ret = v
 145         return (type(v))(ret)
 146
 147     # ----------------------------------------------------------------------------------------------------
 148
 149     @dbus.service.method("com.example.Frob",
 150                          in_signature='a(ii)aa(ii)aasaa{ss}aayavaav', out_signature='a(ii)aa(ii)aasaa{ss}aayavaav')
 151     def TestComplexArrays(self, aii, aaii, aas, ahashes, aay, av, aav):
 152         return aii * 2, aaii * 2, aas * 2, ahashes * 2, aay * 2, av *2, aav * 2
 153
 154     # ----------------------------------------------------------------------------------------------------
 155
 156     @dbus.service.method("com.example.Frob",
 157                           in_signature='a{s(ii)}a{sv}a{sav}a{saav}a{sa(ii)}a{sa{ss}}',
 158                          out_signature='a{s(ii)}a{sv}a{sav}a{saav}a{sa(ii)}a{sa{ss}}')
 159     def TestComplexHashTables(self, h_str_to_pair, h_str_to_variant, h_str_to_av, h_str_to_aav,
 160                               h_str_to_array_of_pairs, hash_of_hashes):
 161
 162         ret_h_str_to_pair = {}
 163         for i in h_str_to_pair:
 164             ret_h_str_to_pair[i + "_baz"] = h_str_to_pair[i]
 165
 166         ret_h_str_to_variant = {}
 167         for i in h_str_to_variant:
 168             ret_h_str_to_variant[i + "_baz"] = h_str_to_variant[i]
 169
 170         return ret_h_str_to_pair, ret_h_str_to_variant, h_str_to_av, h_str_to_aav, h_str_to_array_of_pairs, hash_of_hashes
 171
 172     # ----------------------------------------------------------------------------------------------------
 173
 174     @dbus.service.method("com.example.Frob",
 175                           in_signature='', out_signature='')
 176     def Quit(self):
 177         mainloop.quit()
 178
 179     # ----------------------------------------------------------------------------------------------------
 180
 181     @dbus.service.method("com.example.Frob",
 182                           in_signature='sv', out_signature='')
 183     def FrobSetProperty(self, prop_name, prop_value):
 184         self.frob_props[prop_name] = prop_value
 185         message = dbus.lowlevel.SignalMessage("/com/example/TestObject",
 186                                               "org.freedesktop.DBus.Properties",
 187                                               "PropertiesChanged")
 188         message.append("com.example.Frob")
 189         message.append({prop_name : prop_value})
 190         message.append([], signature="as")
 191         session_bus.send_message(message)
 192
 193     # ----------------------------------------------------------------------------------------------------
 194
 195     @dbus.service.method("com.example.Frob",
 196                           in_signature='s', out_signature='')
 197     def FrobInvalidateProperty(self, new_value):
 198         self.frob_props["PropertyThatWillBeInvalidated"] = new_value
 199         message = dbus.lowlevel.SignalMessage("/com/example/TestObject",
 200                                               "org.freedesktop.DBus.Properties",
 201                                               "PropertiesChanged")
 202         message.append("com.example.Frob")
 203         message.append({}, signature="a{sv}")
 204         message.append(["PropertyThatWillBeInvalidated"])
 205         session_bus.send_message(message)
 206
 207     # ----------------------------------------------------------------------------------------------------
 208
 209     @dbus.service.signal("com.example.Frob",
 210                          signature="sov")
 211     def TestSignal(self, str1, objpath1, variant1):
 212         pass
 213
 214     @dbus.service.method("com.example.Frob",
 215                           in_signature='so', out_signature='')
 216     def EmitSignal(self, str1, objpath1):
 217         self.TestSignal (str1 + " .. in bed!", objpath1 + "/in/bed", "a variant")
 218
 219     # ----------------------------------------------------------------------------------------------------
 220
 221     @dbus.service.signal("com.example.Frob",
 222                          signature="i")
 223     def TestSignal2(self, int1):
 224         pass
 225
 226     @dbus.service.method("com.example.Frob",
 227                           in_signature='', out_signature='')
 228     def EmitSignal2(self):
 229         self.TestSignal2 (42)
 230
 231     # ----------------------------------------------------------------------------------------------------
 232
 233     @dbus.service.method("com.example.Frob", in_signature='i', out_signature='',
 234                          async_callbacks=('return_cb', 'raise_cb'))
 235     def Sleep(self, msec, return_cb, raise_cb):
 236         def return_from_async_wait():
 237             return_cb()
 238             return False
 239         gobject.timeout_add(msec, return_from_async_wait)
 240
 241     # ----------------------------------------------------------------------------------------------------
 242
 243     @dbus.service.method("org.freedesktop.DBus.Properties",
 244                          in_signature  = 'ss',
 245                          out_signature = 'v')
 246     def Get(self, interface_name, property_name):
 247
 248         if interface_name == "com.example.Frob":
 249             return self.frob_props[property_name]
 250         else:
 251             raise TestException("No such interface " + interface_name)
 252
 253     @dbus.service.method("org.freedesktop.DBus.Properties",
 254                          in_signature  = 's',
 255                          out_signature = 'a{sv}')
 256     def GetAll(self, interface_name):
 257         if interface_name == "com.example.Frob":
 258             return self.frob_props
 259         else:
 260             raise TestException("No such interface " + interface_name)
 261
 262 if __name__ == '__main__':
 263     dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
 264
 265     session_bus = dbus.SessionBus()
 266     name = dbus.service.BusName("com.example.TestService", session_bus)
 267
 268     obj = TestService(session_bus, '/com/example/TestObject')
 269
 270     #print "Our unique name is %s"%(session_bus.get_unique_name())
 271
 272     obj.frob_props = {}
 273     obj.frob_props["y"] = dbus.Byte(1)
 274     obj.frob_props["b"] = dbus.Boolean(True)
 275     obj.frob_props["n"] = dbus.Int16(2)
 276     obj.frob_props["q"] = dbus.UInt16(3)
 277     obj.frob_props["i"] = dbus.Int32(4)
 278     obj.frob_props["u"] = dbus.UInt32(5)
 279     obj.frob_props["x"] = dbus.Int64(6)
 280     obj.frob_props["t"] = dbus.UInt64(7)
 281     obj.frob_props["d"] = dbus.Double(7.5)
 282     obj.frob_props["s"] = dbus.String("a string")
 283     obj.frob_props["o"] = dbus.ObjectPath("/some/path")
 284     obj.frob_props["ay"] = [dbus.Byte(1), dbus.Byte(11)]
 285     obj.frob_props["ab"] = [dbus.Boolean(True), dbus.Boolean(False)]
 286     obj.frob_props["an"] = [dbus.Int16(2), dbus.Int16(12)]
 287     obj.frob_props["aq"] = [dbus.UInt16(3), dbus.UInt16(13)]
 288     obj.frob_props["ai"] = [dbus.Int32(4), dbus.Int32(14)]
 289     obj.frob_props["au"] = [dbus.UInt32(5), dbus.UInt32(15)]
 290     obj.frob_props["ax"] = [dbus.Int64(6), dbus.Int64(16)]
 291     obj.frob_props["at"] = [dbus.UInt64(7), dbus.UInt64(17)]
 292     obj.frob_props["ad"] = [dbus.Double(7.5), dbus.Double(17.5)]
 293     obj.frob_props["as"] = [dbus.String("a string"), dbus.String("another string")]
 294     obj.frob_props["ao"] = [dbus.ObjectPath("/some/path"), dbus.ObjectPath("/another/path")]
 295     obj.frob_props["foo"] = "a frobbed string"
 296     obj.frob_props["PropertyThatWillBeInvalidated"] = "InitialValue"
 297
 298     mainloop = gobject.MainLoop()
 299     mainloop.run()