1
|
|
|
"""Module to TopoController.""" |
2
|
|
|
|
3
|
|
|
from unittest import TestCase |
4
|
|
|
from unittest.mock import MagicMock, patch |
5
|
|
|
|
6
|
|
|
from pymongo.operations import UpdateOne |
7
|
|
|
|
8
|
|
|
from napps.kytos.topology.controllers import TopoController |
9
|
|
|
from napps.kytos.topology.db.client import bootstrap_index |
10
|
|
|
from napps.kytos.topology.db.models import LinkDoc, SwitchDoc |
11
|
|
|
|
12
|
|
|
|
13
|
|
|
class TestTopoController(TestCase): # pylint: disable=too-many-public-methods |
14
|
|
|
"""Test the Main class.""" |
15
|
|
|
|
16
|
|
|
def setUp(self) -> None: |
17
|
|
|
"""Execute steps before each tests.""" |
18
|
|
|
self.topo = TopoController(MagicMock()) |
19
|
|
|
self.dpid = "00:00:00:00:00:00:00:01" |
20
|
|
|
self.interface_id = f"{self.dpid}:1" |
21
|
|
|
self.link_id = "some_id" |
22
|
|
|
|
23
|
|
|
@patch("napps.kytos.topology.controllers.bootstrap_index") |
24
|
|
|
def test_boostrap_indexes(self, mock) -> None: |
25
|
|
|
"""Test_boostrap_indexes.""" |
26
|
|
|
self.topo.bootstrap_indexes() |
27
|
|
|
|
28
|
|
|
expected_indexes = [ |
29
|
|
|
("switches", "interfaces.id", 1), |
30
|
|
|
("links", "endpoints.id", 1), |
31
|
|
|
] |
32
|
|
|
indexes = [(v[0][1], v[0][2], v[0][3]) for v in mock.call_args_list] |
33
|
|
|
assert expected_indexes == indexes |
34
|
|
|
|
35
|
|
|
@staticmethod |
36
|
|
|
def test_boostrap_index_existent() -> None: |
37
|
|
|
"""test_boostrap_index. |
38
|
|
|
|
39
|
|
|
This test will be moved to kytos core onde db client moves too. |
40
|
|
|
""" |
41
|
|
|
db_mock = MagicMock() |
42
|
|
|
db_mock["napps"].index_information.return_value = { |
43
|
|
|
"_id_": {"v": 2, "key": [("_id", 1)]}, |
44
|
|
|
"interfaces.id_1": {"v": 2, "key": [("interfaces.id", 1)]}, |
45
|
|
|
} |
46
|
|
|
bootstrap_index(db_mock, "napps", "interfaces.id", 1) |
47
|
|
|
assert db_mock["napps"].create_index.call_count == 0 |
48
|
|
|
|
49
|
|
|
@staticmethod |
50
|
|
|
def test_boostrap_index() -> None: |
51
|
|
|
"""test_boostrap_index. |
52
|
|
|
|
53
|
|
|
This test will be moved to kytos core onde db client moves too. |
54
|
|
|
""" |
55
|
|
|
db_mock = MagicMock() |
56
|
|
|
db_mock["napps"].index_information.return_value = {} |
57
|
|
|
bootstrap_index(db_mock, "switches", "interfaces.id", 1) |
58
|
|
|
assert db_mock["napps"].create_index.call_count == 1 |
59
|
|
|
|
60
|
|
|
def test_get_topology(self) -> None: |
61
|
|
|
"""Test_get_topology.""" |
62
|
|
|
self.topo.get_switches = MagicMock() |
63
|
|
|
self.topo.get_links = MagicMock() |
64
|
|
|
assert "topology" in self.topo.get_topology() |
65
|
|
|
assert self.topo.get_switches.call_count == 1 |
66
|
|
|
assert self.topo.get_links.call_count == 1 |
67
|
|
|
|
68
|
|
|
def test_get_links(self) -> None: |
69
|
|
|
"""test_get_links.""" |
70
|
|
|
assert "links" in self.topo.get_links() |
71
|
|
|
assert self.topo.db.links.aggregate.call_count == 1 |
72
|
|
|
arg = self.topo.db.links.aggregate.call_args[0] |
73
|
|
|
assert arg[0] == [{"$sort": {"_id": 1}}, |
74
|
|
|
{"$project": LinkDoc.projection()}] |
75
|
|
|
|
76
|
|
|
def test_get_switches(self) -> None: |
77
|
|
|
"""test_get_switches.""" |
78
|
|
|
assert "switches" in self.topo.get_switches() |
79
|
|
|
assert self.topo.db.switches.aggregate.call_count == 1 |
80
|
|
|
arg = self.topo.db.switches.aggregate.call_args[0] |
81
|
|
|
assert arg[0] == [ |
82
|
|
|
{"$sort": {"_id": 1}}, |
83
|
|
|
{"$project": SwitchDoc.projection()}, |
84
|
|
|
] |
85
|
|
|
|
86
|
|
|
def test_get_interfaces(self) -> None: |
87
|
|
|
"""test_get_interfaces.""" |
88
|
|
|
assert "interfaces" in self.topo.get_interfaces() |
89
|
|
|
assert self.topo.db.switches.aggregate.call_count == 1 |
90
|
|
|
arg = self.topo.db.switches.aggregate.call_args[0] |
91
|
|
|
assert arg[0] == [ |
92
|
|
|
{"$sort": {"_id": 1}}, |
93
|
|
|
{"$project": {"interfaces": 1, "_id": 0}}, |
94
|
|
|
{"$unwind": "$interfaces"}, |
95
|
|
|
{"$replaceRoot": {"newRoot": "$interfaces"}}, |
96
|
|
|
] |
97
|
|
|
|
98
|
|
|
def test_enable_switch(self) -> None: |
99
|
|
|
"""test_enable_switch.""" |
100
|
|
|
self.topo.enable_switch(self.dpid) |
101
|
|
|
|
102
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
103
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
104
|
|
|
assert arg1 == {"_id": self.dpid} |
105
|
|
|
assert arg2["$set"]["enabled"] |
106
|
|
|
|
107
|
|
|
def test_disable_switch(self) -> None: |
108
|
|
|
"""test_disable_switch.""" |
109
|
|
|
self.topo.disable_switch(self.dpid) |
110
|
|
|
|
111
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
112
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
113
|
|
|
assert arg1 == {"_id": self.dpid} |
114
|
|
|
assert not arg2["$set"]["enabled"] |
115
|
|
|
|
116
|
|
|
def test_deactivate_switch(self) -> None: |
117
|
|
|
"""test_deactivate_switch.""" |
118
|
|
|
self.topo.deactivate_switch(self.dpid) |
119
|
|
|
|
120
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
121
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
122
|
|
|
assert arg1 == {"_id": self.dpid} |
123
|
|
|
assert not arg2["$set"]["active"] |
124
|
|
|
|
125
|
|
|
def test_add_switch_metadata(self) -> None: |
126
|
|
|
"""test_add_switch_metadata.""" |
127
|
|
|
metadata = {"some": "value"} |
128
|
|
|
self.topo.add_switch_metadata(self.dpid, metadata) |
129
|
|
|
|
130
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
131
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
132
|
|
|
assert arg1 == {"_id": self.dpid} |
133
|
|
|
assert arg2["$set"]["metadata.some"] == "value" |
134
|
|
|
|
135
|
|
|
def test_delete_switch_metadata(self) -> None: |
136
|
|
|
"""test_delete_switch_metadata.""" |
137
|
|
|
key = "some" |
138
|
|
|
self.topo.delete_switch_metadata_key(self.dpid, key) |
139
|
|
|
|
140
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
141
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
142
|
|
|
assert arg1 == {"_id": self.dpid} |
143
|
|
|
assert arg2["$unset"][f"metadata.{key}"] == "" |
144
|
|
|
|
145
|
|
|
def test_enable_interface(self) -> None: |
146
|
|
|
"""test_enable_interface.""" |
147
|
|
|
self.topo.enable_interface(self.interface_id) |
148
|
|
|
|
149
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
150
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
151
|
|
|
assert arg1 == {"interfaces.id": self.interface_id} |
152
|
|
|
assert arg2["$set"]["interfaces.$.enabled"] |
153
|
|
|
|
154
|
|
|
def test_disable_interface(self) -> None: |
155
|
|
|
"""test_disable_interface.""" |
156
|
|
|
self.topo.disable_interface(self.interface_id) |
157
|
|
|
|
158
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
159
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
160
|
|
|
assert arg1 == {"interfaces.id": self.interface_id} |
161
|
|
|
assert not arg2["$set"]["interfaces.$.enabled"] |
162
|
|
|
|
163
|
|
|
def test_activate_interface(self) -> None: |
164
|
|
|
"""test_activate_interface.""" |
165
|
|
|
self.topo.activate_interface(self.interface_id) |
166
|
|
|
|
167
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
168
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
169
|
|
|
assert arg1 == {"interfaces.id": self.interface_id} |
170
|
|
|
assert arg2["$set"]["interfaces.$.active"] |
171
|
|
|
|
172
|
|
|
def test_deactivate_interface(self) -> None: |
173
|
|
|
"""test_deactivate_interface.""" |
174
|
|
|
self.topo.deactivate_interface(self.interface_id) |
175
|
|
|
|
176
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
177
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
178
|
|
|
assert arg1 == {"interfaces.id": self.interface_id} |
179
|
|
|
assert not arg2["$set"]["interfaces.$.active"] |
180
|
|
|
|
181
|
|
|
def test_enable_interface_lldp(self) -> None: |
182
|
|
|
"""test_enable_interface_lldp.""" |
183
|
|
|
self.topo.enable_interface_lldp(self.interface_id) |
184
|
|
|
|
185
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
186
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
187
|
|
|
assert arg1 == {"interfaces.id": self.interface_id} |
188
|
|
|
assert arg2["$set"]["interfaces.$.lldp"] |
189
|
|
|
|
190
|
|
|
def test_disable_interface_lldp(self) -> None: |
191
|
|
|
"""test_disable_interface_lldp.""" |
192
|
|
|
self.topo.disable_interface_lldp(self.interface_id) |
193
|
|
|
|
194
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
195
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
196
|
|
|
assert arg1 == {"interfaces.id": self.interface_id} |
197
|
|
|
assert not arg2["$set"]["interfaces.$.lldp"] |
198
|
|
|
|
199
|
|
|
def test_add_interface_metadata(self) -> None: |
200
|
|
|
"""test_add_interface_metadata.""" |
201
|
|
|
metadata = {"some": "value"} |
202
|
|
|
self.topo.add_interface_metadata(self.interface_id, metadata) |
203
|
|
|
|
204
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
205
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
206
|
|
|
assert arg1 == {"interfaces.id": self.interface_id} |
207
|
|
|
assert arg2["$set"]["interfaces.$.metadata.some"] == "value" |
208
|
|
|
|
209
|
|
|
def test_delete_interface_metadata_key(self) -> None: |
210
|
|
|
"""test_delete_interface_metadata.""" |
211
|
|
|
key = "some" |
212
|
|
|
self.topo.delete_interface_metadata_key(self.interface_id, key) |
213
|
|
|
|
214
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
215
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
216
|
|
|
assert arg1 == {"interfaces.id": self.interface_id} |
217
|
|
|
assert arg2["$unset"][f"interfaces.$.metadata.{key}"] == "" |
218
|
|
|
|
219
|
|
|
def test_enable_link(self) -> None: |
220
|
|
|
"""test_enable_link.""" |
221
|
|
|
self.topo.enable_link(self.link_id) |
222
|
|
|
|
223
|
|
|
self.topo.db.links.find_one_and_update.assert_called() |
224
|
|
|
arg1, arg2 = self.topo.db.links.find_one_and_update.call_args[0] |
225
|
|
|
assert arg1 == {"_id": self.link_id} |
226
|
|
|
assert arg2["$set"]["enabled"] |
227
|
|
|
|
228
|
|
|
def test_disable_link(self) -> None: |
229
|
|
|
"""test_disable_link.""" |
230
|
|
|
self.topo.disable_link(self.link_id) |
231
|
|
|
|
232
|
|
|
self.topo.db.links.find_one_and_update.assert_called() |
233
|
|
|
arg1, arg2 = self.topo.db.links.find_one_and_update.call_args[0] |
234
|
|
|
assert arg1 == {"_id": self.link_id} |
235
|
|
|
assert not arg2["$set"]["enabled"] |
236
|
|
|
|
237
|
|
|
def test_add_link_metadata(self) -> None: |
238
|
|
|
"""test_add_link_metadata.""" |
239
|
|
|
key = "some_key" |
240
|
|
|
value = "some_value" |
241
|
|
|
self.topo.add_link_metadata(self.link_id, {key: value}) |
242
|
|
|
|
243
|
|
|
self.topo.db.links.find_one_and_update.assert_called() |
244
|
|
|
arg1, arg2 = self.topo.db.links.find_one_and_update.call_args[0] |
245
|
|
|
assert arg1 == {"_id": self.link_id} |
246
|
|
|
assert arg2["$set"][f"metadata.{key}"] == value |
247
|
|
|
|
248
|
|
|
def test_delete_link_metadata_key(self) -> None: |
249
|
|
|
"""test_delete_link_metadata_key.""" |
250
|
|
|
key = "some_key" |
251
|
|
|
self.topo.delete_link_metadata_key(self.link_id, key) |
252
|
|
|
|
253
|
|
|
self.topo.db.links.find_one_and_update.assert_called() |
254
|
|
|
arg1, arg2 = self.topo.db.links.find_one_and_update.call_args[0] |
255
|
|
|
assert arg1 == {"_id": self.link_id} |
256
|
|
|
assert arg2["$unset"][f"metadata.{key}"] == "" |
257
|
|
|
|
258
|
|
|
def test_get_interfaces_details(self) -> None: |
259
|
|
|
"""test_get_insterfaces_details.""" |
260
|
|
|
interfaces_ids = ["1", "2", "3"] |
261
|
|
|
self.topo.get_interfaces_details(interfaces_ids) |
262
|
|
|
self.topo.db.interface_details.aggregate.assert_called_with( |
263
|
|
|
[{"$match": {"_id": {"$in": interfaces_ids}}}] |
264
|
|
|
) |
265
|
|
|
|
266
|
|
|
def test_bulk_upsert_interface_details(self) -> None: |
267
|
|
|
"""test_bulk_upsert_interface_details.""" |
268
|
|
|
ids_details = [ |
269
|
|
|
("1", {"_id": "1", "available_vlans": [1]}), |
270
|
|
|
("2", {"_id": "2", "available_vlans": [2]}), |
271
|
|
|
] |
272
|
|
|
self.topo.bulk_upsert_interface_details(ids_details) |
273
|
|
|
|
274
|
|
|
self.topo.db_client.start_session.assert_called() |
275
|
|
|
arg1 = self.topo.db.interface_details.bulk_write.call_args[0] |
276
|
|
|
assert len(arg1[0]) == len(ids_details) |
277
|
|
|
for item in arg1[0]: |
278
|
|
|
assert isinstance(item, UpdateOne) |
279
|
|
|
|
280
|
|
|
def test_upsert_switch(self) -> None: |
281
|
|
|
"""test_upsert_switch.""" |
282
|
|
|
switch_dict = {"enabled": True, "active": True, "_id": self.dpid} |
283
|
|
|
self.topo.upsert_switch(self.dpid, switch_dict) |
284
|
|
|
|
285
|
|
|
self.topo.db.switches.find_one_and_update.assert_called() |
286
|
|
|
arg1, arg2 = self.topo.db.switches.find_one_and_update.call_args[0] |
287
|
|
|
assert arg1 == {"_id": self.dpid} |
288
|
|
|
for key, value in switch_dict.items(): |
289
|
|
|
assert arg2["$set"][key] == value |
290
|
|
|
|
291
|
|
|
def test_upsert_link(self) -> None: |
292
|
|
|
"""test_upsert_link.""" |
293
|
|
|
link_dict = { |
294
|
|
|
"_id": self.link_id, |
295
|
|
|
"enabled": True, |
296
|
|
|
"active": True, |
297
|
|
|
"endpoint_a": {"id": "00:00:00:00:00:00:00:01:1"}, |
298
|
|
|
"endpoint_b": {"id": "00:00:00:00:00:00:00:02:01"}, |
299
|
|
|
} |
300
|
|
|
self.topo.upsert_link(self.link_id, link_dict) |
301
|
|
|
assert self.topo.db.switches.find_one_and_update.call_count == 2 |
302
|
|
|
assert self.topo.db.links.find_one_and_update.call_count == 1 |
303
|
|
|
|