glib/gio/tests/gdbus-testserver.py

#!/usr/bin/env python

import gobject
import time

import dbus
import dbus.service
import dbus.mainloop.glib

class TestException(dbus.DBusException):
    _dbus_error_name = 'com.example.TestException'


class TestService(dbus.service.Object):

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                         in_signature='s', out_signature='s')
    def HelloWorld(self, hello_message):
        if str(hello_message) == 'Yo':
            raise TestException('Yo is not a proper greeting')
        else:
            return "You greeted me with '%s'. Thanks!"%(str(hello_message))

    @dbus.service.method("com.example.Frob",
                         in_signature='ss', out_signature='ss')
    def DoubleHelloWorld(self, hello1, hello2):
        return ("You greeted me with '%s'. Thanks!"%(str(hello1)), "Yo dawg, you uttered '%s'. Thanks!"%(str(hello2)))

    @dbus.service.method("com.example.Frob",
                         in_signature='', out_signature='su')
    def PairReturn(self):
        return ("foo", 42)

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                         in_signature='ybnqiuxtdsog', out_signature='ybnqiuxtdsog')
    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):
        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

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                         in_signature='ayabanaqaiauaxatad', out_signature='ayabanaqaiauaxatad')
    def TestArrayOfPrimitiveTypes(self, val_byte, val_boolean, val_int16, val_uint16, val_int32, val_uint32, val_int64, val_uint64, val_double):
        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

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                         in_signature='asaoag', out_signature='asaoag')
    def TestArrayOfStringTypes(self, val_string, val_objpath, val_signature):
        return val_string * 2, val_objpath * 2, val_signature * 2

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                         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}',
                         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}')
    def TestHashTables(self, hyy, hbb, hnn, hqq, hii, huu, hxx, htt, hdd, hss, hoo, hgg):

        ret_hyy = {}
        for i in hyy:
            ret_hyy[i*2] = (hyy[i]*3) & 255

        ret_hbb = {}
        for i in hbb:
            ret_hbb[i] = True

        ret_hnn = {}
        for i in hnn:
            ret_hnn[i*2] = hnn[i]*3

        ret_hqq = {}
        for i in hqq:
            ret_hqq[i*2] = hqq[i]*3

        ret_hii = {}
        for i in hii:
            ret_hii[i*2] = hii[i]*3

        ret_huu = {}
        for i in huu:
            ret_huu[i*2] = huu[i]*3

        ret_hxx = {}
        for i in hxx:
            ret_hxx[i + 2] = hxx[i] + 1

        ret_htt = {}
        for i in htt:
            ret_htt[i + 2] = htt[i] + 1

        ret_hdd = {}
        for i in hdd:
            ret_hdd[i + 2.5] = hdd[i] + 5.0

        ret_hss = {}
        for i in hss:
            ret_hss[i + "mod"] = hss[i]*2

        ret_hoo = {}
        for i in hoo:
            ret_hoo[i + "/mod"] = hoo[i] + "/mod2"

        ret_hgg = {}
        for i in hgg:
            ret_hgg[i + "assgit"] = hgg[i]*2

        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

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                         in_signature='(ii)(s(ii)aya{ss})', out_signature='(ii)(s(ii)aya{ss})')
    def TestStructureTypes(self, s1, s2):
        (x, y) = s1;
        (desc, (x1, y1), ay, hss) = s2;
        ret_hss = {}
        for i in hss:
            ret_hss[i] = hss[i] + " ... in bed!"
        return (x + 1, y + 1), (desc + " ... in bed!", (x1 + 2, y1 + 2), ay * 2, ret_hss)

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                         in_signature='vb', out_signature='v')
    def TestVariant(self, v, modify):

        if modify:
            if type(v)==dbus.Boolean:
                ret = False
            elif type(v)==dbus.Dictionary:
                ret = {}
                for i in v:
                    ret[i] = v[i] * 2
            elif type(v)==dbus.Struct:
                ret = ["other struct", dbus.Int16(100)]
            else:
                ret = v * 2
        else:
            ret = v
        return (type(v))(ret)

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                         in_signature='a(ii)aa(ii)aasaa{ss}aayavaav', out_signature='a(ii)aa(ii)aasaa{ss}aayavaav')
    def TestComplexArrays(self, aii, aaii, aas, ahashes, aay, av, aav):
        return aii * 2, aaii * 2, aas * 2, ahashes * 2, aay * 2, av *2, aav * 2

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                          in_signature='a{s(ii)}a{sv}a{sav}a{saav}a{sa(ii)}a{sa{ss}}',
                         out_signature='a{s(ii)}a{sv}a{sav}a{saav}a{sa(ii)}a{sa{ss}}')
    def TestComplexHashTables(self, h_str_to_pair, h_str_to_variant, h_str_to_av, h_str_to_aav,
                              h_str_to_array_of_pairs, hash_of_hashes):

        ret_h_str_to_pair = {}
        for i in h_str_to_pair:
            ret_h_str_to_pair[i + "_baz"] = h_str_to_pair[i]

        ret_h_str_to_variant = {}
        for i in h_str_to_variant:
            ret_h_str_to_variant[i + "_baz"] = h_str_to_variant[i]

        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

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                          in_signature='', out_signature='')
    def Quit(self):
        mainloop.quit()

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                          in_signature='sv', out_signature='')
    def FrobSetProperty(self, prop_name, prop_value):
        self.frob_props[prop_name] = prop_value
        message = dbus.lowlevel.SignalMessage("/com/example/TestObject",
                                              "org.freedesktop.DBus.Properties",
                                              "PropertiesChanged")
        message.append("com.example.Frob")
        message.append({prop_name : prop_value})
        message.append([], signature="as")
        session_bus.send_message(message)

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob",
                          in_signature='', out_signature='')
    def FrobInvalidateProperty(self):
        self.frob_props["PropertyThatWillBeInvalidated"] = "OMGInvalidated"
        message = dbus.lowlevel.SignalMessage("/com/example/TestObject",
                                              "org.freedesktop.DBus.Properties",
                                              "PropertiesChanged")
        message.append("com.example.Frob")
        message.append({}, signature="a{sv}")
        message.append(["PropertyThatWillBeInvalidated"])
        session_bus.send_message(message)

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.signal("com.example.Frob",
                         signature="sov")
    def TestSignal(self, str1, objpath1, variant1):
        pass

    @dbus.service.method("com.example.Frob",
                          in_signature='so', out_signature='')
    def EmitSignal(self, str1, objpath1):
        self.TestSignal (str1 + " .. in bed!", objpath1 + "/in/bed", "a variant")

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("com.example.Frob", in_signature='i', out_signature='',
                         async_callbacks=('return_cb', 'raise_cb'))
    def Sleep(self, msec, return_cb, raise_cb):
        def return_from_async_wait():
            return_cb()
            return False
        gobject.timeout_add(msec, return_from_async_wait)

    # ----------------------------------------------------------------------------------------------------

    @dbus.service.method("org.freedesktop.DBus.Properties",
                         in_signature  = 'ss',
                         out_signature = 'v')
    def Get(self, interface_name, property_name):

        if interface_name == "com.example.Frob":
            return self.frob_props[property_name]
        else:
            raise TestException("No such interface " + interface_name)

    @dbus.service.method("org.freedesktop.DBus.Properties",
                         in_signature  = 's',
                         out_signature = 'a{sv}')
    def GetAll(self, interface_name):
        if interface_name == "com.example.Frob":
            return self.frob_props
        else:
            raise TestException("No such interface " + interface_name)

