Passed
Push — master ( 06a873...24a68d )
by Aldo
05:04
created

build.tests.unit.models.test_evc_base   B

Complexity

Total Complexity 46

Size/Duplication

Total Lines 772
Duplicated Lines 3.89 %

Test Coverage

Coverage 100%

Importance

Changes 0
Metric Value
eloc 606
dl 30
loc 772
ccs 338
cts 338
cp 1
rs 8.714
c 0
b 0
f 0
wmc 46

31 Methods

Rating   Name   Duplication   Size   Complexity  
A TestEVC.test_attributes_empty() 0 7 2
A TestEVC.test_expected_requiring_redeploy_attributes() 0 14 1
A TestEVC.test_update_read_only() 0 26 2
A TestEVC.test_expected_updatable_attributes() 0 24 1
A TestEVC.test_update_empty_path_non_dynamic_backup() 0 21 2
A TestEVC.test_without_uni_a() 0 10 2
A TestEVC.test_update_empty_backup_path() 0 18 1
A TestEVC.test_update_empty_primary_path() 0 18 1
A TestEVC.test_update_disable() 0 15 1
A TestEVC.test_without_uni_z() 0 11 2
A TestEVC.test_update_invalid() 0 15 2
A TestEVC.test_circuit_representation() 0 11 1
B TestEVC.test_as_dict() 0 99 3
A TestEVC.test_default_queue_id() 0 13 1
A TestEVC.test_use_uni_vlan() 0 41 2
A TestEVC.test_update_queue() 15 15 1
A TestEVC.test_update_different_tag_lists() 0 15 2
A TestEVC.test_update_empty_backup_path_non_dynamic() 0 29 1
B TestEVC.test_get_unis_use_tags_error() 0 49 3
A TestEVC.test_is_intra_switch() 0 14 1
A TestEVC.test_update_invalid_path_uni() 0 32 2
A TestEVC.test_get_id_from_cookie_with_leading_zeros() 0 17 1
A TestEVC.test_remove_uni_tags() 0 13 1
A TestEVC.test_comparison_method() 0 24 1
A TestEVC.test_check_no_tag_duplicate() 0 20 2
A TestEVC.test_update_invalid_path_vlan() 0 21 2
B TestEVC.test_get_unis_use_tags() 0 46 1
A TestEVC.test_update_queue_null() 15 15 1
A TestEVC.test_make_uni_vlan_available() 0 32 1
A TestEVC.test_get_id_from_cookie() 0 14 1
A TestEVC.test_tag_lists_equal() 0 16 1

How to fix   Duplicated Code    Complexity   

Duplicated Code

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:

Complexity

 Tip:   Before tackling complexity, make sure that you eliminate any duplication first. This often can reduce the size of classes significantly.

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, InvalidPath  # 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, switch_id="00:01"),
167
            "uni_z": get_uni_mocked(is_valid=True, switch_id="00:02"),
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
        uni_a = get_uni_mocked(is_valid=True, switch_id="00:01")
221 1
        uni_z = get_uni_mocked(is_valid=True, switch_id="00:02")
222 1
        link = MagicMock()
223 1
        link.endpoint_a.switch = uni_a.interface.switch
224 1
        link.endpoint_a.link = link
225 1
        link.endpoint_b.switch = uni_z.interface.switch
226 1
        link.endpoint_b.link = link
227 1
        primary_path = Path([link])
228 1
        initial_backup_path = Path([link])
229 1
        attributes = {
230
            "controller": get_controller_mock(),
231
            "name": "circuit_name",
232
            "dynamic_backup_path": False,
233
            "primary_path": primary_path,
234
            "backup_path": initial_backup_path,
235
            "enable": True,
236
            "uni_a": uni_a,
237
            "uni_z": uni_z,
238
        }
239 1
        evc = EVC(**attributes)
240 1
        assert evc.primary_path == primary_path
241 1
        assert evc.backup_path == initial_backup_path
242 1
        update_dict = {"backup_path": Path([])}
243 1
        evc.update(**update_dict)
244 1
        assert evc.primary_path == primary_path
245 1
        assert len(evc.backup_path) == 0
246
247 1 View Code Duplication
    @patch("napps.kytos.mef_eline.models.EVC.sync")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
248 1
    def test_update_queue(self, _sync_mock):
249
        """Test if evc is set to redeploy."""
250 1
        attributes = {
251
            "controller": get_controller_mock(),
252
            "name": "circuit_name",
253
            "enable": True,
254
            "dynamic_backup_path": True,
255
            "uni_a": get_uni_mocked(is_valid=True),
256
            "uni_z": get_uni_mocked(is_valid=True),
257
        }
