Test Failed
Push — master ( a63869...63c1d9 )
by Antonio
04:17 queued 11s
created

build.tests.unit.models.test_evc_deploy   A

Complexity

Total Complexity 21

Size/Duplication

Total Lines 845
Duplicated Lines 0 %

Importance

Changes 0
Metric Value
eloc 605
dl 0
loc 845
rs 9.9949
c 0
b 0
f 0
wmc 21

18 Methods

Rating   Name   Duplication   Size   Complexity  
A TestEVC.test_prepare_pop_flow() 0 27 1
A TestEVC.test_prepare_push_flow() 0 38 4
B TestEVC.test_deploy_successfully() 0 73 1
B TestEVC.test_deploy_fail() 0 80 1
A TestEVC.test_should_deploy_case1() 0 15 1
B TestEVC.test_remove_current_flows() 0 65 1
A TestEVC.test_deploy_to_backup_path1() 0 34 1
A TestEVC.test_send_flow_mods_case1() 0 18 1
A TestEVC.test_send_flow_mods_case2() 0 17 1
A TestEVC.test_prepare_flow_mod() 0 25 1
B TestEVC.test_install_nni_flows() 0 70 1
A TestEVC.test_should_deploy_case4() 0 15 1
B TestEVC.test_deploy_error() 0 72 1
A TestEVC.test_primary_links_zipped() 0 2 1
A TestEVC.test_should_deploy_case2() 0 15 1
B TestEVC.test_deploy_without_path_case1() 0 79 1
A TestEVC.test_should_deploy_case3() 0 15 1
B TestEVC.test_install_uni_flows() 0 139 1
1
"""Method to thest EVCDeploy class."""
2
import sys
3
from unittest import TestCase
4
from unittest.mock import MagicMock, Mock, patch
5
6
from kytos.core.interface import Interface
7
from kytos.core.switch import Switch
8
9
# pylint: disable=wrong-import-position
10
sys.path.insert(0, "/var/lib/kytos/napps/..")
11
# pylint: enable=wrong-import-position
12
13
from napps.kytos.mef_eline.models import EVC, Path  # NOQA
14
from napps.kytos.mef_eline.settings import MANAGER_URL  # NOQA
15
from napps.kytos.mef_eline.exceptions import FlowModException  # NOQA
16
from napps.kytos.mef_eline.tests.helpers import (
17
    get_link_mocked,
18
    get_uni_mocked,
19
    get_controller_mock,
20
)  # NOQA
21
22
23
class TestEVC(TestCase):  # pylint: disable=too-many-public-methods
24
    """Tests to verify EVC class."""
25
26
    def test_primary_links_zipped(self):
27
        """Test primary links zipped method."""
28
29
    @staticmethod
30
    @patch("napps.kytos.mef_eline.models.evc.log")
31
    def test_should_deploy_case1(log_mock):
32
        """Test should deploy method without primary links."""
33
        log_mock.debug.return_value = True
34
        attributes = {
35
            "controller": get_controller_mock(),
36
            "name": "custom_name",
37
            "uni_a": get_uni_mocked(is_valid=True),
38
            "uni_z": get_uni_mocked(is_valid=True),
39
        }
40
41
        evc = EVC(**attributes)
42
        evc.should_deploy()
43
        log_mock.debug.assert_called_with("Path is empty.")
44
45
    @patch("napps.kytos.mef_eline.models.evc.log")
46
    def test_should_deploy_case2(self, log_mock):
47
        """Test should deploy method with disable circuit."""
48
        log_mock.debug.return_value = True
49
        attributes = {
50
            "controller": get_controller_mock(),
51
            "name": "custom_name",
52
            "uni_a": get_uni_mocked(is_valid=True),
53
            "uni_z": get_uni_mocked(is_valid=True),
54
            "primary_links": [get_link_mocked(), get_link_mocked()],
55
        }
56
        evc = EVC(**attributes)
57
58
        self.assertFalse(evc.should_deploy(attributes["primary_links"]))
59
        log_mock.debug.assert_called_with(f"{evc} is disabled.")
60
61
    @patch("napps.kytos.mef_eline.models.evc.log")
62
    def test_should_deploy_case3(self, log_mock):
63
        """Test should deploy method with enabled and not active circuit."""
64
        log_mock.debug.return_value = True
