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 # the Free Software Foundation; either version 2 of the License, or
11 # (at your option) any later version.
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.
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
26 builddir
= os
.path
.normpath(os
.environ
["DBUS_TOP_BUILDDIR"])
27 pydir
= os
.path
.normpath(os
.environ
["DBUS_TOP_SRCDIR"])
31 if not dbus
.__file
__.startswith(pydir
):
32 raise Exception("DBus modules are not being picked up from the package")
39 from dbus
.gobject_service
import ExportedGObject
42 logging
.basicConfig(filename
=builddir
+ '/test/test-service.log', filemode
='w')
43 logging
.getLogger().setLevel(1)
44 logger
= logging
.getLogger('test-service')
47 NAME
= "org.freedesktop.DBus.TestSuitePythonService"
48 IFACE
= "org.freedesktop.DBus.TestSuiteInterface"
49 OBJECT
= "/org/freedesktop/DBus/TestSuitePythonObject"
51 class RemovableObject(dbus
.service
.Object
):
52 # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
53 def __init__(self
, bus_name
, object_path
=OBJECT
+ '/RemovableObject'):
54 super(RemovableObject
, self
).__init
__(bus_name
, object_path
)
56 @dbus.service
.method(IFACE
, in_signature
='', out_signature
='b')
60 @dbus.service
.method(IFACE
, in_signature
='', out_signature
='b')
62 self
.remove_from_connection()
65 class TestGObject(ExportedGObject
):
66 def __init__(self
, bus_name
, object_path
=OBJECT
+ '/GObject'):
67 super(TestGObject
, self
).__init
__(bus_name
, object_path
)
69 @dbus.service
.method(IFACE
)
73 class TestInterface(dbus
.service
.Interface
):
74 @dbus.service
.method(IFACE
, in_signature
='', out_signature
='b')
75 def CheckInheritance(self
):
78 class TestObject(dbus
.service
.Object
, TestInterface
):
79 def __init__(self
, bus_name
, object_path
=OBJECT
):
80 dbus
.service
.Object
.__init
__(self
, bus_name
, object_path
)
83 """ Echo whatever is sent
85 @dbus.service
.method(IFACE
)
89 @dbus.service
.method(IFACE
, in_signature
='s', out_signature
='s')
90 def AcceptUnicodeString(self
, foo
):
91 assert isinstance(foo
, unicode), (foo
, foo
.__class
__.__mro
__)
94 @dbus.service
.method(IFACE
, in_signature
='s', out_signature
='s', utf8_strings
=True)
95 def AcceptUTF8String(self
, foo
):
96 assert isinstance(foo
, str), (foo
, foo
.__class
__.__mro
__)
99 @dbus.service
.method(IFACE
, in_signature
='', out_signature
='soss',
100 sender_keyword
='sender', path_keyword
='path',
101 destination_keyword
='dest', message_keyword
='msg')
102 def MethodExtraInfoKeywords(self
, sender
=None, path
=None, dest
=None,
104 return (sender
, path
, dest
,
105 msg
.__class
__.__module
__ + '.' + msg
.__class
__.__name
__)
107 @dbus.service
.method(IFACE
, in_signature
='ay', out_signature
='ay')
108 def AcceptListOfByte(self
, foo
):
109 assert isinstance(foo
, list), (foo
, foo
.__class
__.__mro
__)
112 @dbus.service
.method(IFACE
, in_signature
='ay', out_signature
='ay', byte_arrays
=True)
113 def AcceptByteArray(self
, foo
):
114 assert isinstance(foo
, str), (foo
, foo
.__class
__.__mro
__)
117 @dbus.service
.method(IFACE
)
118 def GetComplexArray(self
):
120 for i
in range(0,100):
121 ret
.append((random
.randint(0,100), random
.randint(0,100), str(random
.randint(0,100))))
123 return dbus
.Array(ret
, signature
="(uus)")
125 def returnValue(self
, test
):
141 @dbus.service
.method(IFACE
, in_signature
='u', out_signature
='s')
142 def ReturnOneString(self
, test
):
143 return self
.returnValue(test
)
145 @dbus.service
.method(IFACE
, in_signature
='u', out_signature
='ss')
146 def ReturnTwoStrings(self
, test
):
147 return self
.returnValue(test
)
149 @dbus.service
.method(IFACE
, in_signature
='u', out_signature
='(ss)')
150 def ReturnStruct(self
, test
):
151 logger
.info('ReturnStruct(%r) -> %r', test
, self
.returnValue(test
))
152 return self
.returnValue(test
)
154 @dbus.service
.method(IFACE
, in_signature
='u', out_signature
='as')
155 def ReturnArray(self
, test
):
156 return self
.returnValue(test
)
158 @dbus.service
.method(IFACE
, in_signature
='u', out_signature
='a{ss}')
159 def ReturnDict(self
, test
):
160 return self
.returnValue(test
)
162 @dbus.service
.signal(IFACE
, signature
='s')
163 def SignalOneString(self
, test
):
164 logger
.info('SignalOneString(%r)', test
)
166 @dbus.service
.signal(IFACE
, signature
='ss')
167 def SignalTwoStrings(self
, test
, test2
):
168 logger
.info('SignalTwoStrings(%r, %r)', test
, test2
)
170 @dbus.service
.signal(IFACE
, signature
='(ss)')
171 def SignalStruct(self
, test
):
172 logger
.info('SignalStruct(%r)', test
)
174 @dbus.service
.signal(IFACE
, signature
='as')
175 def SignalArray(self
, test
):
178 @dbus.service
.signal(IFACE
, signature
='a{ss}')
179 def SignalDict(self
, test
):
182 @dbus.service
.method(IFACE
, in_signature
='su', out_signature
='')
183 def EmitSignal(self
, signal
, value
):
184 sig
= getattr(self
, str(signal
), None)
187 val
= self
.returnValue(value
)
188 # make two string case work by passing arguments in by tuple
189 if (signal
== 'SignalTwoStrings' and (value
== 1 or value
== 5)):
194 logger
.info('Emitting %s with %r', signal
, val
)
197 def CheckInheritance(self
):
200 @dbus.service
.method(IFACE
, in_signature
='', out_signature
='b')
201 def TestListExportedChildObjects(self
):
202 objs_root
= session_bus
.list_exported_child_objects('/')
203 assert objs_root
== ['org'], objs_root
204 objs_org
= session_bus
.list_exported_child_objects('/org')
205 assert objs_org
== ['freedesktop'], objs_org
208 @dbus.service
.method(IFACE
, in_signature
='bbv', out_signature
='v', async_callbacks
=('return_cb', 'error_cb'))
209 def AsynchronousMethod(self
, async, fail
, variant
, return_cb
, error_cb
):
212 gobject
.timeout_add(500, self
.AsynchronousMethod
, False, fail
, variant
, return_cb
, error_cb
)
220 return False # do not run again
224 @dbus.service
.method(IFACE
, in_signature
='', out_signature
='s', sender_keyword
='sender')
225 def WhoAmI(self
, sender
):
228 @dbus.service
.method(IFACE
, in_signature
='', out_signature
='b')
229 def AddRemovableObject(self
):
230 # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
231 # Keep the removable object reffed, since that's the use case for this
232 self
._removables
.append(RemovableObject(global_name
))
235 @dbus.service
.method(IFACE
, in_signature
='', out_signature
='b')
236 def HasRemovableObject(self
):
237 # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
238 objs
= session_bus
.list_exported_child_objects(OBJECT
)
239 return ('RemovableObject' in objs
)
241 @dbus.service
.method(IFACE
)
242 def MultipleReturnWithoutSignature(self
):
243 # https://bugs.freedesktop.org/show_bug.cgi?id=10174
244 return dbus
.String('abc'), dbus
.Int32(123)
246 session_bus
= dbus
.SessionBus()
247 global_name
= dbus
.service
.BusName(NAME
, bus
=session_bus
)
248 object = TestObject(global_name
)
249 g_object
= TestGObject(global_name
)
250 loop
= gobject
.MainLoop()