Passed
Pull Request — master (#354)
by
unknown
03:41
created

TestEVC.test_default_queue_id()   A

Complexity

Conditions 1

Size

Total Lines 13
Code Lines 9

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 4
CRAP Score 1

Importance

Changes 0
Metric Value
cc 1
eloc 9
nop 1
dl 0
loc 13
ccs 4
cts 4
cp 1
crap 1
rs 9.95
c 0
b 0
f 0
1
"""Module to test the EVCBase class."""
2 1
import sys
3 1
from unittest.mock import MagicMock, patch
4 1
from napps.kytos.mef_eline.models import Path
5 1
import pytest
6
# pylint: disable=wrong-import-position
7 1
sys.path.insert(0, "/var/lib/kytos/napps/..")
8
# pylint: enable=wrong-import-position, disable=ungrouped-imports
9 1
from napps.kytos.mef_eline.models import EVC  # NOQA  pycodestyle
10 1
from napps.kytos.mef_eline.scheduler import (
11
    CircuitSchedule,
12
)  # NOQA  pycodestyle
13 1
from napps.kytos.mef_eline.tests.helpers import (
14
    get_uni_mocked,
15
    get_controller_mock,
16
)  # NOQA  pycodestyle
17
18
19 1
class TestEVC():  # pylint: disable=too-many-public-methods
20
    """Tests to verify EVC class."""
21
22 1
    def test_attributes_empty(self):
23
        """Test if the EVC raises an error with name is required."""
24 1
        attributes = {"controller": get_controller_mock()}
25 1
        error_message = "name is required."
26 1
        with pytest.raises(ValueError) as handle_error:
27 1
            EVC(**attributes)
28 1
        assert error_message in str(handle_error)
29
30 1
    def test_expected_requiring_redeploy_attributes(self) -> None:
31
        """Test expected attributes_requiring_redeploy."""
32 1
        expected = [
33
            "primary_path",
34
            "backup_path",
35
            "dynamic_backup_path",
36
            "queue_id",
37
            "sb_priority",
38
            "primary_constraints",
39
            "secondary_constraints",
40
            "uni_a",
41
            "uni_z",
42
        ]
43 1
        assert EVC.attributes_requiring_redeploy == expected
44
45 1
    def test_expeted_read_only_attributes(self) -> None:
46
        """Test expected read_only_attributes."""
47 1
        expected = [
48
            "creation_time",
49
            "active",
50
            "current_path",
51
            "failover_path",
52
            "_id",
53
            "archived",
54
        ]
55 1
        assert EVC.read_only_attributes == expected
56
57 1
    def test_without_uni_a(self):
58
        """Test if the EVC raises and error with UNI A is required."""
59 1
        attributes = {
60
            "controller": get_controller_mock(),
61
            "name": "circuit_name",
62
        }
63 1
        error_message = "uni_a is required."
64 1
        with pytest.raises(ValueError) as handle_error:
65 1
            EVC(**attributes)
66 1
        assert error_message in str(handle_error)
67
68 1
    def test_with_invalid_uni_a(self):
69
        """Test if the EVC raises and error with invalid UNI A."""
70 1
        attributes = {
71
            "controller": get_controller_mock(),
72
            "name": "circuit_name",
73
            "uni_a": get_uni_mocked(tag_value=82),
74
        }
75 1
        error_message = "VLAN tag 82 is not available in uni_a"
76 1
        with pytest.raises(ValueError) as handle_error:
77 1
            EVC(**attributes)
78 1
        assert error_message in str(handle_error)
79
80 1
    def test_without_uni_z(self):
81
        """Test if the EVC raises and error with UNI Z is required."""
82 1
        attributes = {
83
            "controller": get_controller_mock(),
84
            "name": "circuit_name",
85
            "uni_a": get_uni_mocked(is_valid=True),
86
        }
87 1
        error_message = "uni_z is required."
88 1
        with pytest.raises(ValueError) as handle_error:
89 1
            EVC(**attributes)
90 1
        assert error_message in str(handle_error)
91
92 1
    def test_with_invalid_uni_z(self):
93
        """Test if the EVC raises and error with UNI Z is required."""
94 1
        attributes = {
95
            "controller": get_controller_mock(),
96
            "name": "circuit_name",
97
            "uni_a": get_uni_mocked(is_valid=True),
98
            "uni_z": get_uni_mocked(tag_value=83),
99
        }
100 1
        error_message = "VLAN tag 83 is not available in uni_z"
101 1
        with pytest.raises(ValueError) as handle_error:
102 1
            EVC(**attributes)
103 1
        assert error_message in str(handle_error)
104
105 1
    @pytest.mark.parametrize(
106
        "name,value",
107
        [
108
            ("archived", True),
109
            ("_id", True),
110
            ("active", True),
111
            ("current_path", []),
112
            ("creation_time", "date"),
113
        ]
114
    )
115 1
    def test_update_read_only(self, name, value):
116
        """Test if raises an error when trying to update read only attr."""
117 1
        attributes = {
118
            "controller": get_controller_mock(),
119
            "name": "circuit_name",
120
            "dynamic_backup_path": True,
121
            "uni_a": get_uni_mocked(is_valid=True),
122
            "uni_z": get_uni_mocked(is_valid=True),
123
        }
124
125 1
        update_dict = {name: value}
126 1
        error_message = f"{name} can't be updated."
127 1
        with pytest.raises(ValueError) as handle_error:
128 1
            evc = EVC(**attributes)
129 1
            evc.update(**update_dict)
130 1
        assert error_message in str(handle_error)
131
132 1
    def test_update_invalid(self):
133
        """Test updating with an invalid attr"""
134 1
        attributes = {
135
            "controller": get_controller_mock(),
136
            "name": "circuit_name",
137
            "dynamic_backup_path": True,
138
            "uni_a": get_uni_mocked(is_valid=True),
139
            "uni_z": get_uni_mocked(is_valid=True),
140
        }
141 1
        evc = EVC(**attributes)
142 1
        with pytest.raises(ValueError) as handle_error:
143 1
            evc.update(xyz="abc")
144 1
        assert (
145
            'The attribute "xyz" is invalid.'
146
            in str(handle_error)
147
        )
148
149 1
    @patch("napps.kytos.mef_eline.models.EVC.sync")
150 1
    def test_update_disable(self, _sync_mock):
151
        """Test if evc is disabled."""
152 1
        attributes = {
153
            "controller": get_controller_mock(),
154
            "name": "circuit_name",
155
            "dynamic_backup_path": True,
156
            "enable": True,
157
            "uni_a": get_uni_mocked(is_valid=True),
158
            "uni_z": get_uni_mocked(is_valid=True),
159
        }
160 1
        update_dict = {"enable": False}
161 1
        evc = EVC(**attributes)
162 1
        evc.update(**update_dict)
163 1
        assert evc.is_enabled() is False
164
165 1
    @patch("napps.kytos.mef_eline.models.EVC.sync")
166 1
    def test_update_empty_primary_path(self, _sync_mock):
167
        """Test if an empty primary path can be set."""
168 1
        initial_primary_path = Path([MagicMock(id=1), MagicMock(id=2)])
169 1
        attributes = {
170
            "controller": get_controller_mock(),
171
            "name": "circuit_name",
172
            "dynamic_backup_path": True,
173
            "primary_path": initial_primary_path,
174
            "enable": True,
175
            "uni_a": get_uni_mocked(is_valid=True),
176
            "uni_z": get_uni_mocked(is_valid=True),
177
        }
178 1
        update_dict = {"primary_path": Path([])}
179 1
        evc = EVC(**attributes)
180 1
        assert evc.primary_path == initial_primary_path
181 1
        evc.update(**update_dict)
182 1
        assert len(evc.primary_path) == 0
183
184 1
    @patch("napps.kytos.mef_eline.models.EVC.sync")
185 1
    def test_update_empty_path_non_dynamic_backup(self, _sync_mock):
186
        """Test if an empty primary path can't be set if dynamic."""
187 1
        initial_primary_path = Path([MagicMock(id=1), MagicMock(id=2)])
188 1
        attributes = {
189
            "controller": get_controller_mock(),
190
            "name": "circuit_name",
191
            "dynamic_backup_path": False,
192
            "primary_path": initial_primary_path,
193
            "enable": True,
194
            "uni_a": get_uni_mocked(is_valid=True),
195
            "uni_z": get_uni_mocked(is_valid=True),
196
        }
197 1
        update_dict = {"primary_path": Path([])}
198 1
        evc = EVC(**attributes)
199 1
        assert evc.primary_path == initial_primary_path
200 1
        with pytest.raises(ValueError) as handle_error:
201 1
            evc.update(**update_dict)
202 1
        assert (
203
            'The EVC must have a primary path or allow dynamic paths.'
204
            in str(handle_error)
205
        )
206
207 1
    @patch("napps.kytos.mef_eline.models.EVC.sync")
208 1
    def test_update_empty_backup_path(self, _sync_mock):
209
        """Test if an empty backup path can be set."""
210 1
        initial_backup_path = Path([MagicMock(id=1), MagicMock(id=2)])
211 1
        attributes = {
212
            "controller": get_controller_mock(),
213
            "name": "circuit_name",
214
            "dynamic_backup_path": True,
215
            "backup_path": initial_backup_path,
216
            "enable": True,
217
            "uni_a": get_uni_mocked(is_valid=True),
218
            "uni_z": get_uni_mocked(is_valid=True),
219
        }
220 1
        update_dict = {"backup_path": Path([])}
221 1
        evc = EVC(**attributes)
222 1
        assert evc.backup_path == initial_backup_path
223 1
        evc.update(**update_dict)
224 1
        assert len(evc.backup_path) == 0
225
226 1
    @patch("napps.kytos.mef_eline.models.EVC.sync")
227 1
    def test_update_empty_backup_path_non_dynamic(self, _sync_mock):
228
        """Test if an empty backup path can be set even if it's non dynamic."""
229 1
        initial_backup_path = Path([MagicMock(id=1), MagicMock(id=2)])
230 1
        primary_path = Path([MagicMock(id=3), MagicMock(id=4)])
231 1
        attributes = {
232
            "controller": get_controller_mock(),
233
            "name": "circuit_name",
234
            "dynamic_backup_path": False,
235
            "primary_path": primary_path,
236
            "backup_path": initial_backup_path,
237
            "enable": True,
238
            "uni_a": get_uni_mocked(is_valid=True),
239
            "uni_z": get_uni_mocked(is_valid=True),
240
        }
241 1
        update_dict = {"backup_path": Path([])}
242 1
        evc = EVC(**attributes)
243 1
        assert evc.primary_path == primary_path
244 1
        assert evc.backup_path == initial_backup_path
245 1
        evc.update(**update_dict)
246 1
        assert evc.primary_path == primary_path
247 1
        assert len(evc.backup_path) == 0
248
249 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...
250 1
    def test_update_queue(self, _sync_mock):
251
        """Test if evc is set to redeploy."""
252 1
        attributes = {
253
            "controller": get_controller_mock(),
254
            "name": "circuit_name",
255
            "enable": True,
256
            "dynamic_backup_path": True,
257
            "uni_a": get_uni_mocked(is_valid=True),
258
            "uni_z": get_uni_mocked(is_valid=True),
259
        }
260 1
        update_dict = {"queue_id": 3}
261 1
        evc = EVC(**attributes)
262 1
        _, redeploy = evc.update(**update_dict)
263 1
        assert redeploy
264
265 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...
266 1
    def test_update_queue_null(self, _sync_mock):
267
        """Test if evc is set to redeploy."""
268 1
        attributes = {
269
            "controller": get_controller_mock(),
270
            "name": "circuit_name",
271
            "enable": True,
272
            "dynamic_backup_path": True,
273
            "uni_a": get_uni_mocked(is_valid=True),
274
            "uni_z": get_uni_mocked(is_valid=True),
275
        }
276 1
        update_dict = {"queue_id": None}
277 1
        evc = EVC(**attributes)
278 1
        _, redeploy = evc.update(**update_dict)
279 1
        assert redeploy
280
281 1
    def test_circuit_representation(self):
282
        """Test the method __repr__."""
283 1
        attributes = {
284
            "controller": get_controller_mock(),
285
            "name": "circuit_name",
286
            "uni_a": get_uni_mocked(is_valid=True),
287
            "uni_z": get_uni_mocked(is_valid=True),
288
        }
289 1
        evc = EVC(**attributes)
290 1
        expected_value = f"EVC({evc.id}, {evc.name})"
291 1
        assert str(evc) == expected_value
292
293 1
    def test_comparison_method(self):
294
        """Test the method __eq__."""
295 1
        attributes = {
296
            "controller": get_controller_mock(),
297
            "name": "circuit_name",
298
            "uni_a": get_uni_mocked(is_valid=True),
299
            "uni_z": get_uni_mocked(is_valid=True),
300
        }
301 1
        evc1 = EVC(**attributes)
302 1
        evc2 = EVC(**attributes)
303
304 1
        attributes = {
305
            "controller": get_controller_mock(),
306
            "name": "circuit_name_2",
307
            "uni_a": get_uni_mocked(is_valid=True),
308
            "uni_z": get_uni_mocked(is_valid=True),
309
        }
310 1
        evc3 = EVC(**attributes)
311 1
        evc4 = EVC(**attributes)
312
313 1
        assert evc1 == evc2
314 1
        assert evc1 != evc3
315 1
        assert evc2 != evc3
316 1
        assert evc3 == evc4
317
318 1
    def test_as_dict(self):
319
        """Test the method as_dict."""
320 1
        attributes = {
321
            "controller": get_controller_mock(),
322
            "id": "custom_id",
323
            "name": "custom_name",
324
            "uni_a": get_uni_mocked(is_valid=True),
325
            "uni_z": get_uni_mocked(is_valid=True),
326
            "start_date": "2018-08-21T18:44:54",
327
            "end_date": "2018-08-21T18:44:55",
328
            "primary_links": [],
329
            "request_time": "2018-08-21T19:10:41",
330
            "creation_time": "2018-08-21T18:44:54",
331
            "owner": "my_name",
332
            "circuit_scheduler": [
333
                CircuitSchedule.from_dict(
334
                    {
335
                        "id": 234243247,
336
                        "action": "create",
337
                        "frequency": "1 * * * *",
338
                    }
339
                ),
340
                CircuitSchedule.from_dict(
341
                    {
342
                        "id": 234243239,
343
                        "action": "create",
344
                        "interval": {"hours": 2},
345
                    }
346
                ),
347
            ],
348
            "enabled": True,
349
            "sb_priority": 2,
350
            "service_level": 7,
351
        }
352 1
        evc = EVC(**attributes)
353
354 1
        expected_dict = {
355
            "id": "custom_id",
356
            "name": "custom_name",
357
            "uni_a": attributes["uni_a"].as_dict(),
358
            "uni_z": attributes["uni_z"].as_dict(),
359
            "start_date": "2018-08-21T18:44:54",
360
            "end_date": "2018-08-21T18:44:55",
361
            "bandwidth": 0,
362
            "primary_links": [],
363
            "backup_links": [],
364
            "current_path": [],
365
            "primary_path": [],
366
            "backup_path": [],
367
            "dynamic_backup_path": False,
368
            "request_time": "2018-08-21T19:10:41",
369
            "creation_time": "2018-08-21T18:44:54",
370
            "circuit_scheduler": [
371
                {
372
                    "id": 234243247,
373
                    "action": "create",
374
                    "frequency": "1 * * * *",
375
                },
376
                {
377
                    "id": 234243239,
378
                    "action": "create",
379
                    "interval": {"hours": 2},
380
                },
381
            ],
382
            "active": False,
383
            "enabled": True,
384
            "sb_priority": 2,
385
            "service_level": 7,
386
        }
387 1
        actual_dict = evc.as_dict()
388 1
        for name, value in expected_dict.items():
389 1
            actual = actual_dict.get(name)
390 1
            assert value == actual
391
392
        # Selected fields
393 1
        expected_dict = {
394
            "enabled": True,
395
            "uni_z": attributes["uni_z"].as_dict(),
396
            "circuit_scheduler": [
397
                {
398
                    "id": 234243247,
399
                    "action": "create",
400
                    "frequency": "1 * * * *",
401
                },
402
                {
403
                    "id": 234243239,
404
                    "action": "create",
405
                    "interval": {"hours": 2},
406
                },
407
            ],
408
            "sb_priority": 2,
409
        }
410 1
        selected_fields = {
411
            "enabled", "uni_z", "circuit_scheduler", "sb_priority"
412
        }
413 1
        actual_dict = evc.as_dict(selected_fields)
414 1
        for name, value in expected_dict.items():
415 1
            actual = actual_dict.get(name)
416 1
            assert value == actual
417
418 1
    @staticmethod
419 1
    def test_get_id_from_cookie():
420
        """Test get_id_from_cookie."""
421 1
        attributes = {
422
            "controller": get_controller_mock(),
423
            "name": "circuit_name",
424
            "enable": True,
425
            "uni_a": get_uni_mocked(is_valid=True),
426
            "uni_z": get_uni_mocked(is_valid=True)
427
        }
428 1
        evc = EVC(**attributes)
429 1
        evc_id = evc.id
430 1
        assert evc_id
431 1
        assert evc.get_id_from_cookie(evc.get_cookie()) == evc_id
432
433 1
    @staticmethod
434 1
    def test_get_id_from_cookie_with_leading_zeros():
435
        """Test get_id_from_cookie with leading zeros."""
436
437 1
        attributes = {
438
            "controller": get_controller_mock(),
439
            "name": "circuit_name",
440
            "enable": True,
441
            "uni_a": get_uni_mocked(is_valid=True),
442
            "uni_z": get_uni_mocked(is_valid=True)
443
        }
444 1
        evc = EVC(**attributes)
445 1
        evc_id = "0a2d672d99ff41"
446
        # pylint: disable=protected-access
447 1
        evc._id = evc_id
448
        # pylint: enable=protected-access
449 1
        assert EVC.get_id_from_cookie(evc.get_cookie()) == evc_id
450
451 1
    def test_is_intra_switch(self):
452
        """Test is_intra_switch method."""
453 1
        attributes = {
454
            "controller": get_controller_mock(),
455
            "name": "circuit_name",
456
            "enable": True,
457
            "uni_a": get_uni_mocked(is_valid=True),
458
            "uni_z": get_uni_mocked(is_valid=True)
459
        }
460 1
        evc = EVC(**attributes)
461 1
        assert not evc.is_intra_switch()
462
463 1
        evc.uni_a.interface.switch = evc.uni_z.interface.switch
464 1
        assert evc.is_intra_switch()
465
466 1
    def test_default_queue_id(self):
467
        """Test default queue_id"""
468
469 1
        attributes = {
470
            "controller": get_controller_mock(),
471
            "name": "circuit_1",
472
            "uni_a": get_uni_mocked(is_valid=True),
473
            "uni_z": get_uni_mocked(is_valid=True),
474
            "dynamic_backup_path": True,
475
        }
476
477 1
        evc = EVC(**attributes)
478
        assert evc.queue_id == -1
479