65
        attributes = {
66
            "controller": get_controller_mock(),
67
            "name": "custom_name",
68
            "uni_a": get_uni_mocked(is_valid=True),
69
            "uni_z": get_uni_mocked(is_valid=True),
70
            "primary_links": [get_link_mocked(), get_link_mocked()],
71
            "enabled": True,
72
        }
73
        evc = EVC(**attributes)
74
        self.assertTrue(evc.should_deploy(attributes["primary_links"]))
75
        log_mock.debug.assert_called_with(f"{evc} will be deployed.")
76
77
    @patch("napps.kytos.mef_eline.models.evc.log")
78
    def test_should_deploy_case4(self, log_mock):
79
        """Test should deploy method with enabled and active circuit."""
80
        log_mock.debug.return_value = True
81
        attributes = {
82
            "controller": get_controller_mock(),
83
            "name": "custom_name",
84
            "uni_a": get_uni_mocked(is_valid=True),
85
            "uni_z": get_uni_mocked(is_valid=True),
86
            "primary_links": [get_link_mocked(), get_link_mocked()],
87
            "enabled": True,
88
            "active": True,
89
        }
90
        evc = EVC(**attributes)
91
        self.assertFalse(evc.should_deploy(attributes["primary_links"]))
92
93
    @patch("napps.kytos.mef_eline.models.evc.requests")
94
    def test_send_flow_mods_case1(self, requests_mock):
95
        """Test if you are sending flow_mods."""
96
        flow_mods = {"id": 20}
97
        switch = Mock(spec=Switch, id=1)
98
99
        response = MagicMock()
100
        response.status_code = 201
101
        requests_mock.post.return_value = response
102
103
        # pylint: disable=protected-access
104
        EVC._send_flow_mods(switch, flow_mods)
105
106
        expected_endpoint = f"{MANAGER_URL}/flows/{switch.id}"
107
        expected_data = {"flows": flow_mods, "force": False}
108
        self.assertEqual(requests_mock.post.call_count, 1)
109
        requests_mock.post.assert_called_once_with(
110
            expected_endpoint, json=expected_data
111
        )
112
113
    @patch("napps.kytos.mef_eline.models.evc.requests")
114
    def test_send_flow_mods_case2(self, requests_mock):
115
        """Test if you are sending flow_mods."""
116
        flow_mods = {"id": 20}
117
        switch = Mock(spec=Switch, id=1)
118
        response = MagicMock()
119
        response.status_code = 201
120
        requests_mock.post.return_value = response
121
122
        # pylint: disable=protected-access
123
        EVC._send_flow_mods(switch, flow_mods, command='delete', force=True)
124
125
        expected_endpoint = f"{MANAGER_URL}/delete/{switch.id}"
126
        expected_data = {"flows": flow_mods, "force": True}
127
        self.assertEqual(requests_mock.post.call_count, 1)
128
        requests_mock.post.assert_called_once_with(
129
            expected_endpoint, json=expected_data
130
        )
131
132
    def test_prepare_flow_mod(self):
133
        """Test prepare flow_mod method."""
134
        interface_a = Interface("eth0", 1, Mock(spec=Switch))
135
        interface_z = Interface("eth1", 3, Mock(spec=Switch))
136
        attributes = {
137
            "controller": get_controller_mock(),
138
            "name": "custom_name",
139
            "uni_a": get_uni_mocked(is_valid=True),
140
            "uni_z": get_uni_mocked(is_valid=True),
141
            "primary_links": [get_link_mocked(), get_link_mocked()],
142
            "enabled": True,
143
            "active": True,
144
        }
145
        evc = EVC(**attributes)
146
147
        # pylint: disable=protected-access
148
        flow_mod = evc._prepare_flow_mod(interface_a, interface_z)
149
        expected_flow_mod = {
150
            "match": {"in_port": interface_a.port_number},
151
            "cookie": evc.get_cookie(),
152
            "actions": [
153
                {"action_type": "output", "port": interface_z.port_number}
154
            ],
155
        }
156
        self.assertEqual(expected_flow_mod, flow_mod)
157
158
    def test_prepare_pop_flow(self):
159
        """Test prepare pop flow  method."""
