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.
23
from time import sleep
28
from dbus import SessionBus, Interface, Array, Byte, Double, Boolean, ByteArray, Int16, Int32, Int64, UInt16, UInt32, UInt64, String, UTF8String, Struct, Dictionary
29
from dbus.service import BusName
32
from crosstest import CROSS_TEST_PATH, CROSS_TEST_BUS_NAME,\
33
INTERFACE_SINGLE_TESTS, INTERFACE_TESTS,\
34
INTERFACE_SIGNAL_TESTS, INTERFACE_CALLBACK_TESTS,\
39
logging.getLogger().setLevel(1)
40
logger = logging.getLogger('cross-test-client')
43
class Client(SignalTestsImpl):
48
for x in self.expected:
50
print "%s fail %d" % (x, self.fail_id)
51
s = "report %d: reply to %s didn't arrive" % (self.fail_id, x)
54
logger.info("asking server to Exit")
55
Interface(self.obj, INTERFACE_TESTS).Exit(reply_handler=self.quit_reply_handler, error_handler=self.quit_error_handler)
56
# if the server doesn't reply we'll just exit anyway
57
gobject.timeout_add(1000, lambda: (loop.quit(), False)[1])
59
def quit_reply_handler(self):
60
logger.info("server says it will exit")
63
def quit_error_handler(self, e):
64
logger.error("error telling server to quit: %s %s",
68
@dbus.service.method(INTERFACE_CALLBACK_TESTS, 'qd')
69
def Response(self, input1, input2):
70
logger.info("signal/callback: Response received (%r,%r)",
72
self.expected.discard('%s.Trigger' % INTERFACE_SIGNAL_TESTS)
73
if (input1, input2) != (42, 23):
75
print "%s.Trigger fail %d" % (INTERFACE_SIGNAL_TESTS, self.fail_id)
76
s = ("report %d: expected (42,23), got %r"
77
% (self.fail_id, (input1, input2)))
81
print "%s.Trigger pass" % INTERFACE_SIGNAL_TESTS
84
def assert_method_matches(self, interface, check_fn, check_arg, member, *args):
85
if_obj = Interface(self.obj, interface)
86
method = getattr(if_obj, member)
88
real_ret = method(*args)
91
print "%s.%s fail %d" % (interface, member, self.fail_id)
92
s = ("report %d: %s.%s%r: raised %r \"%s\""
93
% (self.fail_id, interface, member, args, e, e))
96
__import__('traceback').print_exc()
99
check_fn(real_ret, check_arg)
102
print "%s.%s fail %d" % (interface, member, self.fail_id)
103
s = ("report %d: %s.%s%r: %s"
104
% (self.fail_id, interface, member, args, e))
108
print "%s.%s pass" % (interface, member)
110
def assert_method_eq(self, interface, ret, member, *args):
111
def equals(real_ret, exp):
113
raise AssertionError('expected %r of class %s, got %r of class %s' % (exp, exp.__class__, real_ret, real_ret.__class__))
115
raise AssertionError('expected %r, got %r' % (exp, real_ret))
116
if not isinstance(exp, (tuple, type(None))):
117
if real_ret.variant_level != getattr(exp, 'variant_level', 0):
118
raise AssertionError('expected variant_level=%d, got %r with level %d'
119
% (getattr(exp, 'variant_level', 0), real_ret,
120
real_ret.variant_level))
121
if isinstance(exp, list) or isinstance(exp, tuple):
122
for i in xrange(len(exp)):
124
equals(real_ret[i], exp[i])
125
except AssertionError, e:
126
if not isinstance(e.args, tuple):
128
e.args = e.args + ('(at position %d in sequence)' % i,)
130
elif isinstance(exp, dict):
133
equals(real_ret[k], exp[k])
134
except AssertionError, e:
135
if not isinstance(e.args, tuple):
137
e.args = e.args + ('(at key %r in dict)' % k,)
139
self.assert_method_matches(interface, equals, ret, member, *args)
141
def assert_InvertMapping_eq(self, interface, expected, member, mapping):
142
def check(real_ret, exp):
144
if key not in real_ret:
145
raise AssertionError('missing key %r' % key)
148
raise AssertionError('unexpected key %r' % key)
149
got = list(real_ret[key])
150
wanted = list(exp[key])
154
raise AssertionError('expected %r => %r, got %r'
155
% (key, wanted, got))
156
self.assert_method_matches(interface, check, expected, member, mapping)
158
def triggered_cb(self, param, sender_path):
159
logger.info("method/signal: Triggered(%r) by %r",
161
self.expected.discard('%s.Trigger' % INTERFACE_TESTS)
162
if sender_path != '/Where/Ever':
164
print "%s.Trigger fail %d" % (INTERFACE_TESTS, self.fail_id)
165
s = ("report %d: expected signal from /Where/Ever, got %r"
166
% (self.fail_id, sender_path))
171
print "%s.Trigger fail %d" % (INTERFACE_TESTS, self.fail_id)
172
s = ("report %d: expected signal param 42, got %r"
173
% (self.fail_id, parameter))
177
print "%s.Trigger pass" % INTERFACE_TESTS
179
def trigger_returned_cb(self):
180
logger.info('method/signal: Trigger() returned')
182
logger.info("signal/callback: Emitting signal to trigger callback")
183
self.expected.add('%s.Trigger' % INTERFACE_SIGNAL_TESTS)
184
self.Trigger(UInt16(42), 23.0)
185
logger.info("signal/callback: Emitting signal returned")
187
def run_client(self):
189
obj = bus.get_object(CROSS_TEST_BUS_NAME, CROSS_TEST_PATH)
192
self.run_synchronous_tests(obj)
195
logger.info("Binding signal handler for Triggered")
196
# FIXME: doesn't seem to work when going via the Interface method
197
# FIXME: should be possible to ask the proxy object for its
199
bus.add_signal_receiver(self.triggered_cb, 'Triggered',
200
INTERFACE_SIGNAL_TESTS,
202
path_keyword='sender_path')
203
logger.info("method/signal: Triggering signal")
204
self.expected.add('%s.Trigger' % INTERFACE_TESTS)
205
Interface(obj, INTERFACE_TESTS).Trigger(u'/Where/Ever', dbus.UInt64(42), reply_handler=self.trigger_returned_cb, error_handler=self.trigger_error_handler)
207
def trigger_error_handler(self, e):
208
logger.error("method/signal: %s %s", e.__class__, e)
209
Interface(self.obj, INTERFACE_TESTS).Exit()
212
def run_synchronous_tests(self, obj):
213
# We can't test that coercion works correctly unless the server has
214
# sent us introspection data. Java doesn't :-/
215
have_signatures = True
219
self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', [1, 2, 3])
220
self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', ['\x01', '\x02', '\x03'])
221
self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', [Byte(1), Byte(2), Byte(3)])
222
self.assert_method_eq(INTERFACE_SINGLE_TESTS, 6, 'Sum', ByteArray('\x01\x02\x03'))
225
self.assert_method_eq(INTERFACE_TESTS, String(u'foo', variant_level=1), 'Identity', String('foo'))
226
self.assert_method_eq(INTERFACE_TESTS, String(u'foo', variant_level=1), 'Identity', UTF8String('foo'))
227
self.assert_method_eq(INTERFACE_TESTS, Byte(42, variant_level=1), 'Identity', Byte(42))
228
self.assert_method_eq(INTERFACE_TESTS, Byte(42, variant_level=23), 'Identity', Byte(42, variant_level=23))
229
self.assert_method_eq(INTERFACE_TESTS, Double(42.5, variant_level=1), 'Identity', 42.5)
230
self.assert_method_eq(INTERFACE_TESTS, Double(-42.5, variant_level=1), 'Identity', -42.5)
233
self.assert_method_eq(INTERFACE_TESTS, String(u'foo', variant_level=1), 'Identity', 'foo')
234
self.assert_method_eq(INTERFACE_TESTS, Byte(42, variant_level=1), 'Identity', Byte(42))
235
self.assert_method_eq(INTERFACE_TESTS, Double(42.5, variant_level=1), 'Identity', Double(42.5))
236
self.assert_method_eq(INTERFACE_TESTS, Double(-42.5, variant_level=1), 'Identity', -42.5)
238
for i in (0, 42, 255):
239
self.assert_method_eq(INTERFACE_TESTS, Byte(i), 'IdentityByte', Byte(i))
240
for i in (True, False):
241
self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityBool', i)
243
for i in (-0x8000, 0, 42, 0x7fff):
244
self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityInt16', Int16(i))
245
for i in (0, 42, 0xffff):
246
self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityUInt16', UInt16(i))
247
for i in (-0x7fffffff-1, 0, 42, 0x7fffffff):
248
self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityInt32', Int32(i))
249
for i in (0L, 42L, 0xffffffffL):
250
self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityUInt32', UInt32(i))
251
MANY = 0x8000L * 0x10000L * 0x10000L * 0x10000L
252
for i in (-MANY, 0, 42, MANY-1):
253
self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityInt64', Int64(i))
254
for i in (0, 42, 2*MANY - 1):
255
self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityUInt64', UInt64(i))
257
self.assert_method_eq(INTERFACE_TESTS, 42.3, 'IdentityDouble', 42.3)
258
for i in ('', 'foo'):
259
self.assert_method_eq(INTERFACE_TESTS, i, 'IdentityString', i)
260
for i in (u'\xa9', '\xc2\xa9'):
261
self.assert_method_eq(INTERFACE_TESTS, u'\xa9', 'IdentityString', i)
264
self.assert_method_eq(INTERFACE_TESTS, Byte(0x42), 'IdentityByte', '\x42')
265
self.assert_method_eq(INTERFACE_TESTS, True, 'IdentityBool', 42)
266
self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityInt16', 42)
267
self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityUInt16', 42)
268
self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityInt32', 42)
269
self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityUInt32', 42)
270
self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityInt64', 42)
271
self.assert_method_eq(INTERFACE_TESTS, 42, 'IdentityUInt64', 42)
272
self.assert_method_eq(INTERFACE_TESTS, 42.0, 'IdentityDouble', 42)
274
self.assert_method_eq(INTERFACE_TESTS, [Byte('\x01', variant_level=1),
275
Byte('\x02', variant_level=1),
276
Byte('\x03', variant_level=1)],
283
self.assert_method_eq(INTERFACE_TESTS, [Int32(1, variant_level=1),
284
Int32(2, variant_level=1),
285
Int32(3, variant_level=1)],
291
self.assert_method_eq(INTERFACE_TESTS, [String(u'a', variant_level=1),
292
String(u'b', variant_level=1),
293
String(u'c', variant_level=1)],
301
self.assert_method_eq(INTERFACE_TESTS, [Byte('\x01', variant_level=1),
302
Byte('\x02', variant_level=1),
303
Byte('\x03', variant_level=1)],
305
ByteArray('\x01\x02\x03'))
306
self.assert_method_eq(INTERFACE_TESTS, [Int32(1, variant_level=1),
307
Int32(2, variant_level=1),
308
Int32(3, variant_level=1)],
313
self.assert_method_eq(INTERFACE_TESTS, [String(u'a', variant_level=1),
314
String(u'b', variant_level=1),
315
String(u'c', variant_level=1)],
319
self.assert_method_eq(INTERFACE_TESTS,
320
[Byte(1), Byte(2), Byte(3)],
322
ByteArray('\x01\x02\x03'))
324
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityByteArray', ['\x01', '\x02', '\x03'])
325
self.assert_method_eq(INTERFACE_TESTS, [False,True], 'IdentityBoolArray', [False,True])
327
self.assert_method_eq(INTERFACE_TESTS, [False,True,True], 'IdentityBoolArray', [0,1,2])
329
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt16Array', [Int16(1),Int16(2),Int16(3)])
330
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt16Array', [UInt16(1),UInt16(2),UInt16(3)])
331
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt32Array', [Int32(1),Int32(2),Int32(3)])
332
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt32Array', [UInt32(1),UInt32(2),UInt32(3)])
333
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt64Array', [Int64(1),Int64(2),Int64(3)])
334
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt64Array', [UInt64(1),UInt64(2),UInt64(3)])
337
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt16Array', [1,2,3])
338
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt16Array', [1,2,3])
339
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt32Array', [1,2,3])
340
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt32Array', [1,2,3])
341
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityInt64Array', [1,2,3])
342
self.assert_method_eq(INTERFACE_TESTS, [1,2,3], 'IdentityUInt64Array', [1,2,3])
344
self.assert_method_eq(INTERFACE_TESTS, [1.0,2.5,3.1], 'IdentityDoubleArray', [1.0,2.5,3.1])
346
self.assert_method_eq(INTERFACE_TESTS, [1.0,2.5,3.1], 'IdentityDoubleArray', [1,2.5,3.1])
347
self.assert_method_eq(INTERFACE_TESTS, ['a','b','c'], 'IdentityStringArray', ['a','b','c'])
348
self.assert_method_eq(INTERFACE_TESTS, 6, 'Sum', [Int32(1),Int32(2),Int32(3)])
350
self.assert_method_eq(INTERFACE_TESTS, 6, 'Sum', [1,2,3])
352
self.assert_InvertMapping_eq(INTERFACE_TESTS, {'fps': ['unreal', 'quake'], 'rts': ['warcraft']}, 'InvertMapping', {'unreal': 'fps', 'quake': 'fps', 'warcraft': 'rts'})
354
self.assert_method_eq(INTERFACE_TESTS, ('a', 1, 2), 'DeStruct', ('a', UInt32(1), Int16(2)))
355
self.assert_method_eq(INTERFACE_TESTS, Array([String('x', variant_level=1)]),
356
'Primitize', [String('x', variant_level=1)])
357
self.assert_method_eq(INTERFACE_TESTS, Array([String('x', variant_level=1)]),
358
'Primitize', [String('x', variant_level=23)])
359
self.assert_method_eq(INTERFACE_TESTS,
360
Array([String('x', variant_level=1),
361
Byte(1, variant_level=1),
362
Byte(2, variant_level=1)]),
364
Array([String('x'), Byte(1), Byte(2)],
366
self.assert_method_eq(INTERFACE_TESTS,
367
Array([String('x', variant_level=1),
368
Byte(1, variant_level=1),
369
Byte(2, variant_level=1)]),
371
Array([String('x'), Array([Byte(1), Byte(2)])],
373
self.assert_method_eq(INTERFACE_TESTS, Boolean(False), 'Invert', True)
374
self.assert_method_eq(INTERFACE_TESTS, Boolean(True), 'Invert', False)
376
self.assert_method_eq(INTERFACE_TESTS, Boolean(False), 'Invert', 42)
377
self.assert_method_eq(INTERFACE_TESTS, Boolean(True), 'Invert', 0)
380
if __name__ == '__main__':
381
# FIXME: should be possible to export objects without a bus name
383
client = Client(dbus.SessionBus(), '/Client')
385
# the Java cross test's interpretation is that the client should be
387
client = Client(dbus.SessionBus(), '/Test')
388
gobject.idle_add(client.run_client)
390
loop = gobject.MainLoop()
391
logger.info("running...")
393
logger.info("main loop exited.")