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 def Trigger(self, object, parameter):
263 assert isinstance(object, str)
264 logger.info('method/signal: client wants me to emit Triggered(%r) from %r', parameter, object)
265 tested_things.add(INTERFACE_TESTS + '.Trigger')
266 gobject.idle_add(lambda: self.emit_Triggered_from(object, parameter))
268 def emit_Triggered_from(self, object, parameter):
269 assert isinstance(object, str)
270 logger.info('method/signal: Emitting Triggered(%r) from %r', parameter, object)
271 obj = objects.get(object, None)
273 obj = SignalTestsImpl(dbus.service.BusName(CROSS_TEST_BUS_NAME), object)
274 objects[object] = obj
275 obj.Triggered(parameter)
276 logger.info('method/signal: Emitted Triggered')
278 @dbus.service.method(INTERFACE_TESTS, '', '')
280 logger.info('client wants me to Exit')
281 tested_things.add(INTERFACE_TESTS + '.Exit')
282 for x in testable_things:
283 if x not in tested_things:
284 print '%s untested' % x
285 logger.info('will quit when idle')
286 gobject.idle_add(self._exit_fn)
289 class Server(SingleTestsImpl, TestsImpl, SignalTestsImpl):
291 def triggered_by_client(self, parameter1, parameter2, sender, sender_path):
292 # Called when the client emits TestSignals.Trigger from any object.
293 logger.info('signal/callback: Triggered by client (%s:%s): (%r,%r)', sender, sender_path, parameter1, parameter2)
294 tested_things.add(INTERFACE_SIGNAL_TESTS + '.Trigger')
295 dbus.Interface(dbus.SessionBus().get_object(sender, sender_path),
296 INTERFACE_CALLBACK_TESTS).Response(parameter1, parameter2)
297 logger.info('signal/callback: Sent Response')
301 if __name__ == '__main__':
303 bus_name = BusName(CROSS_TEST_BUS_NAME)
304 loop = gobject.MainLoop()
305 obj = Server(bus_name, CROSS_TEST_PATH, loop.quit)
306 objects[CROSS_TEST_PATH] = obj
307 bus.add_signal_receiver(obj.triggered_by_client,
308 signal_name='Trigger',
309 dbus_interface=INTERFACE_SIGNAL_TESTS,
312 sender_keyword='sender',
313 path_keyword='sender_path',
316 logger.info("running...")
318 logger.info("main loop exited.")