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

TestEVC.test_as_dict()   B

Complexity

Conditions 2

Size

Total Lines 73
Code Lines 59

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 8
CRAP Score 2

Importance

Changes 0
Metric Value
cc 2
eloc 59
nop 1
dl 0
loc 73
ccs 8
cts 8
cp 1
crap 2
rs 8.3417
c 0
b 0
f 0

How to fix   Long Method   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

1
"""Module to test the EVCBase class."""
2 1
import sys
3 1
from unittest import TestCase
4 1
from unittest.mock import MagicMock, patch
5 1
from napps.kytos.mef_eline.models import Path
6
7
# pylint: disable=wrong-import-position
8 1
sys.path.insert(0, "/var/lib/kytos/napps/..")
9
# pylint: enable=wrong-import-position
10 1
from napps.kytos.mef_eline.models import EVC  # NOQA  pycodestyle
11 1
from napps.kytos.mef_eline.scheduler import (
12
    CircuitSchedule,
13
)  # NOQA  pycodestyle
14 1
from napps.kytos.mef_eline.tests.helpers import (
15
    get_uni_mocked,
16
    get_controller_mock,
17
)  # NOQA  pycodestyle
18
19
20 1
class TestEVC(TestCase):  # pylint: disable=too-many-public-methods
21
    """Tests to verify EVC class."""
22
23 1
    def test_attributes_empty(self):
24
        """Test if the EVC raises an error with name is required."""
25 1
        attributes = {"controller": get_controller_mock()}
26 1
        error_message = "name is required."
27 1
        with self.assertRaises(ValueError) as handle_error:
28 1
            EVC(**attributes)
29 1
        self.assertEqual(str(handle_error.exception), error_message)
30
31 1
    def test_expected_requiring_redeploy_attributes(self) -> None:
32
        """Test expected attributes_requiring_redeploy."""
33 1
        expected = [
34
            "primary_path",
35
            "backup_path",
36
            "dynamic_backup_path",
37
            "queue_id",
38
            "sb_priority",
39
            "primary_constraints",
40
            "secondary_constraints",
41
            "uni_a",
42
            "uni_z",
43
        ]
44 1
        assert EVC.attributes_requiring_redeploy == expected
45
46 1
    def test_expeted_read_only_attributes(self) -> None:
47
        """Test expected read_only_attributes."""
48 1
        expected = [
49
            "creation_time",
50
            "active",
51
            "current_path",
52
            "failover_path",
53
            "_id",
54
            "archived",
55
        ]
56 1
        assert EVC.read_only_attributes == expected
57
58 1
    def test_without_uni_a(self):
59
        """Test if the EVC raises and error with UNI A is required."""
60 1
        attributes = {
61
            "controller": get_controller_mock(),
62
            "name": "circuit_name",
63
        }
64 1
        error_message = "uni_a is required."
65 1
        with self.assertRaises(ValueError) as handle_error:
66 1
            EVC(**attributes)
67 1
        self.assertEqual(str(handle_error.exception), error_message)
68
69 1
    def test_with_invalid_uni_a(self):
70
        """Test if the EVC raises and error with invalid UNI A."""
71 1
        attributes = {
72
            "controller": get_controller_mock(),
73
            "name": "circuit_name",
74
            "uni_a": get_uni_mocked(tag_value=82),
75
        }
76 1
        error_message = "VLAN tag 82 is not available in uni_a"
77 1
        with self.assertRaises(ValueError) as handle_error:
78 1
            EVC(**attributes)
79 1
        self.assertEqual(str(handle_error.exception), error_message)
80
81 1
    def test_without_uni_z(self):
82
        """Test if the EVC raises and error with UNI Z is required."""
83 1
        attributes = {
84
            "controller": get_controller_mock(),
85
            "name": "circuit_name",
86
            "uni_a": get_uni_mocked(is_valid=True),
87
        }
88 1
        error_message = "uni_z is required."
89 1
        with self.assertRaises(ValueError) as handle_error:
90 1
            EVC(**attributes)
91 1
        self.assertEqual(str(handle_error.exception), error_message)
92
93 1
    def test_with_invalid_uni_z(self):
94
        """Test if the EVC raises and error with UNI Z is required."""
95 1
        attributes = {
96
            "controller": get_controller_mock(),
97
            "name": "circuit_name",
98
            "uni_a": get_uni_mocked(is_valid=True),
99
            "uni_z": get_uni_mocked(tag_value=83),
100
        }