160
        attributes = {
161
            "controller": get_controller_mock(),
162
            "name": "custom_name",
163
            "uni_a": get_uni_mocked(interface_port=1, is_valid=True),
164
            "uni_z": get_uni_mocked(interface_port=2, is_valid=True),
165
        }
166
        evc = EVC(**attributes)
167
        interface_a = evc.uni_a.interface
168
        interface_z = evc.uni_z.interface
169
        in_vlan = 10
170
171
        # pylint: disable=protected-access
172
        flow_mod = evc._prepare_pop_flow(
173
            interface_a, interface_z, None, in_vlan
174
        )
175
176
        expected_flow_mod = {
177
            "match": {"in_port": interface_a.port_number, "dl_vlan": in_vlan},
178
            "cookie": evc.get_cookie(),
179
            "actions": [
180
                {"action_type": "pop_vlan"},
181
                {"action_type": "output", "port": interface_z.port_number},
182
            ],
183
        }
184
        self.assertEqual(expected_flow_mod, flow_mod)
185
186
    def test_prepare_push_flow(self):
187
        """Test prepare push flow method."""
188
        attributes = {
189
            "controller": get_controller_mock(),
190
            "name": "custom_name",
191
            "uni_a": get_uni_mocked(interface_port=1, is_valid=True),
192
            "uni_z": get_uni_mocked(interface_port=2, is_valid=True),
193
        }
194
        evc = EVC(**attributes)
195
        interface_a = evc.uni_a.interface
196
        interface_z = evc.uni_z.interface
197
        out_vlan_a = 20
198
199
        for in_vlan_a in (10, None):
200
            with self.subTest(in_vlan_a=in_vlan_a):
201
                # pylint: disable=protected-access
202
                flow_mod = evc._prepare_push_flow(
203
                    interface_a, interface_z, in_vlan_a, out_vlan_a
204
                )
205
206
                expected_flow_mod = {
207
                    "match": {"in_port": interface_a.port_number},
208
                    "cookie": evc.get_cookie(),
209
                    "actions": [
210
                        {"action_type": "push_vlan", "tag_type": "s"},
211
                        {"action_type": "set_vlan", "vlan_id": out_vlan_a},
212
                        {
213
                            "action_type": "output",
214
                            "port": interface_z.port_number,
215
                        },
216
                    ],
217
                }
218
                if in_vlan_a:
219
                    expected_flow_mod["match"]["dl_vlan"] = in_vlan_a
220
                    expected_flow_mod["actions"].insert(
221
                        0, {"action_type": "pop_vlan"}
222
                    )
223
                self.assertEqual(expected_flow_mod, flow_mod)
224
225
    @staticmethod
226
    @patch("napps.kytos.mef_eline.models.evc.EVC._send_flow_mods")
227
    def test_install_uni_flows(send_flow_mods_mock):
228
        """Test install uni flows method.
229
230
        This test will verify the flows send to the send_flow_mods method.
231
        """
232
        uni_a = get_uni_mocked(
233
            interface_port=2,
234
            tag_value=82,
235
            switch_id="switch_uni_a",
236
            is_valid=True,
237
        )
238
        uni_z = get_uni_mocked(
239
            interface_port=3,
240
            tag_value=83,
241
            switch_id="switch_uni_z",
242
            is_valid=True,
243
        )
244
245
        attributes = {
246
            "controller": get_controller_mock(),
247
            "name": "custom_name",
248
            "uni_a": uni_a,
249
            "uni_z": uni_z,
250
            "primary_links": [
251
                get_link_mocked(
252
                    endpoint_a_port=9,
253
                    endpoint_b_port=10,
254
                    metadata={"s_vlan": 5},
255
                ),
256
                get_link_mocked(
257
                    endpoint_a_port=11,
258
                    endpoint_b_port=12,
259
                    metadata={"s_vlan": 6},
260
                ),
261
            ],
262
        }
263
        evc = EVC(**attributes)
264
265
        # pylint: disable=protected-access
266
        evc._install_uni_flows(attributes["primary_links"])