if __name__ == '__main__':
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    session_bus = dbus.SessionBus()
    name = dbus.service.BusName("com.example.TestService", session_bus)

    obj = TestService(session_bus, '/com/example/TestObject')

    #print "Our unique name is %s"%(session_bus.get_unique_name())

    obj.frob_props = {}
    obj.frob_props["y"] = dbus.Byte(1)
    obj.frob_props["b"] = dbus.Boolean(True)
    obj.frob_props["n"] = dbus.Int16(2)
    obj.frob_props["q"] = dbus.UInt16(3)
    obj.frob_props["i"] = dbus.Int32(4)
    obj.frob_props["u"] = dbus.UInt32(5)
    obj.frob_props["x"] = dbus.Int64(6)
    obj.frob_props["t"] = dbus.UInt64(7)
    obj.frob_props["d"] = dbus.Double(7.5)
    obj.frob_props["s"] = dbus.String("a string")
    obj.frob_props["o"] = dbus.ObjectPath("/some/path")
    obj.frob_props["ay"] = [dbus.Byte(1), dbus.Byte(11)]
    obj.frob_props["ab"] = [dbus.Boolean(True), dbus.Boolean(False)]
    obj.frob_props["an"] = [dbus.Int16(2), dbus.Int16(12)]
    obj.frob_props["aq"] = [dbus.UInt16(3), dbus.UInt16(13)]
    obj.frob_props["ai"] = [dbus.Int32(4), dbus.Int32(14)]
    obj.frob_props["au"] = [dbus.UInt32(5), dbus.UInt32(15)]
    obj.frob_props["ax"] = [dbus.Int64(6), dbus.Int64(16)]
    obj.frob_props["at"] = [dbus.UInt64(7), dbus.UInt64(17)]
    obj.frob_props["ad"] = [dbus.Double(7.5), dbus.Double(17.5)]
    obj.frob_props["as"] = [dbus.String("a string"), dbus.String("another string")]
    obj.frob_props["ao"] = [dbus.ObjectPath("/some/path"), dbus.ObjectPath("/another/path")]
    obj.frob_props["foo"] = "a frobbed string"
    obj.frob_props["PropertyThatWillBeInvalidated"] = "InitialValue"

    mainloop = gobject.MainLoop()
    mainloop.run()