1
|
|
|
"""Module to test the KytosGraph in graph.py.""" |
2
|
|
|
|
3
|
|
|
# pylint: disable=too-many-public-methods, import-error |
4
|
1 |
|
from tests.integration.metadata_settings import MetadataSettings |
5
|
|
|
|
6
|
|
|
|
7
|
1 |
|
class TestPathsMetadata(MetadataSettings): |
8
|
|
|
"""Tests for the graph class. |
9
|
|
|
|
10
|
|
|
Tests if the metadata in search paths edges have passing values. |
11
|
|
|
""" |
12
|
|
|
|
13
|
1 |
|
def test_path_constrained_user_user_k1(self): |
14
|
|
|
"""Test if there is a constrained path between User - User.""" |
15
|
1 |
|
self.initializer() |
16
|
|
|
|
17
|
1 |
|
source = "User1" |
18
|
1 |
|
destination = "User2" |
19
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
20
|
|
|
source, destination, k=1 |
21
|
|
|
) |
22
|
1 |
|
assert len(paths) == 1 |
23
|
|
|
|
24
|
1 |
|
for path in paths: |
25
|
1 |
|
assert path["hops"][0] == source |
26
|
1 |
|
assert path["hops"][-1] == destination |
27
|
|
|
|
28
|
1 |
|
def test_path_constrained_user_user_k2(self): |
29
|
|
|
"""Test if there are two constrained path between User - User.""" |
30
|
1 |
|
self.initializer() |
31
|
|
|
|
32
|
1 |
|
source = "User1" |
33
|
1 |
|
destination = "User2" |
34
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
35
|
|
|
source, destination, k=2 |
36
|
|
|
) |
37
|
1 |
|
assert len(paths) == 2 |
38
|
|
|
|
39
|
1 |
|
for path in paths: |
40
|
1 |
|
assert path["hops"][0] == source |
41
|
1 |
|
assert path["hops"][-1] == destination |
42
|
|
|
|
43
|
1 |
|
def test_path_constrained_user_user_k4(self): |
44
|
|
|
"""Test if there are four constrained path between User - User.""" |
45
|
1 |
|
self.initializer() |
46
|
|
|
|
47
|
1 |
|
source = "User1" |
48
|
1 |
|
destination = "User2" |
49
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
50
|
|
|
source, destination, k=4 |
51
|
|
|
) |
52
|
1 |
|
assert len(paths) == 4 |
53
|
|
|
|
54
|
1 |
|
for path in paths: |
55
|
1 |
|
assert path["hops"][0] == source |
56
|
1 |
|
assert path["hops"][-1] == destination |
57
|
|
|
|
58
|
1 |
|
def test_path_constrained_user_switch(self): |
59
|
|
|
"""Test if there is a constrained |
60
|
|
|
path between User - Switch.""" |
61
|
1 |
|
self.initializer() |
62
|
|
|
|
63
|
1 |
|
source = "User1" |
64
|
1 |
|
destination = "S4" |
65
|
1 |
|
paths = self.graph.constrained_k_shortest_paths(source, destination) |
66
|
1 |
|
assert paths |
67
|
|
|
|
68
|
1 |
|
for path in paths: |
69
|
1 |
|
assert path["hops"][0] == source |
70
|
1 |
|
assert path["hops"][-1] == destination |
71
|
|
|
|
72
|
1 |
|
def test_path_constrained_switch_switch(self): |
73
|
|
|
"""Test if there is a constrained |
74
|
|
|
path between Switch - Switch.""" |
75
|
1 |
|
self.initializer() |
76
|
|
|
|
77
|
1 |
|
source = "S2" |
78
|
1 |
|
destination = "S4" |
79
|
1 |
|
paths = self.graph.constrained_k_shortest_paths(source, destination) |
80
|
1 |
|
assert paths |
81
|
|
|
|
82
|
1 |
|
for path in paths: |
83
|
1 |
|
assert path["hops"][0] == source |
84
|
1 |
|
assert path["hops"][-1] == destination |
85
|
|
|
|
86
|
1 |
|
def test_no_path_constrained_user_user(self): |
87
|
|
|
"""Test if there is NOT a constrained |
88
|
|
|
path between User - User.""" |
89
|
1 |
|
self.initializer() |
90
|
1 |
|
paths = self.graph.constrained_k_shortest_paths("User1", "User3") |
91
|
1 |
|
assert not paths |
92
|
|
|
|
93
|
1 |
|
def test_path_constrained_user_user_t1(self): |
94
|
|
|
"""Test if there is a constrained path between |
95
|
|
|
User - User using the 2nd topology variant.""" |
96
|
1 |
|
self.initializer(val=1) |
97
|
|
|
|
98
|
1 |
|
source = "User1" |
99
|
1 |
|
destination = "User3" |
100
|
1 |
|
paths = self.graph.constrained_k_shortest_paths(source, destination) |
101
|
1 |
|
assert paths |
102
|
|
|
|
103
|
1 |
|
for path in paths: |
104
|
1 |
|
assert path["hops"][0] == source |
105
|
1 |
|
assert path["hops"][-1] == destination |
106
|
|
|
|
107
|
1 |
|
def test_no_path_constrained_user_user_t1(self): |
108
|
|
|
"""Test if there is NOT a constrained path between |
109
|
|
|
User - User using the 2nd topology variant.""" |
110
|
1 |
|
self.initializer(val=1) |
111
|
1 |
|
paths = self.graph.constrained_k_shortest_paths("User1", "User2") |
112
|
1 |
|
assert not paths |
113
|
|
|
|
114
|
1 |
|
def test_no_path_constrained_switch_switch_t1(self): |
115
|
|
|
"""Test if there is NOT a constrained path between |
116
|
|
|
Switch - Switch using the 2nd topology variant.""" |
117
|
1 |
|
self.initializer(val=1) |
118
|
1 |
|
paths = self.graph.constrained_k_shortest_paths("S1", "S2") |
119
|
1 |
|
assert not paths |
120
|
|
|
|
121
|
1 |
|
def test_path_constrained_user_user_t2(self): |
122
|
|
|
"""Test if there is a constrained path between |
123
|
|
|
User - User using the 3rd topology variant.""" |
124
|
1 |
|
self.initializer(val=2) |
125
|
|
|
|
126
|
1 |
|
source = "User1" |
127
|
1 |
|
destination = "User2" |
128
|
1 |
|
paths = self.graph.constrained_k_shortest_paths(source, destination) |
129
|
1 |
|
assert paths |
130
|
|
|
|
131
|
1 |
|
for path in paths: |
132
|
1 |
|
assert path["hops"][0] == source |
133
|
1 |
|
assert path["hops"][-1] == destination |
134
|
|
|
|
135
|
1 |
|
def test_path_constrained_user_switch_t2(self): |
136
|
|
|
"""Test if there is a constrained path between |
137
|
|
|
User - Switch using the 3rd topology variant.""" |
138
|
1 |
|
self.initializer(val=2) |
139
|
|
|
|
140
|
1 |
|
source = "User1" |
141
|
1 |
|
destination = "S4" |
142
|
1 |
|
paths = self.graph.constrained_k_shortest_paths(source, destination) |
143
|
1 |
|
assert paths |
144
|
|
|
|
145
|
1 |
|
for path in paths: |
146
|
1 |
|
assert path["hops"][0] == source |
147
|
1 |
|
assert path["hops"][-1] == destination |
148
|
1 |
|
paths = self.graph.constrained_k_shortest_paths("User1", "S4") |
149
|
|
|
|
150
|
1 |
|
def test_path_constrained_switch_switch_t2(self): |
151
|
|
|
"""Test if there is a constrained path between |
152
|
|
|
two switches using the 3rd topology variant.""" |
153
|
1 |
|
self.initializer(val=2) |
154
|
|
|
|
155
|
1 |
|
source = "S2" |
156
|
1 |
|
destination = "S4" |
157
|
1 |
|
paths = self.graph.constrained_k_shortest_paths(source, destination) |
158
|
1 |
|
assert paths |
159
|
|
|
|
160
|
1 |
|
for path in paths: |
161
|
1 |
|
assert path["hops"][0] == source |
162
|
1 |
|
assert path["hops"][-1] == destination |
163
|
|
|
|
164
|
1 |
|
def test_path_constrained_reliability(self): |
165
|
|
|
"""Tests if the edges used in the paths |
166
|
|
|
of the paths set do not have poor reliability |
167
|
|
|
""" |
168
|
1 |
|
requirements = {"reliability": 3} |
169
|
|
|
|
170
|
1 |
|
self.initializer() |
171
|
|
|
|
172
|
1 |
|
source = "User1" |
173
|
1 |
|
destination = "User2" |
174
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
175
|
|
|
source, destination, mandatory_metrics=requirements |
176
|
|
|
) |
177
|
1 |
|
assert paths |
178
|
|
|
|
179
|
1 |
|
for path in paths: |
180
|
1 |
|
assert path["hops"][0] == source |
181
|
1 |
|
assert path["hops"][-1] == destination |
182
|
|
|
|
183
|
1 |
|
def test_cspf_with_multiple_owners(self): |
184
|
|
|
"""Tests if the edges with multiple owners""" |
185
|
|
|
|
186
|
1 |
|
owners = ("B", "C") |
187
|
1 |
|
owners_paths = [] |
188
|
1 |
|
for owner in owners: |
189
|
1 |
|
requirements = {"ownership": owner} |
190
|
|
|
|
191
|
1 |
|
self.initializer() |
192
|
|
|
|
193
|
1 |
|
source = "User1" |
194
|
1 |
|
destination = "User2" |
195
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
196
|
|
|
source, destination, mandatory_metrics=requirements, k=1 |
197
|
|
|
) |
198
|
1 |
|
assert paths |
199
|
1 |
|
assert paths[0]["hops"][0] == source |
200
|
1 |
|
assert paths[0]["hops"][-1] == destination |
201
|
1 |
|
assert paths[0]["metrics"] == requirements |
202
|
1 |
|
owners_paths.append(paths[0]["hops"]) |
203
|
1 |
|
assert owners_paths[0] == owners_paths[1] |
204
|
|
|
|
205
|
1 |
|
def test_no_path_constrained_reliability(self): |
206
|
|
|
"""Tests if the edges used in the paths |
207
|
|
|
of the paths set do not have poor reliability |
208
|
|
|
""" |
209
|
1 |
|
requirements = {"reliability": 1} |
210
|
|
|
|
211
|
1 |
|
self.initializer() |
212
|
|
|
|
213
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
214
|
|
|
"User1", "User3", mandatory_metrics=requirements |
215
|
|
|
) |
216
|
1 |
|
assert not paths |
217
|
|
|
|
218
|
1 |
|
def test_path_constrained_reliability_detailed(self): |
219
|
|
|
"""Tests if the edges used in the paths |
220
|
|
|
of the paths set do not have poor reliability |
221
|
|
|
""" |
222
|
1 |
|
reliabilities = [] |
223
|
1 |
|
requirements = {"reliability": 3} |
224
|
1 |
|
poor_reliability = 1 |
225
|
|
|
|
226
|
1 |
|
self.initializer() |
227
|
|
|
|
228
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
229
|
|
|
"User1", "User2", mandatory_metrics=requirements |
230
|
|
|
) |
231
|
|
|
|
232
|
1 |
|
if paths: |
233
|
1 |
|
for path in paths[0]["hops"]: |
234
|
1 |
|
for i in range(1, len(path)): |
235
|
1 |
|
endpoint_a = path[i - 1] |
236
|
1 |
|
endpoint_b = path[i] |
237
|
1 |
|
meta_data = self.graph.get_link_metadata( |
238
|
|
|
endpoint_a, endpoint_b |
239
|
|
|
) |
240
|
1 |
|
if meta_data and "reliability" in meta_data.keys(): |
241
|
|
|
reliabilities.append(meta_data["reliability"]) |
242
|
|
|
|
243
|
1 |
|
assert poor_reliability not in reliabilities |
244
|
|
|
|
245
|
|
|
else: |
246
|
|
|
assert paths |
247
|
|
|
|
248
|
1 |
|
def test_path_constrained_delay(self): |
249
|
|
|
"""Tests if the edges used in the paths |
250
|
|
|
from User 1 to User 2 have less than 30 delay. |
251
|
|
|
""" |
252
|
1 |
|
delays = [] |
253
|
1 |
|
requirements = {"delay": 29} |
254
|
|
|
|
255
|
1 |
|
self.initializer() |
256
|
|
|
|
257
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
258
|
|
|
"User1", "User2", mandatory_metrics=requirements |
259
|
|
|
) |
260
|
1 |
|
assert paths |
261
|
|
|
|
262
|
1 |
|
for path in paths: |
263
|
1 |
|
for i, j in zip( |
264
|
|
|
range(0, len(path["hops"])), range(1, len(path["hops"])) |
265
|
|
|
): |
266
|
1 |
|
endpoint_a = path["hops"][i] |
267
|
1 |
|
endpoint_b = path["hops"][j] |
268
|
1 |
|
meta_data = self.graph.get_link_metadata( |
269
|
|
|
endpoint_a, endpoint_b |
270
|
|
|
) |
271
|
1 |
|
if meta_data and "delay" in meta_data.keys(): |
272
|
1 |
|
delays.append(meta_data["delay"]) |
273
|
|
|
|
274
|
1 |
|
assert delays |
275
|
1 |
|
for delay in delays: |
276
|
1 |
|
assert delay <= requirements["delay"] |
277
|
|
|
|
278
|
1 |
|
def links_metadata_values(self, path, attr): |
279
|
|
|
"""Method to build a list of metadata values of the links of a path""" |
280
|
1 |
|
values = [] |
281
|
1 |
|
for i, j in zip( |
282
|
|
|
range(0, len(path["hops"])), range(1, len(path["hops"])) |
283
|
|
|
): |
284
|
1 |
|
endpoint_a = path["hops"][i] |
285
|
1 |
|
endpoint_b = path["hops"][j] |
286
|
1 |
|
meta_data = self.graph.get_link_metadata(endpoint_a, endpoint_b) |
287
|
1 |
|
if meta_data and attr in meta_data.keys(): |
288
|
1 |
|
values.append(meta_data[attr]) |
289
|
1 |
|
return values |
290
|
|
|
|
291
|
1 |
View Code Duplication |
def test_path_constrained_bandwidth_detailed(self): |
|
|
|
|
292
|
|
|
"""Tests if the edges used in the paths |
293
|
|
|
from User 1 to User 2 have at least 20 bandwidth. |
294
|
|
|
""" |
295
|
1 |
|
requirements = {"bandwidth": 20} |
296
|
|
|
|
297
|
1 |
|
self.initializer() |
298
|
|
|
|
299
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
300
|
|
|
"User1", "User2", mandatory_metrics=requirements |
301
|
|
|
) |
302
|
1 |
|
assert paths |
303
|
|
|
|
304
|
1 |
|
for path in paths: |
305
|
1 |
|
bandwidths = self.links_metadata_values(path, "bandwidth") |
306
|
1 |
|
assert bandwidths |
307
|
|
|
|
308
|
1 |
|
for bandwidth in bandwidths: |
309
|
1 |
|
assert bandwidth >= requirements["bandwidth"] |
310
|
|
|
|
311
|
1 |
View Code Duplication |
def test_path_constrained_bandwidth_detailed_t2(self): |
|
|
|
|
312
|
|
|
"""Tests if the edges used in the paths |
313
|
|
|
from User 1 to User 2 have at least 20 bandwidth. |
314
|
|
|
""" |
315
|
1 |
|
requirements = {"bandwidth": 20} |
316
|
|
|
|
317
|
1 |
|
self.initializer(val=2) |
318
|
|
|
|
319
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
320
|
|
|
"User1", "User2", mandatory_metrics=requirements |
321
|
|
|
) |
322
|
1 |
|
assert paths |
323
|
|
|
|
324
|
1 |
|
for path in paths: |
325
|
1 |
|
bandwidths = self.links_metadata_values(path, "bandwidth") |
326
|
1 |
|
assert bandwidths |
327
|
1 |
|
for bandwidth in bandwidths: |
328
|
1 |
|
assert bandwidth >= requirements["bandwidth"] |
329
|
|
|
|
330
|
1 |
|
def test_path_constrained_bandwidth_delay(self): |
331
|
|
|
"""Tests if the edges used in the paths from User 1 |
332
|
|
|
to User 2 have at least 20 bandwidth and under 30 delay. |
333
|
|
|
""" |
334
|
1 |
|
requirements = {"bandwidth": 20, "delay": 29} |
335
|
|
|
|
336
|
1 |
|
self.initializer() |
337
|
|
|
|
338
|
1 |
|
paths = self.graph.constrained_k_shortest_paths( |
339
|
|
|
"User1", "User2", mandatory_metrics=requirements |
340
|
|
|
) |
341
|
1 |
|
assert paths |
342
|
|
|
|
343
|
1 |
|
for path in paths: |
344
|
|
|
|
345
|
1 |
|
bandwidths = self.links_metadata_values(path, "bandwidth") |
346
|
1 |
|
assert bandwidths |
347
|
1 |
|
for bandwidth in bandwidths: |
348
|
1 |
|
assert bandwidth >= requirements["bandwidth"] |
349
|
|
|
|
350
|
1 |
|
delays = self.links_metadata_values(path, "delay") |
351
|
1 |
|
assert delays |
352
|
1 |
|
for delay in delays: |
353
|
1 |
|
assert delay <= requirements["delay"] |
354
|
|
|
|
355
|
|
|
assert len(bandwidths) == len(delays) |
356
|
|
|
|