258 1
        update_dict = {"queue_id": 3}
259 1
        evc = EVC(**attributes)
260 1
        _, redeploy = evc.update(**update_dict)
261 1
        assert redeploy
262
263 1 View Code Duplication
    @patch("napps.kytos.mef_eline.models.EVC.sync")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
264 1
    def test_update_queue_null(self, _sync_mock):
265
        """Test if evc is set to redeploy."""
266 1
        attributes = {
267
            "controller": get_controller_mock(),
268
            "name": "circuit_name",
269
            "enable": True,
270
            "dynamic_backup_path": True,
271
            "uni_a": get_uni_mocked(is_valid=True),
272
            "uni_z": get_uni_mocked(is_valid=True),
273
        }
274 1
        update_dict = {"queue_id": None}
275 1
        evc = EVC(**attributes)
276 1
        _, redeploy = evc.update(**update_dict)
277 1
        assert redeploy
278
279 1
    def test_update_different_tag_lists(self):
280
        """Test update when tag lists are different."""
281 1
        attributes = {
282
            "controller": get_controller_mock(),
283
            "name": "circuit_name",
284
            "enable": True,
285
            "dynamic_backup_path": True,
286
            "uni_a": get_uni_mocked(is_valid=True),
287
            "uni_z": get_uni_mocked(is_valid=True),
288
        }
289 1
        uni = MagicMock(user_tag=TAGRange("vlan", [[1, 10]]))
290 1
        update_dict = {"uni_a": uni}
291 1
        evc = EVC(**attributes)
292 1
        with pytest.raises(ValueError):
293 1
            evc.update(**update_dict)
294
295 1
    def test_circuit_representation(self):
296
        """Test the method __repr__."""
297 1
        attributes = {
298
            "controller": get_controller_mock(),
299
            "name": "circuit_name",
300
            "uni_a": get_uni_mocked(is_valid=True),
301
            "uni_z": get_uni_mocked(is_valid=True),
302
        }
303 1
        evc = EVC(**attributes)
304 1
        expected_value = f"EVC({evc.id}, {evc.name})"
305 1
        assert str(evc) == expected_value
306
307 1
    def test_comparison_method(self):
308
        """Test the method __eq__."""
309 1
        attributes = {
310
            "controller": get_controller_mock(),
311
            "name": "circuit_name",
312
            "uni_a": get_uni_mocked(is_valid=True),
313
            "uni_z": get_uni_mocked(is_valid=True),
314
        }
315 1
        evc1 = EVC(**attributes)
316 1
        evc2 = EVC(**attributes)
317
318 1
        attributes = {
319
            "controller": get_controller_mock(),
320
            "name": "circuit_name_2",
321
            "uni_a": get_uni_mocked(is_valid=True),
322
            "uni_z": get_uni_mocked(is_valid=True),
323
        }
324 1
        evc3 = EVC(**attributes)
325 1
        evc4 = EVC(**attributes)
326
327 1
        assert evc1 == evc2
328 1
        assert evc1 != evc3
329 1
        assert evc2 != evc3
330 1
        assert evc3 == evc4
331
332 1
    def test_as_dict(self):
333
        """Test the method as_dict."""
334 1
        attributes = {
335
            "controller": get_controller_mock(),
336
            "id": "custom_id",
337
            "name": "custom_name",
338
            "uni_a": get_uni_mocked(is_valid=True),
339
            "uni_z": get_uni_mocked(is_valid=True),
340
            "start_date": "2018-08-21T18:44:54",
341
            "end_date": "2018-08-21T18:44:55",
342
            "primary_links": [],
343
            "request_time": "2018-08-21T19:10:41",
344
            "creation_time": "2018-08-21T18:44:54",
345
            "owner": "my_name",
346
            "circuit_scheduler": [
347
                CircuitSchedule.from_dict(
348
                    {
349
                        "id": 234243247,
350
                        "action": "create",
351
                        "frequency": "1 * * * *",
352
                    }
353
                ),
354
                CircuitSchedule.from_dict(
355
                    {
356
                        "id": 234243239,
357
                        "action": "create",
358
                        "interval": {"hours": 2},
359
                    }
360
                ),
361
            ],
362
            "enabled": True,
363
            "sb_priority": 2,
364
            "service_level": 7,
365
        }
366 1
        evc = EVC(**attributes)
