1 # Copyright (C) 2006 Collabora Ltd. <http://www.collabora.co.uk/>
3 # Permission is hereby granted, free of charge, to any person
4 # obtaining a copy of this software and associated documentation
5 # files (the "Software"), to deal in the Software without
6 # restriction, including without limitation the rights to use, copy,
7 # modify, merge, publish, distribute, sublicense, and/or sell copies
8 # of the Software, and to permit persons to whom the Software is
9 # furnished to do so, subject to the following conditions:
11 # The above copyright notice and this permission notice shall be
12 # included in all copies or substantial portions of the Software.
14 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
15 # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
16 # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
17 # NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
18 # HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
19 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
21 # DEALINGS IN THE SOFTWARE.
29 from dbus
import SessionBus
30 from dbus
.service
import BusName
32 from crosstest
import CROSS_TEST_PATH
, CROSS_TEST_BUS_NAME
, \
33 INTERFACE_SINGLE_TESTS
, INTERFACE_TESTS
,\
34 INTERFACE_CALLBACK_TESTS
, INTERFACE_SIGNAL_TESTS
,\
39 logging
.getLogger().setLevel(1)
40 logger
= logging
.getLogger('cross-test-server')
43 class VerboseSet(Set
):
52 tested_things
= VerboseSet()
54 INTERFACE_SINGLE_TESTS
+ '.Sum',
55 INTERFACE_TESTS
+ '.Identity',
56 INTERFACE_TESTS
+ '.IdentityByte',
57 INTERFACE_TESTS
+ '.IdentityBool',
58 INTERFACE_TESTS
+ '.IdentityInt16',
59 INTERFACE_TESTS
+ '.IdentityUInt16',
60 INTERFACE_TESTS
+ '.IdentityInt32',
61 INTERFACE_TESTS
+ '.IdentityUInt32',
62 INTERFACE_TESTS
+ '.IdentityInt64',
63 INTERFACE_TESTS
+ '.IdentityUInt64',
64 INTERFACE_TESTS
+ '.IdentityDouble',
65 INTERFACE_TESTS
+ '.IdentityString',
66 INTERFACE_TESTS
+ '.IdentityArray',
67 INTERFACE_TESTS
+ '.IdentityByteArray',
68 INTERFACE_TESTS
+ '.IdentityBoolArray',
69 INTERFACE_TESTS
+ '.IdentityInt16Array',
70 INTERFACE_TESTS
+ '.IdentityUInt16Array',
71 INTERFACE_TESTS
+ '.IdentityInt32Array',
72 INTERFACE_TESTS
+ '.IdentityUInt32Array',
73 INTERFACE_TESTS
+ '.IdentityInt64Array',
74 INTERFACE_TESTS
+ '.IdentityUInt64Array',
75 INTERFACE_TESTS
+ '.IdentityDoubleArray',
76 INTERFACE_TESTS
+ '.IdentityStringArray',
77 INTERFACE_TESTS
+ '.Sum',
78 INTERFACE_TESTS
+ '.InvertMapping',
79 INTERFACE_TESTS
+ '.DeStruct',
80 INTERFACE_TESTS
+ '.Primitize',
81 INTERFACE_TESTS
+ '.Trigger',
82 INTERFACE_TESTS
+ '.Exit',
83 INTERFACE_TESTS
+ '.Invert',
84 INTERFACE_SIGNAL_TESTS
+ '.Trigger',
88 class SingleTestsImpl(dbus
.service
.Object
):
90 @dbus.service
.method(INTERFACE_SINGLE_TESTS
, 'ay', 'u')
92 tested_things
.add(INTERFACE_SINGLE_TESTS
+ '.Sum')
94 logger
.info('Sum of %r is %r', input, u
)
98 class TestsImpl(dbus
.service
.Object
):
100 def __init__(self
, bus_name
, service_path
, exit_fn
):
101 self
._exit
_fn
= exit_fn
102 dbus
.service
.Object
.__init
__(self
, bus_name
, service_path
)
104 @dbus.service
.method(INTERFACE_TESTS
, 'v', 'v')
105 def Identity(self
, input):
106 tested_things
.add(INTERFACE_TESTS
+ '.Identity')
109 @dbus.service
.method(INTERFACE_TESTS
, 'y', 'y')
110 def IdentityByte(self
, input):
111 tested_things
.add(INTERFACE_TESTS
+ '.IdentityByte')
114 @dbus.service
.method(INTERFACE_TESTS
, 'b', 'b')
115 def IdentityBool(self
, input):
116 tested_things
.add(INTERFACE_TESTS
+ '.IdentityBool')
119 @dbus.service
.method(INTERFACE_TESTS
, 'n', 'n')
120 def IdentityInt16(self
, input):
121 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt16')
124 @dbus.service
.method(INTERFACE_TESTS
, 'q', 'q')
125 def IdentityUInt16(self
, input):
126 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt16')
129 @dbus.service
.method(INTERFACE_TESTS
, 'i', 'i')
130 def IdentityInt32(self
, input):
131 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt32')
134 @dbus.service
.method(INTERFACE_TESTS
, 'u', 'u')
135 def IdentityUInt32(self
, input):
136 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt32')
139 @dbus.service
.method(INTERFACE_TESTS
, 'x', 'x')
140 def IdentityInt64(self
, input):
141 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt64')
144 @dbus.service
.method(INTERFACE_TESTS
, 't', 't')
145 def IdentityUInt64(self
, input):
146 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt64')
149 @dbus.service
.method(INTERFACE_TESTS
, 'd', 'd')
150 def IdentityDouble(self
, input):
151 tested_things
.add(INTERFACE_TESTS
+ '.IdentityDouble')
154 @dbus.service
.method(INTERFACE_TESTS
, 's', 's')
155 def IdentityString(self
, input):
156 tested_things
.add(INTERFACE_TESTS
+ '.IdentityString')
159 @dbus.service
.method(INTERFACE_TESTS
, 'av', 'av')
160 def IdentityArray(self
, input):
161 tested_things
.add(INTERFACE_TESTS
+ '.IdentityArray')
164 @dbus.service
.method(INTERFACE_TESTS
, 'ay', 'ay')
165 def IdentityByteArray(self
, input):
166 tested_things
.add(INTERFACE_TESTS
+ '.IdentityByteArray')
169 @dbus.service
.method(INTERFACE_TESTS
, 'ab', 'ab')
170 def IdentityBoolArray(self
, input):
171 tested_things
.add(INTERFACE_TESTS
+ '.IdentityBoolArray')
174 @dbus.service
.method(INTERFACE_TESTS
, 'an', 'an')
175 def IdentityInt16Array(self
, input):
176 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt16Array')
179 @dbus.service
.method(INTERFACE_TESTS
, 'aq', 'aq')
180 def IdentityUInt16Array(self
, input):
181 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt16Array')
184 @dbus.service
.method(INTERFACE_TESTS
, 'ai', 'ai')
185 def IdentityInt32Array(self
, input):
186 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt32Array')
189 @dbus.service
.method(INTERFACE_TESTS
, 'au', 'au')
190 def IdentityUInt32Array(self
, input):
191 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt32Array')
194 @dbus.service
.method(INTERFACE_TESTS
, 'ax', 'ax')
195 def IdentityInt64Array(self
, input):
196 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt64Array')
199 @dbus.service
.method(INTERFACE_TESTS
, 'at', 'at')
200 def IdentityUInt64Array(self
, input):
201 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt64Array')
204 @dbus.service
.method(INTERFACE_TESTS
, 'ad', 'ad')
205 def IdentityDoubleArray(self
, input):
206 tested_things
.add(INTERFACE_TESTS
+ '.IdentityDoubleArray')
209 @dbus.service
.method(INTERFACE_TESTS
, 'as', 'as')
210 def IdentityStringArray(self
, input):
211 tested_things
.add(INTERFACE_TESTS
+ '.IdentityStringArray')
214 @dbus.service
.method(INTERFACE_TESTS
, 'ai', 'x')
215 def Sum(self
, input):
216 tested_things
.add(INTERFACE_TESTS
+ '.Sum')
218 logger
.info('Sum of %r is %r', input, x
)
222 @dbus.service
.method(INTERFACE_TESTS
, 'a{ss}', 'a{sas}', utf8_strings
=True)
223 def InvertMapping(self
, input):
224 tested_things
.add(INTERFACE_TESTS
+ '.InvertMapping')
225 output
= dbus
.Dictionary({})
226 for k
, v
in input.iteritems():
227 output
.setdefault(v
, []).append(k
)
230 @dbus.service
.method(INTERFACE_TESTS
, '(sun)', 'sun')
231 def DeStruct(self
, input):
232 tested_things
.add(INTERFACE_TESTS
+ '.DeStruct')
235 @dbus.service
.method(INTERFACE_TESTS
, 'v', 'av')
236 def Primitize(self
, input):
237 tested_things
.add(INTERFACE_TESTS
+ '.Primitize')
238 return list(self
.primitivize_helper(input))
240 def primitivize_helper(self
, input):
241 if (isinstance(input, tuple) or isinstance(input, dbus
.Struct
)
242 or isinstance(input, list) or isinstance(input, dbus
.Array
)):
244 for y
in self
.primitivize_helper(x
):
246 elif isinstance(input, dbus
.ByteArray
):
248 yield dbus
.Byte(ord(x
))
249 elif isinstance(input, dict) or isinstance(input, dbus
.Dictionary
):
251 for y
in self
.primitivize_helper(x
):
253 for y
in self
.primitivize_helper(input[x
]):
255 elif input.variant_level
> 0:
256 yield input.__class
__(input)
260 @dbus.service
.method(INTERFACE_TESTS
, 'b', 'b')
261 def Invert(self
, input):
262 tested_things
.add(INTERFACE_TESTS
+ '.Invert')
265 @dbus.service
.method(INTERFACE_TESTS
, 'st', '', utf8_strings
=True,
266 connection_keyword
='conn')
267 def Trigger(self
, object, parameter
, conn
=None):
268 assert isinstance(object, str)
269 logger
.info('method/signal: client wants me to emit Triggered(%r) from %r', parameter
, object)
270 tested_things
.add(INTERFACE_TESTS
+ '.Trigger')
271 gobject
.idle_add(lambda: self
.emit_Triggered_from(conn
, object,
274 def emit_Triggered_from(self
, conn
, object, parameter
):
275 assert isinstance(object, str)
276 logger
.info('method/signal: Emitting Triggered(%r) from %r', parameter
, object)
277 obj
= objects
.get(object, None)
279 obj
= SignalTestsImpl(conn
, object)
280 objects
[object] = obj
281 obj
.Triggered(parameter
)
282 logger
.info('method/signal: Emitted Triggered')
284 @dbus.service
.method(INTERFACE_TESTS
, '', '')
286 logger
.info('client wants me to Exit')
287 tested_things
.add(INTERFACE_TESTS
+ '.Exit')
288 for x
in testable_things
:
289 if x
not in tested_things
:
290 print '%s untested' % x
291 logger
.info('will quit when idle')
292 gobject
.idle_add(self
._exit
_fn
)
295 class Server(SingleTestsImpl
, TestsImpl
, SignalTestsImpl
):
297 def triggered_by_client(self
, parameter1
, parameter2
, sender
, sender_path
):
298 # Called when the client emits TestSignals.Trigger from any object.
299 logger
.info('signal/callback: Triggered by client (%s:%s): (%r,%r)', sender
, sender_path
, parameter1
, parameter2
)
300 tested_things
.add(INTERFACE_SIGNAL_TESTS
+ '.Trigger')
301 dbus
.Interface(dbus
.SessionBus().get_object(sender
, sender_path
),
302 INTERFACE_CALLBACK_TESTS
).Response(parameter1
, parameter2
)
303 logger
.info('signal/callback: Sent Response')
307 if __name__
== '__main__':
309 bus_name
= BusName(CROSS_TEST_BUS_NAME
, bus
=bus
)
310 loop
= gobject
.MainLoop()
311 obj
= Server(bus_name
, CROSS_TEST_PATH
, loop
.quit
)
312 objects
[CROSS_TEST_PATH
] = obj
313 bus
.add_signal_receiver(obj
.triggered_by_client
,
314 signal_name
='Trigger',
315 dbus_interface
=INTERFACE_SIGNAL_TESTS
,
318 sender_keyword
='sender',
319 path_keyword
='sender_path',
322 logger
.info("running...")
324 logger
.info("main loop exited.")