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
|
from dbus_next.message import MessageFlag
from dbus_next.service import ServiceInterface, method
import dbus_next.introspection as intr
from dbus_next import aio, glib, DBusError
from test.util import check_gi_repository, skip_reason_no_gi
import pytest
has_gi = check_gi_repository()
class ExampleInterface(ServiceInterface):
def __init__(self):
super().__init__('test.interface')
@method()
def Ping(self):
pass
@method()
def EchoInt64(self, what: 'x') -> 'x':
return what
@method()
def EchoString(self, what: 's') -> 's':
return what
@method()
def ConcatStrings(self, what1: 's', what2: 's') -> 's':
return what1 + what2
@method()
def EchoThree(self, what1: 's', what2: 's', what3: 's') -> 'sss':
return [what1, what2, what3]
@method()
def ThrowsError(self):
raise DBusError('test.error', 'something went wrong')
@pytest.mark.asyncio
async def test_aio_proxy_object():
bus_name = 'aio.client.test.methods'
bus = await aio.MessageBus().connect()
bus2 = await aio.MessageBus().connect()
await bus.request_name(bus_name)
service_interface = ExampleInterface()
bus.export('/test/path', service_interface)
# add some more to test nodes
bus.export('/test/path/child1', ExampleInterface())
bus.export('/test/path/child2', ExampleInterface())
introspection = await bus2.introspect(bus_name, '/test/path')
assert type(introspection) is intr.Node
obj = bus2.get_proxy_object(bus_name, '/test/path', introspection)
interface = obj.get_interface(service_interface.name)
children = obj.get_children()
assert len(children) == 2
for child in obj.get_children():
assert type(child) is aio.ProxyObject
result = await interface.call_ping()
assert result is None
result = await interface.call_echo_string('hello')
assert result == 'hello'
result = await interface.call_concat_strings('hello ', 'world')
assert result == 'hello world'
result = await interface.call_echo_three('hello', 'there', 'world')
assert result == ['hello', 'there', 'world']
result = await interface.call_echo_int64(-10000)
assert result == -10000
result = await interface.call_echo_string('no reply', flags=MessageFlag.NO_REPLY_EXPECTED)
assert result is None
with pytest.raises(DBusError):
try:
await interface.call_throws_error()
except DBusError as e:
assert e.reply is not None
assert e.type == 'test.error'
assert e.text == 'something went wrong'
raise e
bus.disconnect()
bus2.disconnect()
@pytest.mark.skipif(not has_gi, reason=skip_reason_no_gi)
def test_glib_proxy_object():
bus_name = 'glib.client.test.methods'
bus = glib.MessageBus().connect_sync()
bus.request_name_sync(bus_name)
service_interface = ExampleInterface()
bus.export('/test/path', service_interface)
bus2 = glib.MessageBus().connect_sync()
introspection = bus2.introspect_sync(bus_name, '/test/path')
assert type(introspection) is intr.Node
obj = bus.get_proxy_object(bus_name, '/test/path', introspection)
interface = obj.get_interface(service_interface.name)
result = interface.call_ping_sync()
assert result is None
result = interface.call_echo_string_sync('hello')
assert result == 'hello'
result = interface.call_concat_strings_sync('hello ', 'world')
assert result == 'hello world'
result = interface.call_echo_three_sync('hello', 'there', 'world')
assert result == ['hello', 'there', 'world']
with pytest.raises(DBusError):
try:
result = interface.call_throws_error_sync()
assert False, result
except DBusError as e:
assert e.reply is not None
assert e.type == 'test.error'
assert e.text == 'something went wrong'
raise e
bus.disconnect()
bus2.disconnect()
|
