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='', out_signature='soss',
  74             sender_keyword='sender', path_keyword='path',
  75             destination_keyword='dest', message_keyword='msg')
  76     def MethodExtraInfoKeywords(self, sender=None, path=None, dest=None,
  77             msg=None):
  78         return (sender, path, dest,
  79                 msg.__class__.__module__ + '.' + msg.__class__.__name__)
  80
  81     @dbus.service.method(IFACE, in_signature='ay', out_signature='ay')
  82     def AcceptListOfByte(self, foo):
  83         assert isinstance(foo, list), (foo, foo.__class__.__mro__)
  84         return foo
  85
  86     @dbus.service.method(IFACE, in_signature='ay', out_signature='ay', byte_arrays=True)
  87     def AcceptByteArray(self, foo):
  88         assert isinstance(foo, str), (foo, foo.__class__.__mro__)
  89         return foo
  90
  91     @dbus.service.method(IFACE)
  92     def GetComplexArray(self):
  93         ret = []
  94         for i in range(0,100):
  95             ret.append((random.randint(0,100), random.randint(0,100), str(random.randint(0,100))))
  96
  97         return dbus.Array(ret, signature="(uus)")
  98
  99     def returnValue(self, test):
 100         if test == 0:
 101             return ""
 102         elif test == 1:
 103             return "",""
 104         elif test == 2:
 105             return "","",""
 106         elif test == 3:
 107             return []
 108         elif test == 4:
 109             return {}
 110         elif test == 5:
 111             return ["",""]
 112         elif test == 6:
 113             return ["","",""]
 114
 115     @dbus.service.method(IFACE, in_signature='u', out_signature='s')
 116     def ReturnOneString(self, test):
 117         return self.returnValue(test)
 118
 119     @dbus.service.method(IFACE, in_signature='u', out_signature='ss')
 120     def ReturnTwoStrings(self, test):
 121         return self.returnValue(test)
 122
 123     @dbus.service.method(IFACE, in_signature='u', out_signature='(ss)')
 124     def ReturnStruct(self, test):
 125         logger.info('ReturnStruct(%r) -> %r', test, self.returnValue(test))
 126         return self.returnValue(test)
 127
 128     @dbus.service.method(IFACE, in_signature='u', out_signature='as')
 129     def ReturnArray(self, test):
 130         return self.returnValue(test)
 131
 132     @dbus.service.method(IFACE, in_signature='u', out_signature='a{ss}')
 133     def ReturnDict(self, test):
 134         return self.returnValue(test)
 135
 136     @dbus.service.signal(IFACE, signature='s')
 137     def SignalOneString(self, test):
 138         logger.info('SignalOneString(%r)', test)
 139
 140     @dbus.service.signal(IFACE, signature='ss')
 141     def SignalTwoStrings(self, test, test2):
 142         logger.info('SignalTwoStrings(%r, %r)', test, test2)
 143
 144     @dbus.service.signal(IFACE, signature='(ss)')
 145     def SignalStruct(self, test):
 146         logger.info('SignalStruct(%r)', test)
 147
 148     @dbus.service.signal(IFACE, signature='as')
 149     def SignalArray(self, test):
 150         pass
 151
 152     @dbus.service.signal(IFACE, signature='a{ss}')
 153     def SignalDict(self, test):
 154         pass
 155
 156     @dbus.service.method(IFACE, in_signature='su', out_signature='')
 157     def EmitSignal(self, signal, value):
 158         sig = getattr(self, str(signal), None)
 159         assert(sig != None)
 160
 161         val = self.returnValue(value)
 162         # make two string case work by passing arguments in by tuple
 163         if (signal == 'SignalTwoStrings' and (value == 1 or value == 5)):
 164             val = tuple(val)
 165         else:
 166             val = tuple([val])
 167
 168         logger.info('Emitting %s with %r', signal, val)
 169         sig(*val)
 170
 171     def CheckInheritance(self):
 172         return True
 173
 174     @dbus.service.method(IFACE, in_signature='bbv', out_signature='v', async_callbacks=('return_cb', 'error_cb'))
 175     def AsynchronousMethod(self, async, fail, variant, return_cb, error_cb):
 176         try:
 177             if async:
 178                 gobject.timeout_add(500, self.AsynchronousMethod, False, fail, variant, return_cb, error_cb)
 179                 return
 180             else:
 181                 if fail:
 182                     raise RuntimeError
 183                 else:
 184                     return_cb(variant)
 185
 186                 return False # do not run again
 187         except Exception, e:
 188             error_cb(e)
 189
 190     @dbus.service.method(IFACE, in_signature='', out_signature='s', sender_keyword='sender')
 191     def WhoAmI(self, sender):
 192         return sender
 193
 194     @dbus.service.method(IFACE)
 195     def MultipleReturnWithoutSignature(self):
 196         # https://bugs.freedesktop.org/show_bug.cgi?id=10174
 197         return dbus.String('abc'), dbus.Int32(123)
 198
 199 session_bus = dbus.SessionBus()
 200 name = dbus.service.BusName("org.freedesktop.DBus.TestSuitePythonService", bus=session_bus)
 201 object = TestObject(name)
 202 loop = gobject.MainLoop()
 203 loop.run()