Passed
Pull Request — master (#246)
by Italo Valcy
03:24
created

TestEVC.test_circuit_representation()   A

Complexity

Conditions 1

Size

Total Lines 11
Code Lines 9

Duplication

Lines 0
Ratio 0 %

Importance

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