1 # Copyright (C) 2006 Collabora Ltd. <http://www.collabora.co.uk/>
3 # Licensed under the Academic Free License version 2.1
5 # This program is free software; you can redistribute it and/or modify
6 # it under the terms of the GNU Lesser General Public License as published by
7 # the Free Software Foundation; either version 2.1 of the License, or
8 # (at your option) any later version.
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU Lesser General Public License for more details.
15 # You should have received a copy of the GNU Lesser General Public License
16 # along with this program; if not, write to the Free Software
17 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
25 from dbus
import SessionBus
26 from dbus
.service
import BusName
28 from crosstest
import CROSS_TEST_PATH
, CROSS_TEST_BUS_NAME
, \
29 INTERFACE_SINGLE_TESTS
, INTERFACE_TESTS
,\
30 INTERFACE_CALLBACK_TESTS
, INTERFACE_SIGNAL_TESTS
,\
35 logging
.getLogger().setLevel(1)
36 logger
= logging
.getLogger('cross-test-server')
39 class VerboseSet(Set
):
48 tested_things
= VerboseSet()
50 INTERFACE_SINGLE_TESTS
+ '.Sum',
51 INTERFACE_TESTS
+ '.Identity',
52 INTERFACE_TESTS
+ '.IdentityByte',
53 INTERFACE_TESTS
+ '.IdentityBool',
54 INTERFACE_TESTS
+ '.IdentityInt16',
55 INTERFACE_TESTS
+ '.IdentityUInt16',
56 INTERFACE_TESTS
+ '.IdentityInt32',
57 INTERFACE_TESTS
+ '.IdentityUInt32',
58 INTERFACE_TESTS
+ '.IdentityInt64',
59 INTERFACE_TESTS
+ '.IdentityUInt64',
60 INTERFACE_TESTS
+ '.IdentityDouble',
61 INTERFACE_TESTS
+ '.IdentityString',
62 INTERFACE_TESTS
+ '.IdentityArray',
63 INTERFACE_TESTS
+ '.IdentityByteArray',
64 INTERFACE_TESTS
+ '.IdentityBoolArray',
65 INTERFACE_TESTS
+ '.IdentityInt16Array',
66 INTERFACE_TESTS
+ '.IdentityUInt16Array',
67 INTERFACE_TESTS
+ '.IdentityInt32Array',
68 INTERFACE_TESTS
+ '.IdentityUInt32Array',
69 INTERFACE_TESTS
+ '.IdentityInt64Array',
70 INTERFACE_TESTS
+ '.IdentityUInt64Array',
71 INTERFACE_TESTS
+ '.IdentityDoubleArray',
72 INTERFACE_TESTS
+ '.IdentityStringArray',
73 INTERFACE_TESTS
+ '.Sum',
74 INTERFACE_TESTS
+ '.InvertMapping',
75 INTERFACE_TESTS
+ '.DeStruct',
76 INTERFACE_TESTS
+ '.Primitize',
77 INTERFACE_TESTS
+ '.Trigger',
78 INTERFACE_TESTS
+ '.Exit',
79 INTERFACE_TESTS
+ '.Invert',
80 INTERFACE_SIGNAL_TESTS
+ '.Trigger',
84 class SingleTestsImpl(dbus
.service
.Object
):
86 @dbus.service
.method(INTERFACE_SINGLE_TESTS
, 'ay', 'u')
88 tested_things
.add(INTERFACE_SINGLE_TESTS
+ '.Sum')
90 logger
.info('Sum of %r is %r', input, u
)
94 class TestsImpl(dbus
.service
.Object
):
96 def __init__(self
, bus_name
, service_path
, exit_fn
):
97 self
._exit
_fn
= exit_fn
98 dbus
.service
.Object
.__init
__(self
, bus_name
, service_path
)
100 @dbus.service
.method(INTERFACE_TESTS
, 'v', 'v')
101 def Identity(self
, input):
102 tested_things
.add(INTERFACE_TESTS
+ '.Identity')
105 @dbus.service
.method(INTERFACE_TESTS
, 'y', 'y')
106 def IdentityByte(self
, input):
107 tested_things
.add(INTERFACE_TESTS
+ '.IdentityByte')
110 @dbus.service
.method(INTERFACE_TESTS
, 'b', 'b')
111 def IdentityBool(self
, input):
112 tested_things
.add(INTERFACE_TESTS
+ '.IdentityBool')
115 @dbus.service
.method(INTERFACE_TESTS
, 'n', 'n')
116 def IdentityInt16(self
, input):
117 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt16')
120 @dbus.service
.method(INTERFACE_TESTS
, 'q', 'q')
121 def IdentityUInt16(self
, input):
122 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt16')
125 @dbus.service
.method(INTERFACE_TESTS
, 'i', 'i')
126 def IdentityInt32(self
, input):
127 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt32')
130 @dbus.service
.method(INTERFACE_TESTS
, 'u', 'u')
131 def IdentityUInt32(self
, input):
132 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt32')
135 @dbus.service
.method(INTERFACE_TESTS
, 'x', 'x')
136 def IdentityInt64(self
, input):
137 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt64')
140 @dbus.service
.method(INTERFACE_TESTS
, 't', 't')
141 def IdentityUInt64(self
, input):
142 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt64')
145 @dbus.service
.method(INTERFACE_TESTS
, 'd', 'd')
146 def IdentityDouble(self
, input):
147 tested_things
.add(INTERFACE_TESTS
+ '.IdentityDouble')
150 @dbus.service
.method(INTERFACE_TESTS
, 's', 's')
151 def IdentityString(self
, input):
152 tested_things
.add(INTERFACE_TESTS
+ '.IdentityString')
155 @dbus.service
.method(INTERFACE_TESTS
, 'av', 'av')
156 def IdentityArray(self
, input):
157 tested_things
.add(INTERFACE_TESTS
+ '.IdentityArray')
160 @dbus.service
.method(INTERFACE_TESTS
, 'ay', 'ay')
161 def IdentityByteArray(self
, input):
162 tested_things
.add(INTERFACE_TESTS
+ '.IdentityByteArray')
165 @dbus.service
.method(INTERFACE_TESTS
, 'ab', 'ab')
166 def IdentityBoolArray(self
, input):
167 tested_things
.add(INTERFACE_TESTS
+ '.IdentityBoolArray')
170 @dbus.service
.method(INTERFACE_TESTS
, 'an', 'an')
171 def IdentityInt16Array(self
, input):
172 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt16Array')
175 @dbus.service
.method(INTERFACE_TESTS
, 'aq', 'aq')
176 def IdentityUInt16Array(self
, input):
177 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt16Array')
180 @dbus.service
.method(INTERFACE_TESTS
, 'ai', 'ai')
181 def IdentityInt32Array(self
, input):
182 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt32Array')
185 @dbus.service
.method(INTERFACE_TESTS
, 'au', 'au')
186 def IdentityUInt32Array(self
, input):
187 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt32Array')
190 @dbus.service
.method(INTERFACE_TESTS
, 'ax', 'ax')
191 def IdentityInt64Array(self
, input):
192 tested_things
.add(INTERFACE_TESTS
+ '.IdentityInt64Array')
195 @dbus.service
.method(INTERFACE_TESTS
, 'at', 'at')
196 def IdentityUInt64Array(self
, input):
197 tested_things
.add(INTERFACE_TESTS
+ '.IdentityUInt64Array')
200 @dbus.service
.method(INTERFACE_TESTS
, 'ad', 'ad')
201 def IdentityDoubleArray(self
, input):
202 tested_things
.add(INTERFACE_TESTS
+ '.IdentityDoubleArray')
205 @dbus.service
.method(INTERFACE_TESTS
, 'as', 'as')
206 def IdentityStringArray(self
, input):
207 tested_things
.add(INTERFACE_TESTS
+ '.IdentityStringArray')
210 @dbus.service
.method(INTERFACE_TESTS
, 'ai', 'x')
211 def Sum(self
, input):
212 tested_things
.add(INTERFACE_TESTS
+ '.Sum')
214 logger
.info('Sum of %r is %r', input, x
)
218 @dbus.service
.method(INTERFACE_TESTS
, 'a{ss}', 'a{sas}', utf8_strings
=True)
219 def InvertMapping(self
, input):
220 tested_things
.add(INTERFACE_TESTS
+ '.InvertMapping')
221 output
= dbus
.Dictionary({})
222 for k
, v
in input.iteritems():
223 output
.setdefault(v
, []).append(k
)
226 @dbus.service
.method(INTERFACE_TESTS
, '(sun)', 'sun')
227 def DeStruct(self
, input):
228 tested_things
.add(INTERFACE_TESTS
+ '.DeStruct')
231 @dbus.service
.method(INTERFACE_TESTS
, 'v', 'av')
232 def Primitize(self
, input):
233 tested_things
.add(INTERFACE_TESTS
+ '.Primitize')
234 return list(self
.primitivize_helper(input))
236 def primitivize_helper(self
, input):
237 if (isinstance(input, tuple) or isinstance(input, dbus
.Struct
)
238 or isinstance(input, list) or isinstance(input, dbus
.Array
)):
240 for y
in self
.primitivize_helper(x
):
242 elif isinstance(input, dbus
.ByteArray
):
244 yield dbus
.Byte(ord(x
))
245 elif isinstance(input, dict) or isinstance(input, dbus
.Dictionary
):
247 for y
in self
.primitivize_helper(x
):
249 for y
in self
.primitivize_helper(input[x
]):
251 elif input.variant_level
> 0:
252 yield input.__class
__(input)
256 @dbus.service
.method(INTERFACE_TESTS
, 'b', 'b')
257 def Invert(self
, input):
258 tested_things
.add(INTERFACE_TESTS
+ '.Invert')
261 @dbus.service
.method(INTERFACE_TESTS
, 'st', '', utf8_strings
=True,
262 connection_keyword
='conn')
263 def Trigger(self
, object, parameter
, conn
=None):
264 assert isinstance(object, str)
265 logger
.info('method/signal: client wants me to emit Triggered(%r) from %r', parameter
, object)
266 tested_things
.add(INTERFACE_TESTS
+ '.Trigger')
267 gobject
.idle_add(lambda: self
.emit_Triggered_from(conn
, object,
270 def emit_Triggered_from(self
, conn
, object, parameter
):
271 assert isinstance(object, str)
272 logger
.info('method/signal: Emitting Triggered(%r) from %r', parameter
, object)
273 obj
= objects
.get(object, None)
275 obj
= SignalTestsImpl(conn
, object)
276 objects
[object] = obj
277 obj
.Triggered(parameter
)
278 logger
.info('method/signal: Emitted Triggered')
280 @dbus.service
.method(INTERFACE_TESTS
, '', '')
282 logger
.info('client wants me to Exit')
283 tested_things
.add(INTERFACE_TESTS
+ '.Exit')
284 for x
in testable_things
:
285 if x
not in tested_things
:
286 print '%s untested' % x
287 logger
.info('will quit when idle')
288 gobject
.idle_add(self
._exit
_fn
)
291 class Server(SingleTestsImpl
, TestsImpl
, SignalTestsImpl
):
293 def triggered_by_client(self
, parameter1
, parameter2
, sender
, sender_path
):
294 # Called when the client emits TestSignals.Trigger from any object.
295 logger
.info('signal/callback: Triggered by client (%s:%s): (%r,%r)', sender
, sender_path
, parameter1
, parameter2
)
296 tested_things
.add(INTERFACE_SIGNAL_TESTS
+ '.Trigger')
297 dbus
.Interface(dbus
.SessionBus().get_object(sender
, sender_path
),
298 INTERFACE_CALLBACK_TESTS
).Response(parameter1
, parameter2
)
299 logger
.info('signal/callback: Sent Response')
303 if __name__
== '__main__':
305 bus_name
= BusName(CROSS_TEST_BUS_NAME
, bus
=bus
)
306 loop
= gobject
.MainLoop()
307 obj
= Server(bus_name
, CROSS_TEST_PATH
, loop
.quit
)
308 objects
[CROSS_TEST_PATH
] = obj
309 bus
.add_signal_receiver(obj
.triggered_by_client
,
310 signal_name
='Trigger',
311 dbus_interface
=INTERFACE_SIGNAL_TESTS
,
314 sender_keyword
='sender',
315 path_keyword
='sender_path',
318 logger
.info("running...")
320 logger
.info("main loop exited.")