267
268
        expected_flow_mod_a = [
269
            {
270
                "match": {
271
                    "in_port": uni_a.interface.port_number,
272
                    "dl_vlan": uni_a.user_tag.value,
273
                },
274
                "cookie": evc.get_cookie(),
275
                "actions": [
276
                    {"action_type": "pop_vlan"},
277
                    {"action_type": "push_vlan", "tag_type": "s"},
278
                    {
279
                        "action_type": "set_vlan",
280
                        "vlan_id": evc.primary_links[0]
281
                        .get_metadata("s_vlan")
282
                        .value,
283
                    },
284
                    {
285
                        "action_type": "output",
286
                        "port": evc.primary_links[0].endpoint_a.port_number,
287
                    },
288
                ],
289
            },
290
            {
291
                "match": {
292
                    "in_port": evc.primary_links[0].endpoint_a.port_number,
293
                    "dl_vlan": evc.primary_links[0]
294
                    .get_metadata("s_vlan")
295
                    .value,
296
                },
297
                "cookie": evc.get_cookie(),
298
                "actions": [
299
                    {"action_type": "pop_vlan"},
300
                    {"action_type": "push_vlan", "tag_type": "c"},
301
                    {
302
                        "action_type": "set_vlan",
303
                        "vlan_id": uni_a.user_tag.value,
304
                    },
305
                    {
306
                        "action_type": "output",
307
                        "port": uni_a.interface.port_number,
308
                    },
309
                ],
310
            },
311
        ]
312
313
        send_flow_mods_mock.assert_any_call(
314
            uni_a.interface.switch, expected_flow_mod_a
315
        )
316
317
        expected_flow_mod_z = [
318
            {
319
                "match": {
320
                    "in_port": uni_z.interface.port_number,
321
                    "dl_vlan": uni_z.user_tag.value,
322
                },
323
                "cookie": evc.get_cookie(),
324
                "actions": [
325
                    {"action_type": "pop_vlan"},
326
                    {"action_type": "push_vlan", "tag_type": "s"},
327
                    {
328
                        "action_type": "set_vlan",
329
                        "vlan_id": evc.primary_links[-1]
330
                        .get_metadata("s_vlan")
331
                        .value,
332
                    },
333
                    {
334
                        "action_type": "output",
335
                        "port": evc.primary_links[-1].endpoint_b.port_number,
336
                    },
337
                ],
338
            },
339
            {
340
                "match": {
341
                    "in_port": evc.primary_links[-1].endpoint_b.port_number,
342
                    "dl_vlan": evc.primary_links[-1]
343
                    .get_metadata("s_vlan")
344
                    .value,
345
                },
346
                "cookie": evc.get_cookie(),
347
                "actions": [
348
                    {"action_type": "pop_vlan"},
349
                    {"action_type": "push_vlan", "tag_type": "c"},
350
                    {
351
                        "action_type": "set_vlan",
352
                        "vlan_id": uni_z.user_tag.value,
353
                    },
354
                    {
355
                        "action_type": "output",
356
                        "port": uni_z.interface.port_number,
357
                    },
358
                ],
359
            },
360
        ]
361
362
        send_flow_mods_mock.assert_any_call(
363
            uni_z.interface.switch, expected_flow_mod_z
364
        )
365
366
    @staticmethod
367
    @patch("napps.kytos.mef_eline.models.evc.EVC._send_flow_mods")
368
    def test_install_nni_flows(send_flow_mods_mock):
369
        """Test install nni flows method.
370
371
        This test will verify the flows send to the send_flow_mods method.
372
        """
373
        uni_a = get_uni_mocked(
374
            interface_port=2,
375
            tag_value=82,
376
            switch_id="switch_uni_a",
377
            is_valid=True,
378
        )
379
        uni_z = get_uni_mocked(
380
            interface_port=3,
381
            tag_value=83,
382
            switch_id="switch_uni_z",
383
            is_valid=True,
384
        )
385
386
        attributes = {
387
            "controller": get_controller_mock(),
388
            "name": "custom_name",
389
            "uni_a": uni_a,
390
            "uni_z": uni_z,
391
            "primary_links": [
392
                get_link_mocked(
393
                    endpoint_a_port=9,
394
                    endpoint_b_port=10,
395
                    metadata={"s_vlan": 5},
396
                ),
397
                get_link_mocked(
398
                    endpoint_a_port=11,
399
                    endpoint_b_port=12,
400
                    metadata={"s_vlan": 6},
401
                ),
402
            ],
403
        }
404
        evc = EVC(**attributes)
405
406
        # pylint: disable=protected-access
407
        evc._install_nni_flows(attributes["primary_links"])