367
368 1
        expected_dict = {
369
            "id": "custom_id",
370
            "name": "custom_name",
371
            "uni_a": attributes["uni_a"].as_dict(),
372
            "uni_z": attributes["uni_z"].as_dict(),
373
            "start_date": "2018-08-21T18:44:54",
374
            "end_date": "2018-08-21T18:44:55",
375
            "bandwidth": 0,
376
            "primary_links": [],
377
            "backup_links": [],
378
            "current_path": [],
379
            "primary_path": [],
380
            "backup_path": [],
381
            "dynamic_backup_path": False,
382
            "request_time": "2018-08-21T19:10:41",
383
            "creation_time": "2018-08-21T18:44:54",
384
            "circuit_scheduler": [
385
                {
386
                    "id": 234243247,
387
                    "action": "create",
388
                    "frequency": "1 * * * *",
389
                },
390
                {
391
                    "id": 234243239,
392
                    "action": "create",
393
                    "interval": {"hours": 2},
394
                },
395
            ],
396
            "active": False,
397
            "enabled": True,
398
            "sb_priority": 2,
399
            "service_level": 7,
400
        }
401 1
        actual_dict = evc.as_dict()
402 1
        for name, value in expected_dict.items():
403 1
            actual = actual_dict.get(name)
404 1
            assert value == actual
405
406
        # Selected fields
407 1
        expected_dict = {
408
            "enabled": True,
409
            "uni_z": attributes["uni_z"].as_dict(),
410
            "circuit_scheduler": [
411
                {
412
                    "id": 234243247,
413
                    "action": "create",
414
                    "frequency": "1 * * * *",
415
                },
416
                {
417
                    "id": 234243239,
418
                    "action": "create",
419
                    "interval": {"hours": 2},
420
                },
421
            ],
422
            "sb_priority": 2,
423
        }
424 1
        selected_fields = {
425
            "enabled", "uni_z", "circuit_scheduler", "sb_priority"
426
        }
427 1
        actual_dict = evc.as_dict(selected_fields)
428 1
        for name, value in expected_dict.items():
429 1
            actual = actual_dict.get(name)
430 1
            assert value == actual
431
432 1
    @staticmethod
433 1
    def test_get_id_from_cookie():
434
        """Test get_id_from_cookie."""
435 1
        attributes = {
436
            "controller": get_controller_mock(),
437
            "name": "circuit_name",
438
            "enable": True,
439
            "uni_a": get_uni_mocked(is_valid=True),
440
            "uni_z": get_uni_mocked(is_valid=True)
441
        }
442 1
        evc = EVC(**attributes)
443 1
        evc_id = evc.id
444 1
        assert evc_id
445 1
        assert evc.get_id_from_cookie(evc.get_cookie()) == evc_id
446
447 1
    @staticmethod
448 1
    def test_get_id_from_cookie_with_leading_zeros():
449
        """Test get_id_from_cookie with leading zeros."""
450
451 1
        attributes = {
452
            "controller": get_controller_mock(),
453
            "name": "circuit_name",
454
            "enable": True,
455
            "uni_a": get_uni_mocked(is_valid=True),
456
            "uni_z": get_uni_mocked(is_valid=True)
457
        }
458 1
        evc = EVC(**attributes)
459 1
        evc_id = "0a2d672d99ff41"
460
        # pylint: disable=protected-access
461 1
        evc._id = evc_id
462
        # pylint: enable=protected-access
463 1
        assert EVC.get_id_from_cookie(evc.get_cookie()) == evc_id
464
465 1
    def test_is_intra_switch(self):
466
        """Test is_intra_switch method."""
467 1
        attributes = {
468
            "controller": get_controller_mock(),
469
            "name": "circuit_name",
470
            "enable": True,
471
            "uni_a": get_uni_mocked(is_valid=True),
472
            "uni_z": get_uni_mocked(is_valid=True)
473
        }
474 1
        evc = EVC(**attributes)
475 1
        assert not evc.is_intra_switch()
476
477 1
        evc.uni_a.interface.switch = evc.uni_z.interface.switch
478 1
        assert evc.is_intra_switch()
479
480 1
    def test_default_queue_id(self):
481
        """Test default queue_id"""
482
483 1
        attributes = {
484
            "controller": get_controller_mock(),
485
            "name": "circuit_1",
486
            "uni_a": get_uni_mocked(is_valid=True),
487
            "uni_z": get_uni_mocked(is_valid=True),
488
            "dynamic_backup_path": True,
489
        }
