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))
267     
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.")