408
409
        in_vlan = evc.primary_links[0].get_metadata("s_vlan").value
410
        out_vlan = evc.primary_links[-1].get_metadata("s_vlan").value
411
412
        in_port = evc.primary_links[0].endpoint_b.port_number
413
        out_port = evc.primary_links[-1].endpoint_a.port_number
414
415
        expected_flow_mods = [
416
            {
417
                "match": {"in_port": in_port, "dl_vlan": in_vlan},
418
                "cookie": evc.get_cookie(),
419
                "actions": [
420
                    {"action_type": "set_vlan", "vlan_id": out_vlan},
421
                    {"action_type": "output", "port": out_port},
422
                ],
423
            },
424
            {
425
                "match": {"in_port": out_port, "dl_vlan": out_vlan},
426
                "cookie": evc.get_cookie(),
427
                "actions": [
428
                    {"action_type": "set_vlan", "vlan_id": in_vlan},
429
                    {"action_type": "output", "port": in_port},
430
                ],
431
            },
432
        ]
433
434
        switch = evc.primary_links[0].endpoint_b.switch
435
        send_flow_mods_mock.assert_called_once_with(switch, expected_flow_mods)
436
437
    @patch("requests.post")
438
    @patch("napps.kytos.mef_eline.storehouse.StoreHouse.save_evc")
439
    @patch("napps.kytos.mef_eline.models.evc.log")
440
    @patch("napps.kytos.mef_eline.models.path.Path.choose_vlans")
441
    @patch("napps.kytos.mef_eline.models.evc.EVC._install_nni_flows")
442
    @patch("napps.kytos.mef_eline.models.evc.EVC._install_uni_flows")
443
    @patch("napps.kytos.mef_eline.models.evc.EVC.activate")
444
    @patch("napps.kytos.mef_eline.models.evc.EVC.should_deploy")
445
    def test_deploy_successfully(self, *args):
446
        """Test if all methods to deploy are called."""
447
        # pylint: disable=too-many-locals
448
        (
449
            should_deploy_mock,
450
            activate_mock,
451
            install_uni_flows_mock,
452
            install_nni_flows,
453
            chose_vlans_mock,
454
            log_mock,
455
            _,
456
            requests_mock,
457
        ) = args
458
459
        response = MagicMock()
460
        response.status_code = 201
461
        requests_mock.return_value = response
462
463
        should_deploy_mock.return_value = True
464
        uni_a = get_uni_mocked(
465
            interface_port=2,
466
            tag_value=82,
467
            switch_id="switch_uni_a",
468
            is_valid=True,
469
        )
470
        uni_z = get_uni_mocked(
471
            interface_port=3,
472
            tag_value=83,
473
            switch_id="switch_uni_z",
474
            is_valid=True,
475
        )
476
        primary_links = [
477
            get_link_mocked(
478
                endpoint_a_port=9, endpoint_b_port=10, metadata={"s_vlan": 5}
479
            ),
480
            get_link_mocked(
481
                endpoint_a_port=11, endpoint_b_port=12, metadata={"s_vlan": 6}
482
            ),
483
        ]
484
485
        attributes = {
486
            "controller": get_controller_mock(),
487
            "name": "custom_name",
488
            "uni_a": uni_a,
489
            "uni_z": uni_z,
490
            "primary_links": primary_links,
491
            "queue_id": 5,
492
        }
493
494
        # Setup path to deploy
495
        path = Path()
496
        path.append(primary_links[0])
497
        path.append(primary_links[1])
498
499
        evc = EVC(**attributes)
500
501
        deployed = evc.deploy_to_path(path)
502
503
        self.assertEqual(should_deploy_mock.call_count, 1)
504
        self.assertEqual(activate_mock.call_count, 1)
505
        self.assertEqual(install_uni_flows_mock.call_count, 1)
506
        self.assertEqual(install_nni_flows.call_count, 1)
507
        self.assertEqual(chose_vlans_mock.call_count, 1)
508
        log_mock.info.assert_called_with(f"{evc} was deployed.")
509
        self.assertTrue(deployed)
510
511
    @patch("requests.post")
512
    @patch("napps.kytos.mef_eline.models.evc.log")
513
    @patch(
514
        "napps.kytos.mef_eline.models.evc.EVC.discover_new_paths",
515
        return_value=[],
516
    )