490
491 1
        evc = EVC(**attributes)
492 1
        assert evc.queue_id == -1
493
494 1
    def test_get_unis_use_tags(self):
495
        """Test _get_unis_use_tags"""
496 1
        old_uni_a = get_uni_mocked(
497
            interface_port=2,
498
            is_valid=True
499
        )
500 1
        old_uni_z = get_uni_mocked(
501
            interface_port=3,
502
            is_valid=True
503
        )
504 1
        attributes = {
505
            "controller": get_controller_mock(),
506
            "name": "circuit_name",
507
            "enable": True,
508
            "uni_a": old_uni_a,
509
            "uni_z": old_uni_z
510
        }
511 1
        evc = EVC(**attributes)
512 1
        evc._use_uni_vlan = MagicMock()
513 1
        evc.make_uni_vlan_available = MagicMock()
514 1
        new_uni_a = get_uni_mocked(tag_value=200, is_valid=True)
515 1
        new_uni_z = get_uni_mocked(tag_value=200, is_valid=True)
516 1
        unis = {"uni_a": new_uni_a, "uni_z": evc.uni_z}
517 1
        evc._get_unis_use_tags(**unis)
518 1
        assert evc._use_uni_vlan.call_count == 1
519 1
        assert evc._use_uni_vlan.call_args[0][0] == new_uni_a
520 1
        assert evc.make_uni_vlan_available.call_count == 1
521 1
        assert evc.make_uni_vlan_available.call_args[0][0] == old_uni_a
522
523
        # Two UNIs
524 1
        evc = EVC(**attributes)
525 1
        evc._use_uni_vlan = MagicMock()
526 1
        evc.make_uni_vlan_available = MagicMock()
527 1
        unis = {"uni_a": new_uni_a, "uni_z": new_uni_z}
528 1
        evc._get_unis_use_tags(**unis)
529
530 1
        expected = [
531
            call(new_uni_a, uni_dif=old_uni_a),
532
            call(new_uni_z, uni_dif=old_uni_z)
533
        ]
534 1
        evc._use_uni_vlan.assert_has_calls(expected)
535 1
        expected = [
536
            call(old_uni_z, uni_dif=new_uni_z),
537
            call(old_uni_a, uni_dif=new_uni_a)
538
        ]
539 1
        evc.make_uni_vlan_available.assert_has_calls(expected)
540
541 1
    def test_get_unis_use_tags_error(self):
542
        """Test _get_unis_use_tags with KytosTagError"""
543 1
        old_uni_a = get_uni_mocked(
544
            interface_port=2,
545
            is_valid=True
546
        )
547 1
        old_uni_z = get_uni_mocked(
548
            interface_port=3,
549
            is_valid=True
550
        )
551 1
        attributes = {
552
            "controller": get_controller_mock(),
553
            "name": "circuit_name",
554
            "enable": True,
555
            "uni_a": old_uni_a,
556
            "uni_z": old_uni_z
557
        }
558 1
        evc = EVC(**attributes)
559 1
        evc._use_uni_vlan = MagicMock()
560
561
        # UNI Z KytosTagError
562 1
        evc._use_uni_vlan.side_effect = [None, KytosTagError("")]
563 1
        evc.make_uni_vlan_available = MagicMock()
564 1
        new_uni_a = get_uni_mocked(tag_value=200, is_valid=True)
565 1
        new_uni_z = get_uni_mocked(tag_value=200, is_valid=True)
566 1
        unis = {"uni_a": new_uni_a, "uni_z": new_uni_z}
567 1
        with pytest.raises(KytosTagError):
568 1
            evc._get_unis_use_tags(**unis)
569 1
        expected = [
570
            call(new_uni_a, uni_dif=old_uni_a),
571
            call(new_uni_z, uni_dif=old_uni_z)
572
        ]
573 1
        evc._use_uni_vlan.assert_has_calls(expected)
574 1
        assert evc.make_uni_vlan_available.call_count == 1
575 1
        assert evc.make_uni_vlan_available.call_args[0][0] == new_uni_a
576
577
        # UNI A KytosTagError
578 1
        evc = EVC(**attributes)
579 1
        evc._use_uni_vlan = MagicMock()
580 1
        evc._use_uni_vlan.side_effect = [KytosTagError(""), None]
581 1
        evc.make_uni_vlan_available = MagicMock()
582 1
        new_uni_a = get_uni_mocked(tag_value=200, is_valid=True)
583 1
        new_uni_z = get_uni_mocked(tag_value=200, is_valid=True)
