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
167
168
|
import asyncio
import pytest
from dbus_fast import Message, MessageType
from dbus_fast import introspection as intr
from dbus_fast.aio import MessageBus
from dbus_fast.service import ServiceInterface, dbus_method
standard_interfaces_count = len(intr.Node.default().interfaces)
class ExampleInterface(ServiceInterface):
def __init__(self, name):
self._method_called = False
super().__init__(name)
@dbus_method()
def some_method(self):
self._method_called = True
@pytest.mark.asyncio
async def test_export_unexport():
interface = ExampleInterface("test.interface")
interface2 = ExampleInterface("test.interface2")
export_path = "/test/path"
export_path2 = "/test/path/child"
bus = await MessageBus().connect()
bus.export(export_path, interface)
with pytest.raises(ValueError):
# Already exported
bus.export(export_path, interface)
assert export_path in bus._path_exports
assert len(bus._path_exports[export_path]) == 1
assert bus._path_exports[export_path][interface.name] is interface
assert len(ServiceInterface._get_buses(interface)) == 1
bus.export(export_path2, interface2)
node = bus._introspect_export_path(export_path)
assert len(node.interfaces) == standard_interfaces_count + 1
assert len(node.nodes) == 1
# relative path
assert node.nodes[0].name == "child"
bus.unexport(export_path, interface)
assert export_path not in bus._path_exports
assert len(ServiceInterface._get_buses(interface)) == 0
bus.export(export_path2, interface)
assert len(bus._path_exports[export_path2]) == 2
# test unexporting the whole path
bus.unexport(export_path2)
assert not bus._path_exports
assert not ServiceInterface._get_buses(interface)
assert not ServiceInterface._get_buses(interface2)
# test unexporting by name
bus.export(export_path, interface)
bus.unexport(export_path, interface.name)
assert not bus._path_exports
assert not ServiceInterface._get_buses(interface)
# test unexporting by ServiceInterface
bus.export(export_path, interface)
bus.unexport(export_path, interface)
assert not bus._path_exports
assert not ServiceInterface._get_buses(interface)
with pytest.raises(TypeError):
bus.unexport(export_path, object())
node = bus._introspect_export_path("/path/doesnt/exist")
assert type(node) is intr.Node
assert not node.interfaces
assert not node.nodes
# Should to nothing
bus.unexport("/path/doesnt/exist", interface)
bus.disconnect()
await asyncio.wait_for(bus.wait_for_disconnect(), timeout=1)
@pytest.mark.asyncio
async def test_export_twice_raises():
bus = await MessageBus().connect()
interface = ExampleInterface("test.interface")
export_path = "/test/path"
export_path2 = "/test/path/child"
bus.export(export_path, interface)
with pytest.raises(
ValueError, match="instance cannot be added to the same bus twice"
):
bus.export(export_path2, interface)
bus.disconnect()
await asyncio.wait_for(bus.wait_for_disconnect(), timeout=1)
@pytest.mark.asyncio
async def test_export_alias():
bus = await MessageBus().connect()
interface = ExampleInterface("test.interface")
interface2 = ExampleInterface("test.interface")
export_path = "/test/path"
export_path2 = "/test/path/child"
bus.export(export_path, interface)
bus.export(export_path2, interface2)
result = await bus.call(
Message(
destination=bus.unique_name,
path=export_path,
interface="test.interface",
member="some_method",
)
)
assert result.message_type is MessageType.METHOD_RETURN, result.body[0]
assert interface._method_called
interface._method_called = False
result = await bus.call(
Message(
destination=bus.unique_name,
path=export_path2,
interface="test.interface",
member="some_method",
)
)
assert result.message_type is MessageType.METHOD_RETURN, result.body[0]
assert interface2._method_called
bus.disconnect()
await asyncio.wait_for(bus.wait_for_disconnect(), timeout=1)
@pytest.mark.asyncio
async def test_export_introspection():
interface = ExampleInterface("test.interface")
interface2 = ExampleInterface("test.interface2")
export_path = "/test/path"
export_path2 = "/test/path/child"
bus = await MessageBus().connect()
bus.export(export_path, interface)
bus.export(export_path2, interface2)
root = bus._introspect_export_path("/")
assert len(root.nodes) == 1
bus.disconnect()
await asyncio.wait_for(bus.wait_for_disconnect(), timeout=1)
|