517
    @patch("napps.kytos.mef_eline.models.path.Path.choose_vlans")
518
    @patch("napps.kytos.mef_eline.models.evc.EVC._install_nni_flows")
519
    @patch("napps.kytos.mef_eline.models.evc.EVC._install_uni_flows")
520
    @patch("napps.kytos.mef_eline.models.evc.EVC.activate")
521
    @patch(
522
        "napps.kytos.mef_eline.models.evc.EVC.should_deploy",
523
        return_value=False,
524
    )
525
    @patch("napps.kytos.mef_eline.models.EVC.sync")
526
    def test_deploy_fail(self, *args):
527
        """Test if all methods is ignored when the should_deploy is false."""
528
        # pylint: disable=too-many-locals
529
        (
530
            sync_mock,
531
            should_deploy_mock,
532
            activate_mock,
533
            install_uni_flows_mock,
534
            install_nni_flows,
535
            choose_vlans_mock,
536
            discover_new_paths,
537
            log_mock,
538
            requests_mock,
539
        ) = args
540
541
        response = MagicMock()
542
        response.status_code = 201
543
        requests_mock.return_value = response
544
545
        uni_a = get_uni_mocked(
546
            interface_port=2,
547
            tag_value=82,
548
            switch_id="switch_uni_a",
549
            switch_dpid="switch_dpid_uni_a",
550
            is_valid=True,
551
        )
552
        uni_z = get_uni_mocked(
553
            interface_port=3,
554
            tag_value=83,
555
            switch_id="switch_uni_z",
556
            switch_dpid="switch_dpid_uni_a",
557
            is_valid=True,
558
        )
559
560
        attributes = {
561
            "controller": get_controller_mock(),
562
            "name": "custom_name",
563
            "uni_a": uni_a,
564
            "uni_z": uni_z,
565
            "primary_links": [
566
                get_link_mocked(
567
                    endpoint_a_port=9,
568
                    endpoint_b_port=10,
569
                    metadata={"s_vlan": 5},
570
                ),
571
                get_link_mocked(
572
                    endpoint_a_port=11,
573
                    endpoint_b_port=12,
574
                    metadata={"s_vlan": 6},
575
                ),
576
            ],
577
        }
578
579
        evc = EVC(**attributes)
580
        deployed = evc.deploy_to_path()
581
582
        self.assertEqual(discover_new_paths.call_count, 1)
583
        self.assertEqual(should_deploy_mock.call_count, 1)
584
        self.assertEqual(activate_mock.call_count, 0)
585
        self.assertEqual(install_uni_flows_mock.call_count, 0)
586
        self.assertEqual(install_nni_flows.call_count, 0)
587
        self.assertEqual(choose_vlans_mock.call_count, 0)
588
        self.assertEqual(log_mock.info.call_count, 0)
589
        self.assertEqual(sync_mock.call_count, 1)
590
        self.assertFalse(deployed)
591
592
    @patch("napps.kytos.mef_eline.models.evc.log")
593
    @patch(
594
        "napps.kytos.mef_eline.models.evc.EVC.discover_new_paths",
595
        return_value=[],
596
    )
597
    @patch("napps.kytos.mef_eline.models.path.Path.choose_vlans")
598
    @patch("napps.kytos.mef_eline.models.evc.EVC._install_nni_flows")
599
    @patch("napps.kytos.mef_eline.models.evc.EVC.should_deploy")
600
    @patch("napps.kytos.mef_eline.models.evc.EVC.remove_current_flows")
601
    @patch("napps.kytos.mef_eline.models.evc.EVC.sync")
602
    def test_deploy_error(self, *args):
603
        """Test if all methods is ignored when the should_deploy is false."""
604
        # pylint: disable=too-many-locals
605
        (
606
            sync_mock,
607
            remove_current_flows,
608
            should_deploy_mock,
609
            install_nni_flows,
610
            choose_vlans_mock,
611
            discover_new_paths,
612
            log_mock,
613
        ) = args
614
615
        install_nni_flows.side_effect = FlowModException
616
        should_deploy_mock.return_value = True
617
        uni_a = get_uni_mocked(
618
            interface_port=2,
619
            tag_value=82,
620
            switch_id="switch_uni_a",
621
            is_valid=True,
622
        )
