Actually commit the numerous copyright-statement changes.
[dbus-python-phuang.git] / test / cross-test-server.py
blob1db0cbc54592d2b8e5e545bc3c3bc8e4dc817ecb
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
19 from sets import Set
20 import logging
22 import gobject
24 import dbus.glib
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,\
31 SignalTestsImpl
34 logging.basicConfig()
35 logging.getLogger().setLevel(1)
36 logger = logging.getLogger('cross-test-server')
39 class VerboseSet(Set):
40 def add(self, thing):
41 print '%s ok' % thing
42 Set.add(self, thing)
45 objects = {}
48 tested_things = VerboseSet()
49 testable_things = [
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')
87 def Sum(self, input):
88 tested_things.add(INTERFACE_SINGLE_TESTS + '.Sum')
89 u = sum(input)
90 logger.info('Sum of %r is %r', input, u)
91 return 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')
103 return input
105 @dbus.service.method(INTERFACE_TESTS, 'y', 'y')
106 def IdentityByte(self, input):
107 tested_things.add(INTERFACE_TESTS + '.IdentityByte')
108 return input
110 @dbus.service.method(INTERFACE_TESTS, 'b', 'b')
111 def IdentityBool(self, input):
112 tested_things.add(INTERFACE_TESTS + '.IdentityBool')
113 return input
115 @dbus.service.method(INTERFACE_TESTS, 'n', 'n')
116 def IdentityInt16(self, input):
117 tested_things.add(INTERFACE_TESTS + '.IdentityInt16')
118 return input
120 @dbus.service.method(INTERFACE_TESTS, 'q', 'q')
121 def IdentityUInt16(self, input):
122 tested_things.add(INTERFACE_TESTS + '.IdentityUInt16')
123 return input
125 @dbus.service.method(INTERFACE_TESTS, 'i', 'i')
126 def IdentityInt32(self, input):
127 tested_things.add(INTERFACE_TESTS + '.IdentityInt32')
128 return input
130 @dbus.service.method(INTERFACE_TESTS, 'u', 'u')
131 def IdentityUInt32(self, input):
132 tested_things.add(INTERFACE_TESTS + '.IdentityUInt32')
133 return input
135 @dbus.service.method(INTERFACE_TESTS, 'x', 'x')
136 def IdentityInt64(self, input):
137 tested_things.add(INTERFACE_TESTS + '.IdentityInt64')
138 return input
140 @dbus.service.method(INTERFACE_TESTS, 't', 't')
141 def IdentityUInt64(self, input):
142 tested_things.add(INTERFACE_TESTS + '.IdentityUInt64')
143 return input
145 @dbus.service.method(INTERFACE_TESTS, 'd', 'd')
146 def IdentityDouble(self, input):
147 tested_things.add(INTERFACE_TESTS + '.IdentityDouble')
148 return input
150 @dbus.service.method(INTERFACE_TESTS, 's', 's')
151 def IdentityString(self, input):
152 tested_things.add(INTERFACE_TESTS + '.IdentityString')
153 return input
155 @dbus.service.method(INTERFACE_TESTS, 'av', 'av')
156 def IdentityArray(self, input):
157 tested_things.add(INTERFACE_TESTS + '.IdentityArray')
158 return input
160 @dbus.service.method(INTERFACE_TESTS, 'ay', 'ay')
161 def IdentityByteArray(self, input):
162 tested_things.add(INTERFACE_TESTS + '.IdentityByteArray')
163 return input
165 @dbus.service.method(INTERFACE_TESTS, 'ab', 'ab')
166 def IdentityBoolArray(self, input):
167 tested_things.add(INTERFACE_TESTS + '.IdentityBoolArray')
168 return input
170 @dbus.service.method(INTERFACE_TESTS, 'an', 'an')
171 def IdentityInt16Array(self, input):
172 tested_things.add(INTERFACE_TESTS + '.IdentityInt16Array')
173 return input
175 @dbus.service.method(INTERFACE_TESTS, 'aq', 'aq')
176 def IdentityUInt16Array(self, input):
177 tested_things.add(INTERFACE_TESTS + '.IdentityUInt16Array')
178 return input
180 @dbus.service.method(INTERFACE_TESTS, 'ai', 'ai')
181 def IdentityInt32Array(self, input):
182 tested_things.add(INTERFACE_TESTS + '.IdentityInt32Array')
183 return input
185 @dbus.service.method(INTERFACE_TESTS, 'au', 'au')
186 def IdentityUInt32Array(self, input):
187 tested_things.add(INTERFACE_TESTS + '.IdentityUInt32Array')
188 return input
190 @dbus.service.method(INTERFACE_TESTS, 'ax', 'ax')
191 def IdentityInt64Array(self, input):
192 tested_things.add(INTERFACE_TESTS + '.IdentityInt64Array')
193 return input
195 @dbus.service.method(INTERFACE_TESTS, 'at', 'at')
196 def IdentityUInt64Array(self, input):
197 tested_things.add(INTERFACE_TESTS + '.IdentityUInt64Array')
198 return input
200 @dbus.service.method(INTERFACE_TESTS, 'ad', 'ad')
201 def IdentityDoubleArray(self, input):
202 tested_things.add(INTERFACE_TESTS + '.IdentityDoubleArray')
203 return input
205 @dbus.service.method(INTERFACE_TESTS, 'as', 'as')
206 def IdentityStringArray(self, input):
207 tested_things.add(INTERFACE_TESTS + '.IdentityStringArray')
208 return input
210 @dbus.service.method(INTERFACE_TESTS, 'ai', 'x')
211 def Sum(self, input):
212 tested_things.add(INTERFACE_TESTS + '.Sum')
213 x = sum(input)
214 logger.info('Sum of %r is %r', input, x)
215 return 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)
224 return output
226 @dbus.service.method(INTERFACE_TESTS, '(sun)', 'sun')
227 def DeStruct(self, input):
228 tested_things.add(INTERFACE_TESTS + '.DeStruct')
229 return input
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)):
239 for x in input:
240 for y in self.primitivize_helper(x):
241 yield y
242 elif isinstance(input, dbus.ByteArray):
243 for x in input:
244 yield dbus.Byte(ord(x))
245 elif isinstance(input, dict) or isinstance(input, dbus.Dictionary):
246 for x in input:
247 for y in self.primitivize_helper(x):
248 yield y
249 for y in self.primitivize_helper(input[x]):
250 yield y
251 elif input.variant_level > 0:
252 yield input.__class__(input)
253 else:
254 yield input
256 @dbus.service.method(INTERFACE_TESTS, 'b', 'b')
257 def Invert(self, input):
258 tested_things.add(INTERFACE_TESTS + '.Invert')
259 return not input
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)
272 if obj is 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, '', '')
279 def Exit(self):
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__':
302 bus = SessionBus()
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,
310 named_service=None,
311 path=None,
312 sender_keyword='sender',
313 path_keyword='sender_path',
314 utf8_strings=True)
316 logger.info("running...")
317 loop.run()
318 logger.info("main loop exited.")