101 1
        error_message = "VLAN tag 83 is not available in uni_z"
102 1
        with self.assertRaises(ValueError) as handle_error:
103 1
            EVC(**attributes)
104 1
        self.assertEqual(str(handle_error.exception), error_message)
105
106 1
    def test_update_read_only(self):
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 1
        update_attr = [
116
            ("archived", True),
117
            ("_id", True),
118
            ("active", True),
119
            ("current_path", []),
120
            ("creation_time", "date"),
121
        ]
122
123 1
        for name, value in update_attr:
124 1
            with self.subTest(name=name, value=value):
125 1
                update_dict = {name: value}
126 1
                error_message = f"{name} can't be updated."
127 1
                with self.assertRaises(ValueError) as handle_error:
128 1
                    evc = EVC(**attributes)
129 1
                    evc.update(**update_dict)
130 1
                self.assertEqual(str(handle_error.exception), error_message)
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 self.assertRaises(ValueError) as handle_error:
143 1
            evc.update(xyz="abc")
144 1
        self.assertEqual(
145
            str(handle_error.exception),
146
            'The attribute "xyz" is invalid.'
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
        self.assertIs(evc.is_enabled(), 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
        self.assertEqual(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
        self.assertEqual(evc.primary_path, initial_primary_path)
200 1
        with self.assertRaises(ValueError) as handle_error:
201 1
            evc.update(**update_dict)
202 1
        self.assertEqual(
203
            str(handle_error.exception),
204
            'The EVC must have a primary path or allow dynamic paths.'
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
        self.assertEqual(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
        self.assertEqual(evc.primary_path, primary_path)
244 1
        self.assertEqual(evc.backup_path, initial_backup_path)
245 1
        evc.update(**update_dict)
246 1
        self.assertEqual(evc.primary_path, primary_path)
247 1
        self.assertEqual(len(evc.backup_path), 0)
248
249 1
    @patch("napps.kytos.mef_eline.models.EVC.sync")
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
        self.assertTrue(redeploy)
264
265 1
    def test_circuit_representation(self):
266
        """Test the method __repr__."""
267 1
        attributes = {
268
            "controller": get_controller_mock(),
269
            "name": "circuit_name",
270
            "uni_a": get_uni_mocked(is_valid=True),
271
            "uni_z": get_uni_mocked(is_valid=True),
272
        }
273 1
        evc = EVC(**attributes)
274 1
        expected_value = f"EVC({evc.id}, {evc.name})"
275 1
        self.assertEqual(str(evc), expected_value)
276
277 1
    def test_comparison_method(self):
278
        """Test the method __eq__."""
279 1
        attributes = {
280
            "controller": get_controller_mock(),
281
            "name": "circuit_name",
282
            "uni_a": get_uni_mocked(is_valid=True),
283
            "uni_z": get_uni_mocked(is_valid=True),
284
        }
285 1
        evc1 = EVC(**attributes)
286 1
        evc2 = EVC(**attributes)
287
288 1
        attributes = {
289
            "controller": get_controller_mock(),
290
            "name": "circuit_name_2",
291
            "uni_a": get_uni_mocked(is_valid=True),
292
            "uni_z": get_uni_mocked(is_valid=True),
293
        }
294 1
        evc3 = EVC(**attributes)
295 1
        evc4 = EVC(**attributes)
296
297 1
        self.assertEqual(evc1 == evc2, True)
298 1
        self.assertEqual(evc1 == evc3, False)
299 1
        self.assertEqual(evc2 == evc3, False)
300 1
        self.assertEqual(evc3 == evc4, True)
301
302 1
    def test_as_dict(self):
303
        """Test the method as_dict."""
304 1
        attributes = {
305
            "controller": get_controller_mock(),
306
            "id": "custom_id",
307
            "name": "custom_name",
308
            "uni_a": get_uni_mocked(is_valid=True),
309
            "uni_z": get_uni_mocked(is_valid=True),
310
            "start_date": "2018-08-21T18:44:54",
311
            "end_date": "2018-08-21T18:44:55",
312
            "primary_links": [],
313
            "request_time": "2018-08-21T19:10:41",
314
            "creation_time": "2018-08-21T18:44:54",
315
            "owner": "my_name",
316
            "circuit_scheduler": [
317
                CircuitSchedule.from_dict(
318
                    {
319
                        "id": 234243247,
320
                        "action": "create",
321
                        "frequency": "1 * * * *",
322
                    }
323
                ),
324
                CircuitSchedule.from_dict(
325
                    {
326
                        "id": 234243239,
327
                        "action": "create",
328
                        "interval": {"hours": 2},
329
                    }
330
                ),
331
            ],
332
            "enabled": True,
333
            "sb_priority": 2,
334
            "service_level": 7,
335
        }
336 1
        evc = EVC(**attributes)
337
338 1
        expected_dict = {
339
            "id": "custom_id",
340
            "name": "custom_name",
341
            "uni_a": attributes["uni_a"].as_dict(),
342
            "uni_z": attributes["uni_z"].as_dict(),
343
            "start_date": "2018-08-21T18:44:54",
344
            "end_date": "2018-08-21T18:44:55",
345
            "bandwidth": 0,
346
            "primary_links": [],
347
            "backup_links": [],
348
            "current_path": [],
349
            "primary_path": [],
350
            "backup_path": [],
351
            "dynamic_backup_path": False,
352
            "request_time": "2018-08-21T19:10:41",
353
            "creation_time": "2018-08-21T18:44:54",
354
            "circuit_scheduler": [
355
                {
356
                    "id": 234243247,
357
                    "action": "create",
358
                    "frequency": "1 * * * *",
359
                },
360
                {
361
                    "id": 234243239,
362
                    "action": "create",
363
                    "interval": {"hours": 2},
364
                },
365
            ],
366
            "active": False,
367
            "enabled": True,
368
            "sb_priority": 2,
369
            "service_level": 7,
370
        }
371 1
        actual_dict = evc.as_dict()
372 1
        for name, value in expected_dict.items():
373 1
            actual = actual_dict.get(name)
374 1
            self.assertEqual(value, actual)
375
376 1
    @staticmethod
377 1
    def test_get_id_from_cookie():
378
        """Test get_id_from_cookie."""
379 1
        attributes = {
380
            "controller": get_controller_mock(),
381
            "name": "circuit_name",
382
            "enable": True,
383
            "uni_a": get_uni_mocked(is_valid=True),
384
            "uni_z": get_uni_mocked(is_valid=True)
385
        }
386 1
        evc = EVC(**attributes)
387 1
        evc_id = evc.id
388 1
        assert evc_id
389 1
        assert evc.get_id_from_cookie(evc.get_cookie()) == evc_id
390
391 1
    @staticmethod
392 1
    def test_get_id_from_cookie_with_leading_zeros():
393
        """Test get_id_from_cookie with leading zeros."""
394
395 1
        attributes = {
396
            "controller": get_controller_mock(),
397
            "name": "circuit_name",
398
            "enable": True,
399
            "uni_a": get_uni_mocked(is_valid=True),
400
            "uni_z": get_uni_mocked(is_valid=True)
401
        }
402 1
        evc = EVC(**attributes)
403 1
        evc_id = "0a2d672d99ff41"
404
        # pylint: disable=protected-access
405 1
        evc._id = evc_id
406
        # pylint: enable=protected-access
407 1
        assert EVC.get_id_from_cookie(evc.get_cookie()) == evc_id
408
409 1
    def test_is_intra_switch(self):
410
        """Test is_intra_switch method."""
411 1
        attributes = {
412
            "controller": get_controller_mock(),
413
            "name": "circuit_name",
414
            "enable": True,
415
            "uni_a": get_uni_mocked(is_valid=True),
416
            "uni_z": get_uni_mocked(is_valid=True)
417
        }
418 1
        evc = EVC(**attributes)
419 1
        assert not evc.is_intra_switch()
420
421 1
        evc.uni_a.interface.switch = evc.uni_z.interface.switch
422 1
        assert evc.is_intra_switch()
423
424 1
    @patch('napps.kytos.mef_eline.models.evc.settings')
425 1
    def test_default_queue_id(self, mock_queue_id):
426
        """Test default queue_id"""
427
428 1
        attributes = {
429
            "controller": get_controller_mock(),
430
            "name": "circuit_1",
431
            "uni_a": get_uni_mocked(is_valid=True),
432
            "uni_z": get_uni_mocked(is_valid=True),
433
            "dynamic_backup_path": True,
434
        }
435
436 1
        mock_queue_id.QUEUE_ID = 2
437 1
        evc = EVC(**attributes)
438
        assert evc.queue_id == 2
439