584 1
        unis = {"uni_a": new_uni_a, "uni_z": new_uni_z}
585 1
        with pytest.raises(KytosTagError):
586 1
            evc._get_unis_use_tags(**unis)
587 1
        assert evc._use_uni_vlan.call_count == 1
588 1
        assert evc._use_uni_vlan.call_args[0][0] == new_uni_a
589 1
        assert evc.make_uni_vlan_available.call_count == 0
590
591 1
    @patch("napps.kytos.mef_eline.models.evc.range_difference")
592 1
    def test_use_uni_vlan(self, mock_difference):
593
        """Test _use_uni_vlan"""
594 1
        attributes = {
595
            "controller": get_controller_mock(),
596
            "name": "circuit_name",
597
            "enable": True,
598
            "uni_a": get_uni_mocked(is_valid=True),
599
            "uni_z": get_uni_mocked(is_valid=True)
600
        }
601 1
        evc = EVC(**attributes)
602 1
        uni = get_uni_mocked(is_valid=True)
603 1
        uni.interface.use_tags = MagicMock()
604 1
        evc._use_uni_vlan(uni)
605 1
        args = uni.interface.use_tags.call_args[0]
606 1
        assert args[1] == uni.user_tag.value
607 1
        assert args[2] == uni.user_tag.tag_type
608 1
        assert uni.interface.use_tags.call_count == 1
609
610 1
        uni.user_tag.value = "any"
611 1
        evc._use_uni_vlan(uni)
612 1
        assert uni.interface.use_tags.call_count == 2
613
614 1
        uni.user_tag.value = [[1, 10]]
615 1
        uni_dif = get_uni_mocked(tag_value=[[1, 2]])
616 1
        mock_difference.return_value = [[3, 10]]
617 1
        evc._use_uni_vlan(uni, uni_dif)
618 1
        assert uni.interface.use_tags.call_count == 3
619
620 1
        mock_difference.return_value = []
621 1
        evc._use_uni_vlan(uni, uni_dif)
622 1
        assert uni.interface.use_tags.call_count == 3
623
624 1
        uni.interface.use_tags.side_effect = KytosTagError("")
625 1
        with pytest.raises(KytosTagError):
626 1
            evc._use_uni_vlan(uni)
627 1
        assert uni.interface.use_tags.call_count == 4
628
629 1
        uni.user_tag = None
630 1
        evc._use_uni_vlan(uni)
631 1
        assert uni.interface.use_tags.call_count == 4
632
633 1
    @patch("napps.kytos.mef_eline.models.evc.log")
634 1
    def test_make_uni_vlan_available(self, mock_log):
635
        """Test make_uni_vlan_available"""
636 1
        attributes = {
637
            "controller": get_controller_mock(),
638
            "name": "circuit_name",
639
            "enable": True,
640
            "uni_a": get_uni_mocked(is_valid=True),
641
            "uni_z": get_uni_mocked(is_valid=True)
642
        }
643 1
        evc = EVC(**attributes)
644 1
        uni = get_uni_mocked(is_valid=True)
645 1
        uni.interface.make_tags_available = MagicMock()
646
647 1
        evc.make_uni_vlan_available(uni)
648 1
        args = uni.interface.make_tags_available.call_args[0]
649 1
        assert args[1] == uni.user_tag.value
650 1
        assert args[2] == uni.user_tag.tag_type
651 1
        assert uni.interface.make_tags_available.call_count == 1
652
653 1
        uni.user_tag.value = [[1, 10]]
654 1
        uni_dif = get_uni_mocked(tag_value=[[1, 2]])
655 1
        evc.make_uni_vlan_available(uni, uni_dif)
656 1
        assert uni.interface.make_tags_available.call_count == 2
657
658 1
        uni.interface.make_tags_available.side_effect = KytosTagError("")
659 1
        evc.make_uni_vlan_available(uni)
660 1
        assert mock_log.error.call_count == 1
661
662 1
        uni.user_tag = None
663 1
        evc.make_uni_vlan_available(uni)
664 1
        assert uni.interface.make_tags_available.call_count == 3
665
666 1
    def test_remove_uni_tags(self):
667
        """Test remove_uni_tags"""
668 1
        attributes = {
669
            "controller": get_controller_mock(),
670
            "name": "circuit_name",
671
            "enable": True,
672
            "uni_a": get_uni_mocked(is_valid=True),
673
            "uni_z": get_uni_mocked(is_valid=True)
674
        }
675 1
        evc = EVC(**attributes)
