Add an example of asynchronous calls. Run the examples during 'make check'.
[dbus-python-phuang.git] / test / test-service.py
blobf62e2fb26c8f519761cd77d7b822ccbdbc342dd5
1 #!/usr/bin/env python
3 # Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/>
4 # Copyright (C) 2005, 2006 Collabora Ltd. <http://www.collabora.co.uk/>
6 # Licensed under the Academic Free License version 2.1
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.
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.
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
24 import sys
25 import os
26 import logging
28 builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"])
29 pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"])
31 import dbus
33 if not dbus.__file__.startswith(pydir):
34 raise Exception("DBus modules are not being picked up from the package")
36 import dbus.service
37 import dbus.glib
38 import gobject
39 import random
42 logging.basicConfig(filename=builddir + '/test/test-service.log', filemode='w')
43 logging.getLogger().setLevel(1)
44 logger = logging.getLogger('test-service')
47 class TestInterface(dbus.service.Interface):
48 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='', out_signature='b')
49 def CheckInheritance(self):
50 return False
52 class TestObject(dbus.service.Object, TestInterface):
53 def __init__(self, bus_name, object_path="/org/freedesktop/DBus/TestSuitePythonObject"):
54 dbus.service.Object.__init__(self, bus_name, object_path)
56 """ Echo whatever is sent
57 """
58 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface")
59 def Echo(self, arg):
60 return arg
62 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='s', out_signature='s')
63 def AcceptUnicodeString(self, foo):
64 assert isinstance(foo, unicode), (foo, foo.__class__.__mro__)
65 return foo
67 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='s', out_signature='s', utf8_strings=True)
68 def AcceptUTF8String(self, foo):
69 assert isinstance(foo, str), (foo, foo.__class__.__mro__)
70 return foo
72 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='ay', out_signature='ay')
73 def AcceptListOfByte(self, foo):
74 assert isinstance(foo, list), (foo, foo.__class__.__mro__)
75 return foo
77 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='ay', out_signature='ay', byte_arrays=True)
78 def AcceptByteArray(self, foo):
79 assert isinstance(foo, str), (foo, foo.__class__.__mro__)
80 return foo
82 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface")
83 def GetComplexArray(self):
84 ret = []
85 for i in range(0,100):
86 ret.append((random.randint(0,100), random.randint(0,100), str(random.randint(0,100))))
88 return dbus.Array(ret, signature="(uus)")
90 def returnValue(self, test):
91 if test == 0:
92 return ""
93 elif test == 1:
94 return "",""
95 elif test == 2:
96 return "","",""
97 elif test == 3:
98 return []
99 elif test == 4:
100 return {}
101 elif test == 5:
102 return ["",""]
103 elif test == 6:
104 return ["","",""]
106 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='u', out_signature='s')
107 def ReturnOneString(self, test):
108 return self.returnValue(test)
110 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='u', out_signature='ss')
111 def ReturnTwoStrings(self, test):
112 return self.returnValue(test)
114 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='u', out_signature='(ss)')
115 def ReturnStruct(self, test):
116 logger.info('ReturnStruct(%r) -> %r', test, self.returnValue(test))
117 return self.returnValue(test)
119 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='u', out_signature='as')
120 def ReturnArray(self, test):
121 return self.returnValue(test)
123 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='u', out_signature='a{ss}')
124 def ReturnDict(self, test):
125 return self.returnValue(test)
127 @dbus.service.signal("org.freedesktop.DBus.TestSuiteInterface", signature='s')
128 def SignalOneString(self, test):
129 logger.info('SignalOneString(%r)', test)
131 @dbus.service.signal("org.freedesktop.DBus.TestSuiteInterface", signature='ss')
132 def SignalTwoStrings(self, test, test2):
133 logger.info('SignalTwoStrings(%r, %r)', test, test2)
135 @dbus.service.signal("org.freedesktop.DBus.TestSuiteInterface", signature='(ss)')
136 def SignalStruct(self, test):
137 logger.info('SignalStruct(%r)', test)
139 @dbus.service.signal("org.freedesktop.DBus.TestSuiteInterface", signature='as')
140 def SignalArray(self, test):
141 pass
143 @dbus.service.signal("org.freedesktop.DBus.TestSuiteInterface", signature='a{ss}')
144 def SignalDict(self, test):
145 pass
147 @dbus.service.method("org.freedesktop.DBus.TestSuiteInterface", in_signature='su', out_signature='')
148 def EmitSignal(self, signal, value):
149 sig = getattr(self, str(signal), None)
150 assert(sig != None)
152 val = self.returnValue(value)
153 # make two string case work by passing arguments in by tuple
154 if (signal == 'SignalTwoStrings' and (value == 1 or value == 5)):
155 val = tuple(val)
156 else:
157 val = tuple([val])
159 logger.info('Emitting %s with %r', signal, val)
160 sig(*val)
162 def CheckInheritance(self):
163 return True
165 @dbus.service.method('org.freedesktop.DBus.TestSuiteInterface', in_signature='bbv', out_signature='v', async_callbacks=('return_cb', 'error_cb'))
166 def AsynchronousMethod(self, async, fail, variant, return_cb, error_cb):
167 try:
168 if async:
169 gobject.timeout_add(500, self.AsynchronousMethod, False, fail, variant, return_cb, error_cb)
170 return
171 else:
172 if fail:
173 raise RuntimeError
174 else:
175 return_cb(variant)
177 return False # do not run again
178 except Exception, e:
179 error_cb(e)
181 @dbus.service.method('org.freedesktop.DBus.TestSuiteInterface', in_signature='', out_signature='s', sender_keyword='sender')
182 def WhoAmI(self, sender):
183 return sender
185 session_bus = dbus.SessionBus()
186 name = dbus.service.BusName("org.freedesktop.DBus.TestSuitePythonService", bus=session_bus)
187 object = TestObject(name)
188 loop = gobject.MainLoop()
189 loop.run()