623
        uni_z = get_uni_mocked(
624
            interface_port=3,
625
            tag_value=83,
626
            switch_id="switch_uni_z",
627
            is_valid=True,
628
        )
629
630
        primary_links = [
631
            get_link_mocked(
632
                endpoint_a_port=9, endpoint_b_port=10, metadata={"s_vlan": 5}
633
            ),
634
            get_link_mocked(
635
                endpoint_a_port=11, endpoint_b_port=12, metadata={"s_vlan": 6}
636
            ),
637
        ]
638
639
        attributes = {
640
            "controller": get_controller_mock(),
641
            "name": "custom_name",
642
            "uni_a": uni_a,
643
            "uni_z": uni_z,
644
            "primary_links": primary_links,
645
            "queue_id": 5,
646
        }
647
        # Setup path to deploy
648
        path = Path()
649
        path.append(primary_links[0])
650
        path.append(primary_links[1])
651
652
        evc = EVC(**attributes)
653
654
        deployed = evc.deploy_to_path(path)
655
656
        self.assertEqual(discover_new_paths.call_count, 0)
657
        self.assertEqual(should_deploy_mock.call_count, 1)
658
        self.assertEqual(install_nni_flows.call_count, 1)
659
        self.assertEqual(choose_vlans_mock.call_count, 1)
660
        self.assertEqual(log_mock.error.call_count, 1)
661
        self.assertEqual(sync_mock.call_count, 0)
662
        self.assertEqual(remove_current_flows.call_count, 2)
663
        self.assertFalse(deployed)
664
665
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy_to_path")
666
    @patch("napps.kytos.mef_eline.models.evc.EVC.discover_new_paths")
667
    def test_deploy_to_backup_path1(
668
        self, discover_new_paths_mocked, deploy_to_path_mocked
669
    ):
670
        """Test deployment when dynamic_backup_path is False in same switch"""
671
        uni_a = get_uni_mocked(interface_port=2, tag_value=82, is_valid=True)
672
        uni_z = get_uni_mocked(interface_port=3, tag_value=83, is_valid=True)
673
674
        switch = Mock(spec=Switch)
675
        uni_a.interface.switch = switch
676
        uni_z.interface.switch = switch
677
678
        attributes = {
679
            "controller": get_controller_mock(),
680
            "name": "custom_name",
681
            "uni_a": uni_a,
682
            "uni_z": uni_z,
683
            "enabled": True,
684
            "dynamic_backup_path": False,
685
        }
686
687
        evc = EVC(**attributes)
688
        discover_new_paths_mocked.return_value = []
689
        deploy_to_path_mocked.return_value = True
690
691
        # storehouse initialization mock
692
        evc._storehouse.box = Mock()  # pylint: disable=protected-access
693
        evc._storehouse.box.data = {}  # pylint: disable=protected-access
694
695
        deployed = evc.deploy_to_backup_path()
696
697
        deploy_to_path_mocked.assert_called_once_with()
698
        self.assertEqual(deployed, True)
699
700
    @patch("requests.post")
701
    @patch("napps.kytos.mef_eline.storehouse.StoreHouse.save_evc")
702
    @patch("napps.kytos.mef_eline.models.evc.log")
703
    @patch("napps.kytos.mef_eline.models.path.Path.choose_vlans")
704
    @patch("napps.kytos.mef_eline.models.evc.EVC._install_nni_flows")
705
    @patch("napps.kytos.mef_eline.models.evc.EVC._install_uni_flows")
706
    @patch("napps.kytos.mef_eline.models.evc.EVC.activate")
707
    @patch("napps.kytos.mef_eline.models.evc.EVC.should_deploy")
708
    @patch("napps.kytos.mef_eline.models.evc.EVC.discover_new_paths")
709
    def test_deploy_without_path_case1(self, *args):
710
        """Test if not path is found a dynamic path is used."""
711
        # pylint: disable=too-many-locals
712
        (
713
            discover_new_paths_mocked,
714
            should_deploy_mock,
715
            activate_mock,
716
            install_uni_flows_mock,
717
            install_nni_flows,
718
            chose_vlans_mock,
719
            log_mock,
720
            _,
721
            requests_mock,
722
        ) = args
723
724
        response = MagicMock()
725
        response.status_code = 201
726
        requests_mock.return_value = response
727
728
        should_deploy_mock.return_value = False
