Total Complexity | 42 |
Total Lines | 710 |
Duplicated Lines | 4.23 % |
Coverage | 100% |
Changes | 0 |
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like build.tests.unit.models.test_evc_base often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
1 | """Module to test the EVCBase class.""" |
||
2 | 1 | import sys |
|
3 | 1 | from unittest.mock import MagicMock, call, patch |
|
4 | |||
5 | 1 | import pytest |
|
6 | |||
7 | 1 | from kytos.core.exceptions import KytosTagError |
|
8 | 1 | from kytos.core.interface import TAGRange |
|
9 | 1 | from napps.kytos.mef_eline.models import Path |
|
10 | |||
11 | # pylint: disable=wrong-import-position |
||
12 | 1 | sys.path.insert(0, "/var/lib/kytos/napps/..") |
|
13 | # pylint: enable=wrong-import-position, disable=ungrouped-imports |
||
14 | 1 | from napps.kytos.mef_eline.exceptions import DuplicatedNoTagUNI # NOQA pycodestyle |
|
15 | 1 | from napps.kytos.mef_eline.models import EVC # NOQA pycodestyle |
|
16 | 1 | from napps.kytos.mef_eline.scheduler import \ |
|
17 | CircuitSchedule # NOQA pycodestyle |
||
18 | 1 | from napps.kytos.mef_eline.tests.helpers import ( # NOQA pycodestyle |
|
19 | get_controller_mock, get_uni_mocked) |
||
20 | |||
21 | |||
22 | 1 | class TestEVC(): # pylint: disable=too-many-public-methods, no-member |
|
23 | """Tests to verify EVC class.""" |
||
24 | |||
25 | 1 | def test_attributes_empty(self): |
|
26 | """Test if the EVC raises an error with name is required.""" |
||
27 | 1 | attributes = {"controller": get_controller_mock()} |
|
28 | 1 | error_message = "name is required." |
|
29 | 1 | with pytest.raises(ValueError) as handle_error: |
|
30 | 1 | EVC(**attributes) |
|
31 | 1 | assert error_message in str(handle_error) |
|
32 | |||
33 | 1 | def test_expected_requiring_redeploy_attributes(self) -> None: |
|
34 | """Test expected attributes_requiring_redeploy.""" |
||
35 | 1 | expected = [ |
|
36 | "primary_path", |
||
37 | "backup_path", |
||
38 | "dynamic_backup_path", |
||
39 | "queue_id", |
||
40 | "sb_priority", |
||
41 | "primary_constraints", |
||
42 | "secondary_constraints", |
||
43 | "uni_a", |
||
44 | "uni_z", |
||
45 | ] |
||
46 | 1 | assert EVC.attributes_requiring_redeploy == expected |
|
47 | |||
48 | 1 | def test_expected_updatable_attributes(self) -> None: |
|
49 | """Test expected updatable_attributes.""" |
||
50 | 1 | expected = [ |
|
51 | "uni_a", |
||
52 | "uni_z", |
||
53 | "name", |
||
54 | "start_date", |
||
55 | "end_date", |
||
56 | "queue_id", |
||
57 | "bandwidth", |
||
58 | "primary_path", |
||
59 | "backup_path", |
||
60 | "dynamic_backup_path", |
||
61 | "primary_constraints", |
||
62 | "secondary_constraints", |
||
63 | "owner", |
||
64 | "sb_priority", |
||
65 | "service_level", |
||
66 | "circuit_scheduler", |
||
67 | "metadata", |
||
68 | "enabled", |
||
69 | "max_paths", |
||
70 | ] |
||
71 | 1 | assert EVC.updatable_attributes == set(expected) |
|
72 | |||
73 | 1 | def test_without_uni_a(self): |
|
74 | """Test if the EVC raises and error with UNI A is required.""" |
||
75 | 1 | attributes = { |
|
76 | "controller": get_controller_mock(), |
||
77 | "name": "circuit_name", |
||
78 | } |
||
79 | 1 | error_message = "uni_a is required." |
|
80 | 1 | with pytest.raises(ValueError) as handle_error: |
|
81 | 1 | EVC(**attributes) |
|
82 | 1 | assert error_message in str(handle_error) |
|
83 | |||
84 | 1 | def test_without_uni_z(self): |
|
85 | """Test if the EVC raises and error with UNI Z is required.""" |
||
86 | 1 | attributes = { |
|
87 | "controller": get_controller_mock(), |
||
88 | "name": "circuit_name", |
||
89 | "uni_a": get_uni_mocked(is_valid=True), |
||
90 | } |
||
91 | 1 | error_message = "uni_z is required." |
|
92 | 1 | with pytest.raises(ValueError) as handle_error: |
|
93 | 1 | EVC(**attributes) |
|
94 | 1 | assert error_message in str(handle_error) |
|
95 | |||
96 | 1 | @pytest.mark.parametrize( |
|
97 | "name,value", |
||
98 | [ |
||
99 | ("archived", True), |
||
100 | ("_id", True), |
||
101 | ("active", True), |
||
102 | ("current_path", []), |
||
103 | ("creation_time", "date"), |
||
104 | ] |
||
105 | ) |
||
106 | 1 | def test_update_read_only(self, name, value): |
|
107 | """Test if raises an error when trying to update read only attr.""" |
||
108 | 1 | attributes = { |
|
109 | "controller": get_controller_mock(), |
||
110 | "name": "circuit_name", |
||
111 | "dynamic_backup_path": True, |
||
112 | "uni_a": get_uni_mocked(is_valid=True), |
||
113 | "uni_z": get_uni_mocked(is_valid=True), |
||
114 | } |
||
115 | |||
116 | 1 | update_dict = {name: value} |
|
117 | 1 | error_message = f"{name} can't be updated." |
|
118 | 1 | with pytest.raises(ValueError) as handle_error: |
|
119 | 1 | evc = EVC(**attributes) |
|
120 | 1 | evc.update(**update_dict) |
|
121 | 1 | assert error_message in str(handle_error) |
|
122 | |||
123 | 1 | def test_update_invalid(self): |
|
124 | """Test updating with an invalid attr""" |
||
125 | 1 | attributes = { |
|
126 | "controller": get_controller_mock(), |
||
127 | "name": "circuit_name", |
||
128 | "dynamic_backup_path": True, |
||
129 | "uni_a": get_uni_mocked(is_valid=True), |
||
130 | "uni_z": get_uni_mocked(is_valid=True), |
||
131 | } |
||
132 | 1 | evc = EVC(**attributes) |
|
133 | 1 | with pytest.raises(ValueError) as handle_error: |
|
134 | 1 | evc.update(xyz="abc") |
|
135 | 1 | assert ( |
|
136 | "xyz can't be updated." |
||
137 | in str(handle_error) |
||
138 | ) |
||
139 | |||
140 | 1 | @patch("napps.kytos.mef_eline.models.EVC.sync") |
|
141 | 1 | def test_update_disable(self, _sync_mock): |
|
142 | """Test if evc is disabled.""" |
||
143 | 1 | attributes = { |
|
144 | "controller": get_controller_mock(), |
||
145 | "name": "circuit_name", |
||
146 | "dynamic_backup_path": True, |
||
147 | "enable": True, |
||
148 | "uni_a": get_uni_mocked(is_valid=True), |
||
149 | "uni_z": get_uni_mocked(is_valid=True), |
||
150 | } |
||
151 | 1 | update_dict = {"enabled": False} |
|
152 | 1 | evc = EVC(**attributes) |
|
153 | 1 | evc.update(**update_dict) |
|
154 | 1 | assert evc.is_enabled() is False |
|
155 | |||
156 | 1 | @patch("napps.kytos.mef_eline.models.EVC.sync") |
|
157 | 1 | def test_update_empty_primary_path(self, _sync_mock): |
|
158 | """Test if an empty primary path can be set.""" |
||
159 | 1 | initial_primary_path = Path([MagicMock(id=1), MagicMock(id=2)]) |
|
160 | 1 | attributes = { |
|
161 | "controller": get_controller_mock(), |
||
162 | "name": "circuit_name", |
||
163 | "dynamic_backup_path": True, |
||
164 | "primary_path": initial_primary_path, |
||
165 | "enable": True, |
||
166 | "uni_a": get_uni_mocked(is_valid=True), |
||
167 | "uni_z": get_uni_mocked(is_valid=True), |
||
168 | } |
||
169 | 1 | update_dict = {"primary_path": Path([])} |
|
170 | 1 | evc = EVC(**attributes) |
|
171 | 1 | assert evc.primary_path == initial_primary_path |
|
172 | 1 | evc.update(**update_dict) |
|
173 | 1 | assert len(evc.primary_path) == 0 |
|
174 | |||
175 | 1 | @patch("napps.kytos.mef_eline.models.EVC.sync") |
|
176 | 1 | def test_update_empty_path_non_dynamic_backup(self, _sync_mock): |
|
177 | """Test if an empty primary path can't be set if dynamic.""" |
||
178 | 1 | initial_primary_path = Path([MagicMock(id=1), MagicMock(id=2)]) |
|
179 | 1 | attributes = { |
|
180 | "controller": get_controller_mock(), |
||
181 | "name": "circuit_name", |
||
182 | "dynamic_backup_path": False, |
||
183 | "primary_path": initial_primary_path, |
||
184 | "enable": True, |
||
185 | "uni_a": get_uni_mocked(is_valid=True), |
||
186 | "uni_z": get_uni_mocked(is_valid=True), |
||
187 | } |
||
188 | 1 | update_dict = {"primary_path": Path([])} |
|
189 | 1 | evc = EVC(**attributes) |
|
190 | 1 | assert evc.primary_path == initial_primary_path |
|
191 | 1 | with pytest.raises(ValueError) as handle_error: |
|
192 | 1 | evc.update(**update_dict) |
|
193 | 1 | assert ( |
|
194 | 'The EVC must have a primary path or allow dynamic paths.' |
||
195 | in str(handle_error) |
||
196 | ) |
||
197 | |||
198 | 1 | @patch("napps.kytos.mef_eline.models.EVC.sync") |
|
199 | 1 | def test_update_empty_backup_path(self, _sync_mock): |
|
200 | """Test if an empty backup path can be set.""" |
||
201 | 1 | initial_backup_path = Path([MagicMock(id=1), MagicMock(id=2)]) |
|
202 | 1 | attributes = { |
|
203 | "controller": get_controller_mock(), |
||
204 | "name": "circuit_name", |
||
205 | "dynamic_backup_path": True, |
||
206 | "backup_path": initial_backup_path, |
||
207 | "enable": True, |
||
208 | "uni_a": get_uni_mocked(is_valid=True), |
||
209 | "uni_z": get_uni_mocked(is_valid=True), |
||
210 | } |
||
211 | 1 | update_dict = {"backup_path": Path([])} |
|
212 | 1 | evc = EVC(**attributes) |
|
213 | 1 | assert evc.backup_path == initial_backup_path |
|
214 | 1 | evc.update(**update_dict) |
|
215 | 1 | assert len(evc.backup_path) == 0 |
|
216 | |||
217 | 1 | @patch("napps.kytos.mef_eline.models.EVC.sync") |
|
218 | 1 | def test_update_empty_backup_path_non_dynamic(self, _sync_mock): |
|
219 | """Test if an empty backup path can be set even if it's non dynamic.""" |
||
220 | 1 | initial_backup_path = Path([MagicMock(id=1), MagicMock(id=2)]) |
|
221 | 1 | primary_path = Path([MagicMock(id=3), MagicMock(id=4)]) |
|
222 | 1 | attributes = { |
|
223 | "controller": get_controller_mock(), |
||
224 | "name": "circuit_name", |
||
225 | "dynamic_backup_path": False, |
||
226 | "primary_path": primary_path, |
||
227 | "backup_path": initial_backup_path, |
||
228 | "enable": True, |
||
229 | "uni_a": get_uni_mocked(is_valid=True), |
||
230 | "uni_z": get_uni_mocked(is_valid=True), |
||
231 | } |
||
232 | 1 | update_dict = {"backup_path": Path([])} |
|
233 | 1 | evc = EVC(**attributes) |
|
234 | 1 | assert evc.primary_path == primary_path |
|
235 | 1 | assert evc.backup_path == initial_backup_path |
|
236 | 1 | evc.update(**update_dict) |
|
237 | 1 | assert evc.primary_path == primary_path |
|
238 | 1 | assert len(evc.backup_path) == 0 |
|
239 | |||
240 | 1 | View Code Duplication | @patch("napps.kytos.mef_eline.models.EVC.sync") |
|
|||
241 | 1 | def test_update_queue(self, _sync_mock): |
|
242 | """Test if evc is set to redeploy.""" |
||
243 | 1 | attributes = { |
|
244 | "controller": get_controller_mock(), |
||
245 | "name": "circuit_name", |
||
246 | "enable": True, |
||
247 | "dynamic_backup_path": True, |
||
248 | "uni_a": get_uni_mocked(is_valid=True), |
||
249 | "uni_z": get_uni_mocked(is_valid=True), |
||
250 | } |
||
251 | 1 | update_dict = {"queue_id": 3} |
|
252 | 1 | evc = EVC(**attributes) |
|
253 | 1 | _, redeploy = evc.update(**update_dict) |
|
254 | 1 | assert redeploy |
|
255 | |||
256 | 1 | View Code Duplication | @patch("napps.kytos.mef_eline.models.EVC.sync") |
257 | 1 | def test_update_queue_null(self, _sync_mock): |
|
258 | """Test if evc is set to redeploy.""" |
||
259 | 1 | attributes = { |
|
260 | "controller": get_controller_mock(), |
||
261 | "name": "circuit_name", |
||
262 | "enable": True, |
||
263 | "dynamic_backup_path": True, |
||
264 | "uni_a": get_uni_mocked(is_valid=True), |
||
265 | "uni_z": get_uni_mocked(is_valid=True), |
||
266 | } |
||
267 | 1 | update_dict = {"queue_id": None} |
|
268 | 1 | evc = EVC(**attributes) |
|
269 | 1 | _, redeploy = evc.update(**update_dict) |
|
270 | 1 | assert redeploy |
|
271 | |||
272 | 1 | def test_update_different_tag_lists(self): |
|
273 | """Test update when tag lists are different.""" |
||
274 | 1 | attributes = { |
|
275 | "controller": get_controller_mock(), |
||
276 | "name": "circuit_name", |
||
277 | "enable": True, |
||
278 | "dynamic_backup_path": True, |
||
279 | "uni_a": get_uni_mocked(is_valid=True), |
||
280 | "uni_z": get_uni_mocked(is_valid=True), |
||
281 | } |
||
282 | 1 | uni = MagicMock(user_tag=TAGRange("vlan", [[1, 10]])) |
|
283 | 1 | update_dict = {"uni_a": uni} |
|
284 | 1 | evc = EVC(**attributes) |
|
285 | 1 | with pytest.raises(ValueError): |
|
286 | 1 | evc.update(**update_dict) |
|
287 | |||
288 | 1 | def test_circuit_representation(self): |
|
289 | """Test the method __repr__.""" |
||
290 | 1 | attributes = { |
|
291 | "controller": get_controller_mock(), |
||
292 | "name": "circuit_name", |
||
293 | "uni_a": get_uni_mocked(is_valid=True), |
||
294 | "uni_z": get_uni_mocked(is_valid=True), |
||
295 | } |
||
296 | 1 | evc = EVC(**attributes) |
|
297 | 1 | expected_value = f"EVC({evc.id}, {evc.name})" |
|
298 | 1 | assert str(evc) == expected_value |
|
299 | |||
300 | 1 | def test_comparison_method(self): |
|
301 | """Test the method __eq__.""" |
||
302 | 1 | attributes = { |
|
303 | "controller": get_controller_mock(), |
||
304 | "name": "circuit_name", |
||
305 | "uni_a": get_uni_mocked(is_valid=True), |
||
306 | "uni_z": get_uni_mocked(is_valid=True), |
||
307 | } |
||
308 | 1 | evc1 = EVC(**attributes) |
|
309 | 1 | evc2 = EVC(**attributes) |
|
310 | |||
311 | 1 | attributes = { |
|
312 | "controller": get_controller_mock(), |
||
313 | "name": "circuit_name_2", |
||
314 | "uni_a": get_uni_mocked(is_valid=True), |
||
315 | "uni_z": get_uni_mocked(is_valid=True), |
||
316 | } |
||
317 | 1 | evc3 = EVC(**attributes) |
|
318 | 1 | evc4 = EVC(**attributes) |
|
319 | |||
320 | 1 | assert evc1 == evc2 |
|
321 | 1 | assert evc1 != evc3 |
|
322 | 1 | assert evc2 != evc3 |
|
323 | 1 | assert evc3 == evc4 |
|
324 | |||
325 | 1 | def test_as_dict(self): |
|
326 | """Test the method as_dict.""" |
||
327 | 1 | attributes = { |
|
328 | "controller": get_controller_mock(), |
||
329 | "id": "custom_id", |
||
330 | "name": "custom_name", |
||
331 | "uni_a": get_uni_mocked(is_valid=True), |
||
332 | "uni_z": get_uni_mocked(is_valid=True), |
||
333 | "start_date": "2018-08-21T18:44:54", |
||
334 | "end_date": "2018-08-21T18:44:55", |
||
335 | "primary_links": [], |
||
336 | "request_time": "2018-08-21T19:10:41", |
||
337 | "creation_time": "2018-08-21T18:44:54", |
||
338 | "owner": "my_name", |
||
339 | "circuit_scheduler": [ |
||
340 | CircuitSchedule.from_dict( |
||
341 | { |
||
342 | "id": 234243247, |
||
343 | "action": "create", |
||
344 | "frequency": "1 * * * *", |
||
345 | } |
||
346 | ), |
||
347 | CircuitSchedule.from_dict( |
||
348 | { |
||
349 | "id": 234243239, |
||
350 | "action": "create", |
||
351 | "interval": {"hours": 2}, |
||
352 | } |
||
353 | ), |
||
354 | ], |
||
355 | "enabled": True, |
||
356 | "sb_priority": 2, |
||
357 | "service_level": 7, |
||
358 | } |
||
359 | 1 | evc = EVC(**attributes) |
|
360 | |||
361 | 1 | expected_dict = { |
|
362 | "id": "custom_id", |
||
363 | "name": "custom_name", |
||
364 | "uni_a": attributes["uni_a"].as_dict(), |
||
365 | "uni_z": attributes["uni_z"].as_dict(), |
||
366 | "start_date": "2018-08-21T18:44:54", |
||
367 | "end_date": "2018-08-21T18:44:55", |
||
368 | "bandwidth": 0, |
||
369 | "primary_links": [], |
||
370 | "backup_links": [], |
||
371 | "current_path": [], |
||
372 | "primary_path": [], |
||
373 | "backup_path": [], |
||
374 | "dynamic_backup_path": False, |
||
375 | "request_time": "2018-08-21T19:10:41", |
||
376 | "creation_time": "2018-08-21T18:44:54", |
||
377 | "circuit_scheduler": [ |
||
378 | { |
||
379 | "id": 234243247, |
||
380 | "action": "create", |
||
381 | "frequency": "1 * * * *", |
||
382 | }, |
||
383 | { |
||
384 | "id": 234243239, |
||
385 | "action": "create", |
||
386 | "interval": {"hours": 2}, |
||
387 | }, |
||
388 | ], |
||
389 | "active": False, |
||
390 | "enabled": True, |
||
391 | "sb_priority": 2, |
||
392 | "service_level": 7, |
||
393 | } |
||
394 | 1 | actual_dict = evc.as_dict() |
|
395 | 1 | for name, value in expected_dict.items(): |
|
396 | 1 | actual = actual_dict.get(name) |
|
397 | 1 | assert value == actual |
|
398 | |||
399 | # Selected fields |
||
400 | 1 | expected_dict = { |
|
401 | "enabled": True, |
||
402 | "uni_z": attributes["uni_z"].as_dict(), |
||
403 | "circuit_scheduler": [ |
||
404 | { |
||
405 | "id": 234243247, |
||
406 | "action": "create", |
||
407 | "frequency": "1 * * * *", |
||
408 | }, |
||
409 | { |
||
410 | "id": 234243239, |
||
411 | "action": "create", |
||
412 | "interval": {"hours": 2}, |
||
413 | }, |
||
414 | ], |
||
415 | "sb_priority": 2, |
||
416 | } |
||
417 | 1 | selected_fields = { |
|
418 | "enabled", "uni_z", "circuit_scheduler", "sb_priority" |
||
419 | } |
||
420 | 1 | actual_dict = evc.as_dict(selected_fields) |
|
421 | 1 | for name, value in expected_dict.items(): |
|
422 | 1 | actual = actual_dict.get(name) |
|
423 | 1 | assert value == actual |
|
424 | |||
425 | 1 | @staticmethod |
|
426 | 1 | def test_get_id_from_cookie(): |
|
427 | """Test get_id_from_cookie.""" |
||
428 | 1 | attributes = { |
|
429 | "controller": get_controller_mock(), |
||
430 | "name": "circuit_name", |
||
431 | "enable": True, |
||
432 | "uni_a": get_uni_mocked(is_valid=True), |
||
433 | "uni_z": get_uni_mocked(is_valid=True) |
||
434 | } |
||
435 | 1 | evc = EVC(**attributes) |
|
436 | 1 | evc_id = evc.id |
|
437 | 1 | assert evc_id |
|
438 | 1 | assert evc.get_id_from_cookie(evc.get_cookie()) == evc_id |
|
439 | |||
440 | 1 | @staticmethod |
|
441 | 1 | def test_get_id_from_cookie_with_leading_zeros(): |
|
442 | """Test get_id_from_cookie with leading zeros.""" |
||
443 | |||
444 | 1 | attributes = { |
|
445 | "controller": get_controller_mock(), |
||
446 | "name": "circuit_name", |
||
447 | "enable": True, |
||
448 | "uni_a": get_uni_mocked(is_valid=True), |
||
449 | "uni_z": get_uni_mocked(is_valid=True) |
||
450 | } |
||
451 | 1 | evc = EVC(**attributes) |
|
452 | 1 | evc_id = "0a2d672d99ff41" |
|
453 | # pylint: disable=protected-access |
||
454 | 1 | evc._id = evc_id |
|
455 | # pylint: enable=protected-access |
||
456 | 1 | assert EVC.get_id_from_cookie(evc.get_cookie()) == evc_id |
|
457 | |||
458 | 1 | def test_is_intra_switch(self): |
|
459 | """Test is_intra_switch method.""" |
||
460 | 1 | attributes = { |
|
461 | "controller": get_controller_mock(), |
||
462 | "name": "circuit_name", |
||
463 | "enable": True, |
||
464 | "uni_a": get_uni_mocked(is_valid=True), |
||
465 | "uni_z": get_uni_mocked(is_valid=True) |
||
466 | } |
||
467 | 1 | evc = EVC(**attributes) |
|
468 | 1 | assert not evc.is_intra_switch() |
|
469 | |||
470 | 1 | evc.uni_a.interface.switch = evc.uni_z.interface.switch |
|
471 | 1 | assert evc.is_intra_switch() |
|
472 | |||
473 | 1 | def test_default_queue_id(self): |
|
474 | """Test default queue_id""" |
||
475 | |||
476 | 1 | attributes = { |
|
477 | "controller": get_controller_mock(), |
||
478 | "name": "circuit_1", |
||
479 | "uni_a": get_uni_mocked(is_valid=True), |
||
480 | "uni_z": get_uni_mocked(is_valid=True), |
||
481 | "dynamic_backup_path": True, |
||
482 | } |
||
483 | |||
484 | 1 | evc = EVC(**attributes) |
|
485 | 1 | assert evc.queue_id == -1 |
|
486 | |||
487 | 1 | def test_get_unis_use_tags(self): |
|
488 | """Test _get_unis_use_tags""" |
||
489 | 1 | old_uni_a = get_uni_mocked( |
|
490 | interface_port=2, |
||
491 | is_valid=True |
||
492 | ) |
||
493 | 1 | old_uni_z = get_uni_mocked( |
|
494 | interface_port=3, |
||
495 | is_valid=True |
||
496 | ) |
||
497 | 1 | attributes = { |
|
498 | "controller": get_controller_mock(), |
||
499 | "name": "circuit_name", |
||
500 | "enable": True, |
||
501 | "uni_a": old_uni_a, |
||
502 | "uni_z": old_uni_z |
||
503 | } |
||
504 | 1 | evc = EVC(**attributes) |
|
505 | 1 | evc._use_uni_vlan = MagicMock() |
|
506 | 1 | evc.make_uni_vlan_available = MagicMock() |
|
507 | 1 | new_uni_a = get_uni_mocked(tag_value=200, is_valid=True) |
|
508 | 1 | new_uni_z = get_uni_mocked(tag_value=200, is_valid=True) |
|
509 | 1 | unis = {"uni_a": new_uni_a} |
|
510 | 1 | evc._get_unis_use_tags(**unis) |
|
511 | 1 | assert evc._use_uni_vlan.call_count == 1 |
|
512 | 1 | assert evc._use_uni_vlan.call_args[0][0] == new_uni_a |
|
513 | 1 | assert evc.make_uni_vlan_available.call_count == 1 |
|
514 | 1 | assert evc.make_uni_vlan_available.call_args[0][0] == old_uni_a |
|
515 | |||
516 | # Two UNIs |
||
517 | 1 | evc = EVC(**attributes) |
|
518 | 1 | evc._use_uni_vlan = MagicMock() |
|
519 | 1 | evc.make_uni_vlan_available = MagicMock() |
|
520 | 1 | unis = {"uni_a": new_uni_a, "uni_z": new_uni_z} |
|
521 | 1 | evc._get_unis_use_tags(**unis) |
|
522 | |||
523 | 1 | expected = [ |
|
524 | call(new_uni_a, uni_dif=old_uni_a), |
||
525 | call(new_uni_z, uni_dif=old_uni_z) |
||
526 | ] |
||
527 | 1 | evc._use_uni_vlan.assert_has_calls(expected) |
|
528 | 1 | expected = [ |
|
529 | call(old_uni_z, uni_dif=new_uni_z), |
||
530 | call(old_uni_a, uni_dif=new_uni_a) |
||
531 | ] |
||
532 | 1 | evc.make_uni_vlan_available.assert_has_calls(expected) |
|
533 | |||
534 | 1 | def test_get_unis_use_tags_error(self): |
|
535 | """Test _get_unis_use_tags with KytosTagError""" |
||
536 | 1 | old_uni_a = get_uni_mocked( |
|
537 | interface_port=2, |
||
538 | is_valid=True |
||
539 | ) |
||
540 | 1 | old_uni_z = get_uni_mocked( |
|
541 | interface_port=3, |
||
542 | is_valid=True |
||
543 | ) |
||
544 | 1 | attributes = { |
|
545 | "controller": get_controller_mock(), |
||
546 | "name": "circuit_name", |
||
547 | "enable": True, |
||
548 | "uni_a": old_uni_a, |
||
549 | "uni_z": old_uni_z |
||
550 | } |
||
551 | 1 | evc = EVC(**attributes) |
|
552 | 1 | evc._use_uni_vlan = MagicMock() |
|
553 | |||
554 | # UNI Z KytosTagError |
||
555 | 1 | evc._use_uni_vlan.side_effect = [None, KytosTagError("")] |
|
556 | 1 | evc.make_uni_vlan_available = MagicMock() |
|
557 | 1 | new_uni_a = get_uni_mocked(tag_value=200, is_valid=True) |
|
558 | 1 | new_uni_z = get_uni_mocked(tag_value=200, is_valid=True) |
|
559 | 1 | unis = {"uni_a": new_uni_a, "uni_z": new_uni_z} |
|
560 | 1 | with pytest.raises(KytosTagError): |
|
561 | 1 | evc._get_unis_use_tags(**unis) |
|
562 | 1 | expected = [ |
|
563 | call(new_uni_a, uni_dif=old_uni_a), |
||
564 | call(new_uni_z, uni_dif=old_uni_z) |
||
565 | ] |
||
566 | 1 | evc._use_uni_vlan.assert_has_calls(expected) |
|
567 | 1 | assert evc.make_uni_vlan_available.call_count == 1 |
|
568 | 1 | assert evc.make_uni_vlan_available.call_args[0][0] == new_uni_a |
|
569 | |||
570 | # UNI A KytosTagError |
||
571 | 1 | evc = EVC(**attributes) |
|
572 | 1 | evc._use_uni_vlan = MagicMock() |
|
573 | 1 | evc._use_uni_vlan.side_effect = [KytosTagError(""), None] |
|
574 | 1 | evc.make_uni_vlan_available = MagicMock() |
|
575 | 1 | new_uni_a = get_uni_mocked(tag_value=200, is_valid=True) |
|
576 | 1 | new_uni_z = get_uni_mocked(tag_value=200, is_valid=True) |
|
577 | 1 | unis = {"uni_a": new_uni_a, "uni_z": new_uni_z} |
|
578 | 1 | with pytest.raises(KytosTagError): |
|
579 | 1 | evc._get_unis_use_tags(**unis) |
|
580 | 1 | assert evc._use_uni_vlan.call_count == 1 |
|
581 | 1 | assert evc._use_uni_vlan.call_args[0][0] == new_uni_a |
|
582 | 1 | assert evc.make_uni_vlan_available.call_count == 0 |
|
583 | |||
584 | 1 | @patch("napps.kytos.mef_eline.models.evc.range_difference") |
|
585 | 1 | def test_use_uni_vlan(self, mock_difference): |
|
586 | """Test _use_uni_vlan""" |
||
587 | 1 | attributes = { |
|
588 | "controller": get_controller_mock(), |
||
589 | "name": "circuit_name", |
||
590 | "enable": True, |
||
591 | "uni_a": get_uni_mocked(is_valid=True), |
||
592 | "uni_z": get_uni_mocked(is_valid=True) |
||
593 | } |
||
594 | 1 | evc = EVC(**attributes) |
|
595 | 1 | uni = get_uni_mocked(is_valid=True) |
|
596 | 1 | uni.interface.use_tags = MagicMock() |
|
597 | 1 | evc._use_uni_vlan(uni) |
|
598 | 1 | args = uni.interface.use_tags.call_args[0] |
|
599 | 1 | assert args[1] == uni.user_tag.value |
|
600 | 1 | assert args[2] == uni.user_tag.tag_type |
|
601 | 1 | assert uni.interface.use_tags.call_count == 1 |
|
602 | |||
603 | 1 | uni.user_tag.value = "any" |
|
604 | 1 | evc._use_uni_vlan(uni) |
|
605 | 1 | assert uni.interface.use_tags.call_count == 2 |
|
606 | |||
607 | 1 | uni.user_tag.value = [[1, 10]] |
|
608 | 1 | uni_dif = get_uni_mocked(tag_value=[[1, 2]]) |
|
609 | 1 | mock_difference.return_value = [[3, 10]] |
|
610 | 1 | evc._use_uni_vlan(uni, uni_dif) |
|
611 | 1 | assert uni.interface.use_tags.call_count == 3 |
|
612 | |||
613 | 1 | mock_difference.return_value = [] |
|
614 | 1 | evc._use_uni_vlan(uni, uni_dif) |
|
615 | 1 | assert uni.interface.use_tags.call_count == 3 |
|
616 | |||
617 | 1 | uni.interface.use_tags.side_effect = KytosTagError("") |
|
618 | 1 | with pytest.raises(KytosTagError): |
|
619 | 1 | evc._use_uni_vlan(uni) |
|
620 | 1 | assert uni.interface.use_tags.call_count == 4 |
|
621 | |||
622 | 1 | uni.user_tag = None |
|
623 | 1 | evc._use_uni_vlan(uni) |
|
624 | 1 | assert uni.interface.use_tags.call_count == 4 |
|
625 | |||
626 | 1 | @patch("napps.kytos.mef_eline.models.evc.log") |
|
627 | 1 | def test_make_uni_vlan_available(self, mock_log): |
|
628 | """Test make_uni_vlan_available""" |
||
629 | 1 | attributes = { |
|
630 | "controller": get_controller_mock(), |
||
631 | "name": "circuit_name", |
||
632 | "enable": True, |
||
633 | "uni_a": get_uni_mocked(is_valid=True), |
||
634 | "uni_z": get_uni_mocked(is_valid=True) |
||
635 | } |
||
636 | 1 | evc = EVC(**attributes) |
|
637 | 1 | uni = get_uni_mocked(is_valid=True) |
|
638 | 1 | uni.interface.make_tags_available = MagicMock() |
|
639 | |||
640 | 1 | evc.make_uni_vlan_available(uni) |
|
641 | 1 | args = uni.interface.make_tags_available.call_args[0] |
|
642 | 1 | assert args[1] == uni.user_tag.value |
|
643 | 1 | assert args[2] == uni.user_tag.tag_type |
|
644 | 1 | assert uni.interface.make_tags_available.call_count == 1 |
|
645 | |||
646 | 1 | uni.user_tag.value = [[1, 10]] |
|
647 | 1 | uni_dif = get_uni_mocked(tag_value=[[1, 2]]) |
|
648 | 1 | evc.make_uni_vlan_available(uni, uni_dif) |
|
649 | 1 | assert uni.interface.make_tags_available.call_count == 2 |
|
650 | |||
651 | 1 | uni.interface.make_tags_available.side_effect = KytosTagError("") |
|
652 | 1 | evc.make_uni_vlan_available(uni) |
|
653 | 1 | assert mock_log.error.call_count == 1 |
|
654 | |||
655 | 1 | uni.user_tag = None |
|
656 | 1 | evc.make_uni_vlan_available(uni) |
|
657 | 1 | assert uni.interface.make_tags_available.call_count == 3 |
|
658 | |||
659 | 1 | def test_remove_uni_tags(self): |
|
660 | """Test remove_uni_tags""" |
||
661 | 1 | attributes = { |
|
662 | "controller": get_controller_mock(), |
||
663 | "name": "circuit_name", |
||
664 | "enable": True, |
||
665 | "uni_a": get_uni_mocked(is_valid=True), |
||
666 | "uni_z": get_uni_mocked(is_valid=True) |
||
667 | } |
||
668 | 1 | evc = EVC(**attributes) |
|
669 | 1 | evc.make_uni_vlan_available = MagicMock() |
|
670 | 1 | evc.remove_uni_tags() |
|
671 | 1 | assert evc.make_uni_vlan_available.call_count == 2 |
|
672 | |||
673 | 1 | def test_tag_lists_equal(self): |
|
674 | """Test _tag_lists_equal""" |
||
675 | 1 | attributes = { |
|
676 | "controller": get_controller_mock(), |
||
677 | "name": "circuit_name", |
||
678 | "enable": True, |
||
679 | "uni_a": get_uni_mocked(is_valid=True), |
||
680 | "uni_z": get_uni_mocked(is_valid=True) |
||
681 | } |
||
682 | 1 | evc = EVC(**attributes) |
|
683 | 1 | uni = MagicMock(user_tag=TAGRange("vlan", [[1, 10]])) |
|
684 | 1 | update_dict = {"uni_z": uni} |
|
685 | 1 | assert evc._tag_lists_equal(**update_dict) is False |
|
686 | |||
687 | 1 | update_dict = {"uni_a": uni, "uni_z": uni} |
|
688 | 1 | assert evc._tag_lists_equal(**update_dict) |
|
689 | |||
690 | 1 | def test_check_no_tag_duplicate(self): |
|
691 | """Test check_no_tag_duplicate""" |
||
692 | 1 | uni_01_1 = get_uni_mocked(interface_port=1, switch_dpid="01") |
|
693 | 1 | uni_01_1.user_tag = None |
|
694 | 1 | uni_a = uni_01_1 |
|
695 | 1 | uni_z = get_uni_mocked(is_valid=True) |
|
696 | 1 | attributes = { |
|
697 | "controller": get_controller_mock(), |
||
698 | "name": "circuit_name", |
||
699 | "enable": True, |
||
700 | "uni_a": uni_a, |
||
701 | "uni_z": uni_z |
||
702 | } |
||
703 | 1 | evc = EVC(**attributes) |
|
704 | 1 | other_uni = uni_01_1 |
|
705 | 1 | with pytest.raises(DuplicatedNoTagUNI): |
|
706 | 1 | evc.check_no_tag_duplicate(other_uni) |
|
707 | |||
708 | 1 | other_uni = get_uni_mocked(interface_port=1, switch_dpid="02") |
|
709 | assert evc.check_no_tag_duplicate(other_uni) is None |
||
710 |