test/test-service.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/>
   4 # Copyright (C) 2005, 2006 Collabora Ltd. <http://www.collabora.co.uk/>
   5 #
   6 # Licensed under the Academic Free License version 2.1
   7 #
   8 # This program is free software; you can redistribute it and/or modify
   9 # it under the terms of the GNU General Public License as published by
  10 # the Free Software Foundation; either version 2 of the License, or
  11 # (at your option) any later version.
  12 #
  13 # This program is distributed in the hope that it will be useful,
  14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16 # GNU General Public License for more details.
  17 #
  18 # You should have received a copy of the GNU General Public License
  19 # along with this program; if not, write to the Free Software
  20 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  21
  22 import sys
  23 import os
  24 import logging
  25
  26 builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"])
  27 pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"])
  28
  29 import dbus
  30
  31 if not dbus.__file__.startswith(pydir):
  32     raise Exception("DBus modules are not being picked up from the package")
  33
  34 import dbus.service
  35 import dbus.glib
  36 import gobject
  37 import random
  38
  39
  40 logging.basicConfig(filename=builddir + '/test/test-service.log', filemode='w')
  41 logging.getLogger().setLevel(1)
  42 logger = logging.getLogger('test-service')
  43
  44
  45 IFACE = 'org.freedesktop.DBus.TestSuiteInterface'
  46
  47
  48 class TestInterface(dbus.service.Interface):
  49     @dbus.service.method(IFACE, in_signature='', out_signature='b')
  50     def CheckInheritance(self):
  51         return False
  52
  53 class TestObject(dbus.service.Object, TestInterface):
  54     def __init__(self, bus_name, object_path="/org/freedesktop/DBus/TestSuitePythonObject"):
  55         dbus.service.Object.__init__(self, bus_name, object_path)
  56
  57     """ Echo whatever is sent
  58     """
  59     @dbus.service.method(IFACE)
  60     def Echo(self, arg):
  61         return arg
  62
  63     @dbus.service.method(IFACE, in_signature='s', out_signature='s')
  64     def AcceptUnicodeString(self, foo):
  65         assert isinstance(foo, unicode), (foo, foo.__class__.__mro__)
  66         return foo
  67
  68     @dbus.service.method(IFACE, in_signature='s', out_signature='s', utf8_strings=True)
  69     def AcceptUTF8String(self, foo):
  70         assert isinstance(foo, str), (foo, foo.__class__.__mro__)
  71         return foo
  72
  73     @dbus.service.method(IFACE, in_signature='ay', out_signature='ay')
  74     def AcceptListOfByte(self, foo):
  75         assert isinstance(foo, list), (foo, foo.__class__.__mro__)
  76         return foo
  77
  78     @dbus.service.method(IFACE, in_signature='ay', out_signature='ay', byte_arrays=True)
  79     def AcceptByteArray(self, foo):
  80         assert isinstance(foo, str), (foo, foo.__class__.__mro__)
  81         return foo
  82
  83     @dbus.service.method(IFACE)
  84     def GetComplexArray(self):
  85         ret = []
  86         for i in range(0,100):
  87             ret.append((random.randint(0,100), random.randint(0,100), str(random.randint(0,100))))
  88
  89         return dbus.Array(ret, signature="(uus)")
  90
  91     def returnValue(self, test):
  92         if test == 0:
  93             return ""
  94         elif test == 1:
  95             return "",""
  96         elif test == 2:
  97             return "","",""
  98         elif test == 3:
  99             return []
 100         elif test == 4:
 101             return {}
 102         elif test == 5:
 103             return ["",""]
 104         elif test == 6:
 105             return ["","",""]
 106
 107     @dbus.service.method(IFACE, in_signature='u', out_signature='s')
 108     def ReturnOneString(self, test):
 109         return self.returnValue(test)
 110
 111     @dbus.service.method(IFACE, in_signature='u', out_signature='ss')
 112     def ReturnTwoStrings(self, test):
 113         return self.returnValue(test)
 114
 115     @dbus.service.method(IFACE, in_signature='u', out_signature='(ss)')
 116     def ReturnStruct(self, test):
 117         logger.info('ReturnStruct(%r) -> %r', test, self.returnValue(test))
 118         return self.returnValue(test)
 119
 120     @dbus.service.method(IFACE, in_signature='u', out_signature='as')
 121     def ReturnArray(self, test):
 122         return self.returnValue(test)
 123
 124     @dbus.service.method(IFACE, in_signature='u', out_signature='a{ss}')
 125     def ReturnDict(self, test):
 126         return self.returnValue(test)
 127
 128     @dbus.service.signal(IFACE, signature='s')
 129     def SignalOneString(self, test):
 130         logger.info('SignalOneString(%r)', test)
 131
 132     @dbus.service.signal(IFACE, signature='ss')
 133     def SignalTwoStrings(self, test, test2):
 134         logger.info('SignalTwoStrings(%r, %r)', test, test2)
 135
 136     @dbus.service.signal(IFACE, signature='(ss)')
 137     def SignalStruct(self, test):
 138         logger.info('SignalStruct(%r)', test)
 139
 140     @dbus.service.signal(IFACE, signature='as')
 141     def SignalArray(self, test):
 142         pass
 143
 144     @dbus.service.signal(IFACE, signature='a{ss}')
 145     def SignalDict(self, test):
 146         pass
 147
 148     @dbus.service.method(IFACE, in_signature='su', out_signature='')
 149     def EmitSignal(self, signal, value):
 150         sig = getattr(self, str(signal), None)
 151         assert(sig != None)
 152
 153         val = self.returnValue(value)
 154         # make two string case work by passing arguments in by tuple
 155         if (signal == 'SignalTwoStrings' and (value == 1 or value == 5)):
 156             val = tuple(val)
 157         else:
 158             val = tuple([val])
 159
 160         logger.info('Emitting %s with %r', signal, val)
 161         sig(*val)
 162
 163     def CheckInheritance(self):
 164         return True
 165
 166     @dbus.service.method(IFACE, in_signature='bbv', out_signature='v', async_callbacks=('return_cb', 'error_cb'))
 167     def AsynchronousMethod(self, async, fail, variant, return_cb, error_cb):
 168         try:
 169             if async:
 170                 gobject.timeout_add(500, self.AsynchronousMethod, False, fail, variant, return_cb, error_cb)
 171                 return
 172             else:
 173                 if fail:
 174                     raise RuntimeError
 175                 else:
 176                     return_cb(variant)
 177
 178                 return False # do not run again
 179         except Exception, e:
 180             error_cb(e)
 181
 182     @dbus.service.method(IFACE, in_signature='', out_signature='s', sender_keyword='sender')
 183     def WhoAmI(self, sender):
 184         return sender
 185
 186     @dbus.service.method(IFACE)
 187     def MultipleReturnWithoutSignature(self):
 188         # https://bugs.freedesktop.org/show_bug.cgi?id=10174
 189         return dbus.String('abc'), dbus.Int32(123)
 190
 191 session_bus = dbus.SessionBus()
 192 name = dbus.service.BusName("org.freedesktop.DBus.TestSuitePythonService", bus=session_bus)
 193 object = TestObject(name)
 194 loop = gobject.MainLoop()
 195 loop.run()