729
        uni_a = get_uni_mocked(
730
            interface_port=2,
731
            tag_value=82,
732
            switch_id="switch_uni_a",
733
            is_valid=True,
734
        )
735
        uni_z = get_uni_mocked(
736
            interface_port=3,
737
            tag_value=83,
738
            switch_id="switch_uni_z",
739
            is_valid=True,
740
        )
741
742
        attributes = {
743
            "controller": get_controller_mock(),
744
            "name": "custom_name",
745
            "uni_a": uni_a,
746
            "uni_z": uni_z,
747
            "enabled": True,
748
            "dynamic_backup_path": False,
749
        }
750
751
        dynamic_backup_path = Path(
752
            [
753
                get_link_mocked(
754
                    endpoint_a_port=9,
755
                    endpoint_b_port=10,
756
                    metadata={"s_vlan": 5},
757
                ),
758
                get_link_mocked(
759
                    endpoint_a_port=11,
760
                    endpoint_b_port=12,
761
                    metadata={"s_vlan": 6},
762
                ),
763
            ]
764
        )
765
766
        evc = EVC(**attributes)
767
        discover_new_paths_mocked.return_value = [dynamic_backup_path]
768
769
        deployed = evc.deploy_to_path()
770
771
        self.assertEqual(should_deploy_mock.call_count, 1)
772
        self.assertEqual(discover_new_paths_mocked.call_count, 1)
773
        self.assertEqual(activate_mock.call_count, 1)
774
        self.assertEqual(install_uni_flows_mock.call_count, 1)
775
        self.assertEqual(install_nni_flows.call_count, 1)
776
        self.assertEqual(chose_vlans_mock.call_count, 1)
777
        log_mock.info.assert_called_with(f"{evc} was deployed.")
778
        self.assertTrue(deployed)
779
780
    @patch("napps.kytos.mef_eline.models.evc.EVC._send_flow_mods")
781
    def test_remove_current_flows(self, send_flow_mods_mocked):
782
        """Test remove current flows."""
783
        uni_a = get_uni_mocked(
784
            interface_port=2,
785
            tag_value=82,
786
            switch_id="switch_uni_a",
787
            is_valid=True,
788
        )
789
        uni_z = get_uni_mocked(
790
            interface_port=3,
791
            tag_value=83,
792
            switch_id="switch_uni_z",
793
            is_valid=True,
794
        )
795
796
        switch_a = Switch("00:00:00:00:00:01")
797
        switch_b = Switch("00:00:00:00:00:02")
798
        switch_c = Switch("00:00:00:00:00:03")
799
800
        attributes = {
801
            "controller": get_controller_mock(),
802
            "name": "custom_name",
803
            "uni_a": uni_a,
804
            "uni_z": uni_z,
805
            "active": True,
806
            "enabled": True,
807
            "primary_links": [
808
                get_link_mocked(
809
                    switch_a=switch_a,
810
                    switch_b=switch_b,
811
                    endpoint_a_port=9,
812
                    endpoint_b_port=10,
813
                    metadata={"s_vlan": 5},
814
                ),
815
                get_link_mocked(
816
                    switch_a=switch_b,
817
                    switch_b=switch_c,
818
                    endpoint_a_port=11,
819
                    endpoint_b_port=12,
820
                    metadata={"s_vlan": 6},
821
                ),
822
            ],
823
        }
824
825
        evc = EVC(**attributes)
826
827
        # storehouse initialization mock
828
        evc._storehouse.box = Mock()  # pylint: disable=protected-access
829
        evc._storehouse.box.data = {}  # pylint: disable=protected-access
830
831
        evc.current_path = evc.primary_links
832
        evc.remove_current_flows()
833
834
        self.assertEqual(send_flow_mods_mocked.call_count, 5)
835
        self.assertFalse(evc.is_active())
836
        flows = [
837
            {"cookie": evc.get_cookie(), "cookie_mask": 18446744073709551615}
838
        ]
839
        switch_1 = evc.primary_links[0].endpoint_a.switch
840
        switch_2 = evc.primary_links[0].endpoint_b.switch
841
        send_flow_mods_mocked.assert_any_call(switch_1, flows, 'delete',
842
                                              force=True)
843
        send_flow_mods_mocked.assert_any_call(switch_2, flows, 'delete',
844
                                              force=True)
845