1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
from dbus_next.service import ServiceInterface, method
from dbus_next.aio import MessageBus
from dbus_next import Message, MessageType, ErrorType, Variant, SignatureTree, DBusError, MessageFlag
import pytest
class ExampleInterface(ServiceInterface):
def __init__(self, name):
super().__init__(name)
@method()
def echo(self, what: 's') -> 's':
assert type(self) is ExampleInterface
return what
@method()
def echo_multiple(self, what1: 's', what2: 's') -> 'ss':
assert type(self) is ExampleInterface
return [what1, what2]
@method()
def echo_containers(self, array: 'as', variant: 'v', dict_entries: 'a{sv}',
struct: '(s(s(v)))') -> 'asva{sv}(s(s(v)))':
assert type(self) is ExampleInterface
return [array, variant, dict_entries, struct]
@method()
def ping(self):
assert type(self) is ExampleInterface
pass
@method(name='renamed')
def original_name(self):
assert type(self) is ExampleInterface
pass
@method(disabled=True)
def not_here(self):
assert type(self) is ExampleInterface
pass
@method()
def throws_unexpected_error(self):
assert type(self) is ExampleInterface
raise Exception('oops')
@method()
def throws_dbus_error(self):
assert type(self) is ExampleInterface
raise DBusError('test.error', 'an error ocurred')
class AsyncInterface(ServiceInterface):
def __init__(self, name):
super().__init__(name)
@method()
async def echo(self, what: 's') -> 's':
assert type(self) is AsyncInterface
return what
@method()
async def echo_multiple(self, what1: 's', what2: 's') -> 'ss':
assert type(self) is AsyncInterface
return [what1, what2]
@method()
async def echo_containers(self, array: 'as', variant: 'v', dict_entries: 'a{sv}',
struct: '(s(s(v)))') -> 'asva{sv}(s(s(v)))':
assert type(self) is AsyncInterface
return [array, variant, dict_entries, struct]
@method()
async def ping(self):
assert type(self) is AsyncInterface
pass
@method(name='renamed')
async def original_name(self):
assert type(self) is AsyncInterface
pass
@method(disabled=True)
async def not_here(self):
assert type(self) is AsyncInterface
pass
@method()
async def throws_unexpected_error(self):
assert type(self) is AsyncInterface
raise Exception('oops')
@method()
def throws_dbus_error(self):
assert type(self) is AsyncInterface
raise DBusError('test.error', 'an error ocurred')
@pytest.mark.parametrize('interface_class', [ExampleInterface, AsyncInterface])
@pytest.mark.asyncio
async def test_methods(interface_class):
bus1 = await MessageBus().connect()
bus2 = await MessageBus().connect()
interface = interface_class('test.interface')
export_path = '/test/path'
async def call(member, signature='', body=[], flags=MessageFlag.NONE):
return await bus2.call(
Message(destination=bus1.unique_name,
path=export_path,
interface=interface.name,
member=member,
signature=signature,
body=body,
flags=flags))
bus1.export(export_path, interface)
body = ['hello world']
reply = await call('echo', 's', body)
assert reply.message_type == MessageType.METHOD_RETURN, reply.body[0]
assert reply.signature == 's'
assert reply.body == body
body = ['hello', 'world']
reply = await call('echo_multiple', 'ss', body)
assert reply.message_type == MessageType.METHOD_RETURN, reply.body[0]
assert reply.signature == 'ss'
assert reply.body == body
body = [['hello', 'world'],
Variant('v', Variant('(ss)', ['hello', 'world'])), {
'foo': Variant('t', 100)
}, ['one', ['two', [Variant('s', 'three')]]]]
signature = 'asva{sv}(s(s(v)))'
SignatureTree(signature).verify(body)
reply = await call('echo_containers', signature, body)
assert reply.message_type == MessageType.METHOD_RETURN, reply.body[0]
assert reply.signature == signature
assert reply.body == body
reply = await call('ping')
assert reply.message_type == MessageType.METHOD_RETURN, reply.body[0]
assert reply.signature == ''
assert reply.body == []
reply = await call('throws_unexpected_error')
assert reply.message_type == MessageType.ERROR, reply.body[0]
assert reply.error_name == ErrorType.SERVICE_ERROR.value, reply.body[0]
reply = await call('throws_dbus_error')
assert reply.message_type == MessageType.ERROR, reply.body[0]
assert reply.error_name == 'test.error', reply.body[0]
assert reply.body == ['an error ocurred']
reply = await call('ping', flags=MessageFlag.NO_REPLY_EXPECTED)
assert reply is None
reply = await call('throws_unexpected_error', flags=MessageFlag.NO_REPLY_EXPECTED)
assert reply is None
reply = await call('throws_dbus_error', flags=MessageFlag.NO_REPLY_EXPECTED)
assert reply is None
|