676 1
        evc.make_uni_vlan_available = MagicMock()
677 1
        evc.remove_uni_tags()
678 1
        assert evc.make_uni_vlan_available.call_count == 2
679
680 1
    def test_tag_lists_equal(self):
681
        """Test _tag_lists_equal"""
682 1
        attributes = {
683
            "controller": get_controller_mock(),
684
            "name": "circuit_name",
685
            "enable": True,
686
            "uni_a": get_uni_mocked(is_valid=True),
687
            "uni_z": get_uni_mocked(is_valid=True)
688
        }
689 1
        evc = EVC(**attributes)
690 1
        uni = MagicMock(user_tag=TAGRange("vlan", [[1, 10]]))
691 1
        update_dict = {"uni_z": uni}
692 1
        assert evc._tag_lists_equal(**update_dict) is False
693
694 1
        update_dict = {"uni_a": uni, "uni_z": uni}
695 1
        assert evc._tag_lists_equal(**update_dict)
696
697 1
    def test_check_no_tag_duplicate(self):
698
        """Test check_no_tag_duplicate"""
699 1
        uni_01_1 = get_uni_mocked(interface_port=1, switch_dpid="01")
700 1
        uni_01_1.user_tag = None
701 1
        uni_a = uni_01_1
702 1
        uni_z = get_uni_mocked(is_valid=True)
703 1
        attributes = {
704
            "controller": get_controller_mock(),
705
            "name": "circuit_name",
706
            "enable": True,
707
            "uni_a": uni_a,
708
            "uni_z": uni_z
709
        }
710 1
        evc = EVC(**attributes)
711 1
        other_uni = uni_01_1
712 1
        with pytest.raises(DuplicatedNoTagUNI):
713 1
            evc.check_no_tag_duplicate(other_uni)
714
715 1
        other_uni = get_uni_mocked(interface_port=1, switch_dpid="02")
716 1
        assert evc.check_no_tag_duplicate(other_uni) is None
717
718 1
    @patch("napps.kytos.mef_eline.models.EVC.sync")
719 1
    def test_update_invalid_path_uni(self, _sync_mock):
720
        """Test update by modifying a UNI
721
         so path is invalid."""
722 1
        uni_a = get_uni_mocked(is_valid=True, switch_id="00:01")
723 1
        uni_z = get_uni_mocked(is_valid=True, switch_id="00:02")
724 1
        link = MagicMock()
725 1
        link.endpoint_a.switch = uni_a.interface.switch
726 1
        link.endpoint_a.link = link
727 1
        link.endpoint_b.switch = uni_z.interface.switch
728 1
        link.endpoint_b.link = link
729 1
        primary_path = Path([link])
730 1
        attributes = {
731
            "controller": get_controller_mock(),
732
            "name": "circuit_name",
733
            "dynamic_backup_path": False,
734
            "primary_path": primary_path,
735
            "enable": True,
736
            "uni_a": uni_a,
737
            "uni_z": uni_z,
738
        }
739 1
        evc = EVC(**attributes)
740
741
        # Modify with valid UNI, no exception raised
742 1
        update = {"uni_a": uni_a}
743 1
        evc.update(**update)
744
745
        # Modify uni that makes path invalid
746 1
        new_uni_a = get_uni_mocked(is_valid=True, switch_id="00:03")
747 1
        update = {"uni_a": new_uni_a}
748 1
        with pytest.raises(ValueError):
749 1
            evc.update(**update)
750
751 1
    @patch("napps.kytos.mef_eline.models.EVC._get_unis_use_tags")
752 1
    def test_update_invalid_path_vlan(self, mock_use_tags):
753
        """Test update a VLAN with invalid path."""
754 1
        uni_a = get_uni_mocked(is_valid=True, switch_id="00:01")
755 1
        uni_z = get_uni_mocked(is_valid=True, switch_id="00:02")
756 1
        attributes = {
757
            "controller": get_controller_mock(),
758
            "name": "circuit_name",
759
            "dynamic_backup_path": False,
760
            "primary_path": MagicMock(),
761
            "enable": True,
762
            "uni_a": uni_a,
763
            "uni_z": uni_z,
764
        }
765 1
        evc = EVC(**attributes)
766
767 1
        new_primary_path = MagicMock()
768 1
        new_primary_path.is_valid.side_effect = InvalidPath()
769 1
        with pytest.raises(ValueError):
770 1
            evc.update(**{"primary_path": new_primary_path})
771
        mock_use_tags.assert_not_called()
772