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