TestMain._uni_from_dict_side_effect()   A
last analyzed

Complexity

Conditions 1

Size

Total Lines 5
Code Lines 5

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 5
CRAP Score 1

Importance

Changes 0
Metric Value
cc 1
eloc 5
nop 2
dl 0
loc 5
ccs 5
cts 5
cp 1
crap 1
rs 10
c 0
b 0
f 0
1
"""Module to test the main napp file."""
2 1
import asyncio
3 1
from unittest.mock import (AsyncMock, MagicMock, Mock, call,
4
                           create_autospec, patch)
5
6 1
import pytest
7 1
from kytos.lib.helpers import get_controller_mock, get_test_client
8 1
from kytos.core.helpers import now
9 1
from kytos.core.common import EntityStatus
10 1
from kytos.core.events import KytosEvent
11 1
from kytos.core.exceptions import KytosTagError
12 1
from kytos.core.interface import TAGRange, UNI, Interface
13 1
from napps.kytos.mef_eline.exceptions import FlowModException, InvalidPath
14 1
from napps.kytos.mef_eline.models import EVC, Path
15 1
from napps.kytos.mef_eline.tests.helpers import get_uni_mocked
16
17
18 1
async def test_on_table_enabled():
19
    """Test on_table_enabled"""
20
    # pylint: disable=import-outside-toplevel
21 1
    from napps.kytos.mef_eline.main import Main
22 1
    controller = get_controller_mock()
23 1
    controller.buffers.app.aput = AsyncMock()
24 1
    Main.get_eline_controller = MagicMock()
25 1
    napp = Main(controller)
26
27
    # Succesfully setting table groups
28 1
    content = {"mef_eline": {"epl": 2}}
29 1
    event = KytosEvent(name="kytos/of_multi_table.enable_table",
30
                       content=content)
31 1
    await napp.on_table_enabled(event)
32 1
    assert napp.table_group["epl"] == 2
33 1
    assert napp.table_group["evpl"] == 0
34 1
    assert controller.buffers.app.aput.call_count == 1
35
36
    # Failure at setting table groups
37 1
    content = {"mef_eline": {"unknown": 123}}
38 1
    event = KytosEvent(name="kytos/of_multi_table.enable_table",
39
                       content=content)
40 1
    await napp.on_table_enabled(event)
41 1
    assert controller.buffers.app.aput.call_count == 1
42
43
    # Failure with early return
44 1
    content = {}
45 1
    event = KytosEvent(name="kytos/of_multi_table.enable_table",
46
                       content=content)
47 1
    await napp.on_table_enabled(event)
48 1
    assert controller.buffers.app.aput.call_count == 1
49
50
51
# pylint: disable=too-many-public-methods, too-many-lines
52
# pylint: disable=too-many-arguments,too-many-locals
53 1
class TestMain:
54
    """Test the Main class."""
55
56 1
    def setup_method(self):
57
        """Execute steps before each tests.
58
59
        Set the server_name_url_url from kytos/mef_eline
60
        """
61
62
        # The decorator run_on_thread is patched, so methods that listen
63
        # for events do not run on threads while tested.
64
        # Decorators have to be patched before the methods that are
65
        # decorated with them are imported.
66 1
        patch("kytos.core.helpers.run_on_thread", lambda x: x).start()
67
        # pylint: disable=import-outside-toplevel
68 1
        from napps.kytos.mef_eline.main import Main
69 1
        Main.get_eline_controller = MagicMock()
70 1
        controller = get_controller_mock()
71 1
        self.napp = Main(controller)
72 1
        self.api_client = get_test_client(controller, self.napp)
73 1
        self.base_endpoint = "kytos/mef_eline"
74
75 1
    def test_get_event_listeners(self):
76
        """Verify all event listeners registered."""
77 1
        expected_events = [
78
            "kytos/core.shutdown",
79
            "kytos/core.shutdown.kytos/mef_eline",
80
            "kytos/topology.link_up",
81
            "kytos/topology.link_down",
82
        ]
83 1
        actual_events = self.napp.listeners()
84
85 1
        for _event in expected_events:
86 1
            assert _event in actual_events, _event
87
88 1
    @patch('napps.kytos.mef_eline.main.log')
89 1
    @patch('napps.kytos.mef_eline.main.Main.execute_consistency')
90 1
    def test_execute(self, mock_execute_consistency, mock_log):
91
        """Test execute."""
92 1
        self.napp.execution_rounds = 0
93 1
        self.napp.execute()
94 1
        mock_execute_consistency.assert_called()
95 1
        assert mock_log.debug.call_count == 2
96
97
        # Test locked should return
98 1
        mock_execute_consistency.call_count = 0
99 1
        mock_log.info.call_count = 0
100
        # pylint: disable=protected-access
101 1
        self.napp._lock = MagicMock()
102 1
        self.napp._lock.locked.return_value = True
103
        # pylint: enable=protected-access
104 1
        self.napp.execute()
105 1
        mock_execute_consistency.assert_not_called()
106 1
        mock_log.info.assert_not_called()
107
108 1
    @patch('napps.kytos.mef_eline.main.settings')
109 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
110 1
    @patch("napps.kytos.mef_eline.models.evc.EVCDeploy.check_list_traces")
111 1
    def test_execute_consistency(self, mock_check_list_traces, *args):
112
        """Test execute_consistency."""
113 1
        (mongo_controller_upsert_mock, mock_settings) = args
114
115 1
        stored_circuits = {'1': {'name': 'circuit_1'},
116
                           '2': {'name': 'circuit_2'},
117
                           '3': {'name': 'circuit_3'}}
118 1
        mongo_controller_upsert_mock.return_value = True
119 1
        self.napp.mongo_controller.get_circuits.return_value = {
120
            "circuits": stored_circuits
121
        }
122
123 1
        mock_settings.WAIT_FOR_OLD_PATH = -1
124 1
        evc1 = MagicMock(id=1, service_level=0, creation_time=1)
125 1
        evc1.is_enabled.return_value = True
126 1
        evc1.is_active.return_value = False
127 1
        evc1.lock.locked.return_value = False
128 1
        evc1.has_recent_removed_flow.return_value = False
129 1
        evc1.is_recent_updated.return_value = False
130 1
        evc1.execution_rounds = 0
131 1
        evc2 = MagicMock(id=2, service_level=7, creation_time=1)
132 1
        evc2.is_enabled.return_value = True
133 1
        evc2.is_active.return_value = False
134 1
        evc2.lock.locked.return_value = False
135 1
        evc2.has_recent_removed_flow.return_value = False
136 1
        evc2.is_recent_updated.return_value = False
137 1
        evc2.execution_rounds = 0
138 1
        self.napp.circuits = {'1': evc1, '2': evc2}
139 1
        assert self.napp.get_evcs_by_svc_level() == [evc2, evc1]
140
141 1
        mock_check_list_traces.return_value = {
142
                                                1: True,
143
                                                2: False
144
                                            }
145
146 1
        self.napp.execute_consistency()
147 1
        assert evc1.activate.call_count == 1
148 1
        assert evc1.sync.call_count == 1
149 1
        evc1.try_setup_failover_path.assert_called_with(warn_if_not_path=False)
150 1
        assert evc2.deploy.call_count == 1
151 1
        evc2.try_setup_failover_path.assert_called_with(warn_if_not_path=False)
152
153 1
    @patch('napps.kytos.mef_eline.main.settings')
154 1
    @patch('napps.kytos.mef_eline.main.Main._load_evc')
155 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
156 1
    @patch("napps.kytos.mef_eline.models.evc.EVCDeploy.check_list_traces")
157 1
    def test_execute_consistency_wait_for(self, mock_check_list_traces, *args):
158
        """Test execute and wait for setting."""
159 1
        (mongo_controller_upsert_mock, _, mock_settings) = args
160
161 1
        stored_circuits = {'1': {'name': 'circuit_1'}}
162 1
        mongo_controller_upsert_mock.return_value = True
163 1
        self.napp.mongo_controller.get_circuits.return_value = {
164
            "circuits": stored_circuits
165
        }
166
167 1
        mock_settings.WAIT_FOR_OLD_PATH = -1
168 1
        evc1 = MagicMock(id=1, service_level=0, creation_time=1)
169 1
        evc1.is_enabled.return_value = True
170 1
        evc1.is_active.return_value = False
171 1
        evc1.lock.locked.return_value = False
172 1
        evc1.has_recent_removed_flow.return_value = False
173 1
        evc1.is_recent_updated.return_value = False
174 1
        evc1.execution_rounds = 0
175 1
        evc1.deploy.call_count = 0
176 1
        self.napp.circuits = {'1': evc1}
177 1
        assert self.napp.get_evcs_by_svc_level() == [evc1]
178 1
        mock_settings.WAIT_FOR_OLD_PATH = 1
179
180 1
        mock_check_list_traces.return_value = {1: False}
181
182 1
        self.napp.execute_consistency()
183 1
        assert evc1.deploy.call_count == 0
184 1
        self.napp.execute_consistency()
185 1
        assert evc1.deploy.call_count == 1
186
187 1
    @patch('napps.kytos.mef_eline.main.Main._uni_from_dict')
188 1
    @patch('napps.kytos.mef_eline.models.evc.EVCBase._validate')
189 1
    def test_evc_from_dict(self, _validate_mock, uni_from_dict_mock):
190
        """
191
        Test the helper method that create an EVN from dict.
192
193
        Verify object creation with circuit data and schedule data.
194
        """
195 1
        _validate_mock.return_value = True
196 1
        uni_from_dict_mock.side_effect = ["uni_a", "uni_z"]
197 1
        payload = {
198
            "name": "my evc1",
199
            "uni_a": {
200
                "interface_id": "00:00:00:00:00:00:00:01:1",
201
                "tag": {"tag_type": 'vlan', "value": 80},
202
            },
203
            "uni_z": {
204
                "interface_id": "00:00:00:00:00:00:00:02:2",
205
                "tag": {"tag_type": 'vlan', "value": 1},
206
            },
207
            "circuit_scheduler": [
208
                {"frequency": "* * * * *", "action": "create"}
209
            ],
210
            "queue_id": 5,
211
        }
212
        # pylint: disable=protected-access
213 1
        evc_response = self.napp._evc_from_dict(payload)
214 1
        assert evc_response is not None
215 1
        assert evc_response.uni_a is not None
216 1
        assert evc_response.uni_z is not None
217 1
        assert evc_response.circuit_scheduler is not None
218 1
        assert evc_response.name is not None
219 1
        assert evc_response.queue_id is not None
220
221 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
222 1
    @patch("napps.kytos.mef_eline.models.evc.EVCBase._validate")
223 1
    @patch("kytos.core.Controller.get_interface_by_id")
224 1
    def test_evc_from_dict_paths(
225
        self, _get_interface_by_id_mock, _validate_mock, uni_from_dict_mock
226
    ):
227
        """
228
        Test the helper method that create an EVN from dict.
229
230
        Verify object creation with circuit data and schedule data.
231
        """
232
233 1
        _get_interface_by_id_mock.return_value = get_uni_mocked().interface
234 1
        _validate_mock.return_value = True
235 1
        uni_from_dict_mock.side_effect = ["uni_a", "uni_z"]
236 1
        payload = {
237
            "name": "my evc1",
238
            "uni_a": {
239
                "interface_id": "00:00:00:00:00:00:00:01:1",
240
                "tag": {"tag_type": 'vlan', "value": 80},
241
            },
242
            "uni_z": {
243
                "interface_id": "00:00:00:00:00:00:00:02:2",
244
                "tag": {"tag_type": 'vlan', "value": 1},
245
            },
246
            "current_path": [],
247
            "primary_path": [
248
                {
249
                    "endpoint_a": {
250
                        "interface_id": "00:00:00:00:00:00:00:01:1"
251
                    },
252
                    "endpoint_b": {
253
                        "interface_id": "00:00:00:00:00:00:00:02:2"
254
                    },
255
                }
256
            ],
257
            "backup_path": [],
258
        }
259
260
        # pylint: disable=protected-access
261 1
        evc_response = self.napp._evc_from_dict(payload)
262 1
        assert evc_response is not None
263 1
        assert evc_response.uni_a is not None
264 1
        assert evc_response.uni_z is not None
265 1
        assert evc_response.circuit_scheduler is not None
266 1
        assert evc_response.name is not None
267 1
        assert len(evc_response.current_path) == 0
268 1
        assert len(evc_response.backup_path) == 0
269 1
        assert len(evc_response.primary_path) == 1
270
271 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
272 1
    @patch("napps.kytos.mef_eline.models.evc.EVCBase._validate")
273 1
    @patch("kytos.core.Controller.get_interface_by_id")
274 1
    def test_evc_from_dict_links(
275
        self, _get_interface_by_id_mock, _validate_mock, uni_from_dict_mock
276
    ):
277
        """
278
        Test the helper method that create an EVN from dict.
279
280
        Verify object creation with circuit data and schedule data.
281
        """
282 1
        _get_interface_by_id_mock.return_value = get_uni_mocked().interface
283 1
        _validate_mock.return_value = True
284 1
        uni_from_dict_mock.side_effect = ["uni_a", "uni_z"]
285 1
        payload = {
286
            "name": "my evc1",
287
            "uni_a": {
288
                "interface_id": "00:00:00:00:00:00:00:01:1",
289
                "tag": {"tag_type": 'vlan', "value": 80},
290
            },
291
            "uni_z": {
292
                "interface_id": "00:00:00:00:00:00:00:02:2",
293
                "tag": {"tag_type": 'vlan', "value": 1},
294
            },
295
            "primary_links": [
296
                {
297
                    "endpoint_a": {
298
                        "interface_id": "00:00:00:00:00:00:00:01:1"
299
                    },
300
                    "endpoint_b": {
301
                        "interface_id": "00:00:00:00:00:00:00:02:2"
302
                    },
303
                    "metadata": {
304
                        "s_vlan": {
305
                            "tag_type": 'vlan',
306
                            "value": 100
307
                        }
308
                    },
309
                }
310
            ],
311
            "backup_links": [],
312
        }
313
314
        # pylint: disable=protected-access
315 1
        evc_response = self.napp._evc_from_dict(payload)
316 1
        assert evc_response is not None
317 1
        assert evc_response.uni_a is not None
318 1
        assert evc_response.uni_z is not None
319 1
        assert evc_response.circuit_scheduler is not None
320 1
        assert evc_response.name is not None
321 1
        assert len(evc_response.current_links_cache) == 0
322 1
        assert len(evc_response.backup_links) == 0
323 1
        assert len(evc_response.primary_links) == 1
324
325 1
    async def test_list_without_circuits(self):
326
        """Test if list circuits return 'no circuit stored.'."""
327 1
        circuits = {"circuits": {}}
328 1
        self.napp.mongo_controller.get_circuits.return_value = circuits
329 1
        url = f"{self.base_endpoint}/v2/evc/"
330 1
        response = await self.api_client.get(url)
331 1
        assert response.status_code == 200, response.data
332 1
        assert not response.json()
333
334 1
    async def test_list_no_circuits_stored(self):
335
        """Test if list circuits return all circuits stored."""
336 1
        circuits = {"circuits": {}}
337 1
        self.napp.mongo_controller.get_circuits.return_value = circuits
338
339 1
        url = f"{self.base_endpoint}/v2/evc/"
340 1
        response = await self.api_client.get(url)
341 1
        expected_result = circuits["circuits"]
342 1
        assert response.json() == expected_result
343
344 1 View Code Duplication
    async def test_list_with_circuits_stored(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
345
        """Test if list circuits return all circuits stored."""
346 1
        circuits = {
347
            'circuits':
348
            {"1": {"name": "circuit_1"}, "2": {"name": "circuit_2"}}
349
        }
350 1
        get_circuits = self.napp.mongo_controller.get_circuits
351 1
        get_circuits.return_value = circuits
352
353 1
        url = f"{self.base_endpoint}/v2/evc/"
354 1
        response = await self.api_client.get(url)
355 1
        expected_result = circuits["circuits"]
356 1
        get_circuits.assert_called_with(archived="false", metadata={})
357 1
        assert response.json() == expected_result
358
359 1 View Code Duplication
    async def test_list_with_archived_circuits_archived(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
360
        """Test if list circuits only archived circuits."""
361 1
        circuits = {
362
            'circuits':
363
            {
364
                "1": {"name": "circuit_1", "archived": True},
365
            }
366
        }
367 1
        get_circuits = self.napp.mongo_controller.get_circuits
368 1
        get_circuits.return_value = circuits
369
370 1
        url = f"{self.base_endpoint}/v2/evc/?archived=true&metadata.a=1"
371 1
        response = await self.api_client.get(url)
372 1
        get_circuits.assert_called_with(archived="true",
373
                                        metadata={"metadata.a": "1"})
374 1
        expected_result = {"1": circuits["circuits"]["1"]}
375 1
        assert response.json() == expected_result
376
377 1
    async def test_list_with_archived_circuits_all(self):
378
        """Test if list circuits return all circuits."""
379 1
        circuits = {
380
            'circuits': {
381
                "1": {"name": "circuit_1"},
382
                "2": {"name": "circuit_2", "archived": True},
383
            }
384
        }
385 1
        self.napp.mongo_controller.get_circuits.return_value = circuits
386
387 1
        url = f"{self.base_endpoint}/v2/evc/?archived=null"
388 1
        response = await self.api_client.get(url)
389 1
        expected_result = circuits["circuits"]
390 1
        assert response.json() == expected_result
391
392 1
    async def test_circuit_with_valid_id(self):
393
        """Test if get_circuit return the circuit attributes."""
394 1
        circuit = {"name": "circuit_1"}
395 1
        self.napp.mongo_controller.get_circuit.return_value = circuit
396
397 1
        url = f"{self.base_endpoint}/v2/evc/1"
398 1
        response = await self.api_client.get(url)
399 1
        expected_result = circuit
400 1
        assert response.json() == expected_result
401
402 1
    async def test_circuit_with_invalid_id(self):
403
        """Test if get_circuit return invalid circuit_id."""
404 1
        self.napp.mongo_controller.get_circuit.return_value = None
405 1
        url = f"{self.base_endpoint}/v2/evc/3"
406 1
        response = await self.api_client.get(url)
407
        expected_result = "circuit_id 3 not found"
408
        assert response.json()["description"] == expected_result
409
410 1
    @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
411 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
412 1
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
413 1
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
414 1
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
415 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
416 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
417 1
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
418 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
419 1
    async def test_create_a_circuit_case_1(
420
        self,
421
        validate_mock,
422
        evc_as_dict_mock,
423
        mongo_controller_upsert_mock,
424
        uni_from_dict_mock,
425
        sched_add_mock,
426
        evc_deploy_mock,
427
        mock_use_uni_tags,
428
        mock_tags_equal,
429
        mock_check_duplicate
430
    ):
431
        """Test create a new circuit."""
432
        # pylint: disable=too-many-locals
433 1
        self.napp.controller.loop = asyncio.get_running_loop()
434 1
        validate_mock.return_value = True
435 1
        mongo_controller_upsert_mock.return_value = True
436 1
        evc_deploy_mock.return_value = True
437 1
        mock_use_uni_tags.return_value = True
438 1
        mock_tags_equal.return_value = True
439 1
        mock_check_duplicate.return_value = True
440 1
        uni1 = create_autospec(UNI)
441 1
        uni2 = create_autospec(UNI)
442 1
        uni1.interface = create_autospec(Interface)
443 1
        uni2.interface = create_autospec(Interface)
444 1
        uni1.interface.switch = "00:00:00:00:00:00:00:01"
445 1
        uni2.interface.switch = "00:00:00:00:00:00:00:02"
446 1
        uni_from_dict_mock.side_effect = [uni1, uni2]
447 1
        evc_as_dict_mock.return_value = {}
448 1
        sched_add_mock.return_value = True
449 1
        self.napp.mongo_controller.get_circuits.return_value = {}
450
451 1
        url = f"{self.base_endpoint}/v2/evc/"
452 1
        payload = {
453
            "name": "my evc1",
454
            "frequency": "* * * * *",
455
            "uni_a": {
456
                "interface_id": "00:00:00:00:00:00:00:01:1",
457
                "tag": {"tag_type": 'vlan', "value": 80},
458
            },
459
            "uni_z": {
460
                "interface_id": "00:00:00:00:00:00:00:02:2",
461
                "tag": {"tag_type": 'vlan', "value": 1},
462
            },
463
            "dynamic_backup_path": True,
464
            "primary_constraints": {
465
                "spf_max_path_cost": 8,
466
                "mandatory_metrics": {
467
                    "ownership": "red"
468
                }
469
            },
470
            "secondary_constraints": {
471
                "spf_attribute": "priority",
472
                "mandatory_metrics": {
473
                    "ownership": "blue"
474
                }
475
            }
476
        }
477
478 1
        response = await self.api_client.post(url, json=payload)
479 1
        current_data = response.json()
480
481
        # verify expected result from request
482 1
        assert 201 == response.status_code
483 1
        assert "circuit_id" in current_data
484
485
        # verify uni called
486 1
        uni_from_dict_mock.called_twice()
487 1
        uni_from_dict_mock.assert_any_call(payload["uni_z"])
488 1
        uni_from_dict_mock.assert_any_call(payload["uni_a"])
489
490
        # verify validation called
491 1
        validate_mock.assert_called_once()
492 1
        validate_mock.assert_called_with(
493
            table_group={'evpl': 0, 'epl': 0},
494
            frequency="* * * * *",
495
            name="my evc1",
496
            uni_a=uni1,
497
            uni_z=uni2,
498
            dynamic_backup_path=True,
499
            primary_constraints=payload["primary_constraints"],
500
            secondary_constraints=payload["secondary_constraints"],
501
        )
502
        # verify save method is called
503 1
        mongo_controller_upsert_mock.assert_called_once()
504
505
        # verify evc as dict is called to save in the box
506 1
        evc_as_dict_mock.assert_called()
507
        # verify add circuit in sched
508 1
        sched_add_mock.assert_called_once()
509
510 1
    async def test_create_a_circuit_case_2(self):
511
        """Test create a new circuit trying to send request without a json."""
512 1
        self.napp.controller.loop = asyncio.get_running_loop()
513 1
        url = f"{self.base_endpoint}/v2/evc/"
514
515 1
        response = await self.api_client.post(url)
516
        current_data = response.json()
517
        assert 400 == response.status_code
518
        assert "Missing required request body" in current_data["description"]
519
520 1
    async def test_create_a_circuit_case_3(self):
521
        """Test create a new circuit trying to send request with an
522
        invalid json."""
523 1
        self.napp.controller.loop = asyncio.get_running_loop()
524 1
        url = f"{self.base_endpoint}/v2/evc/"
525
526 1
        response = await self.api_client.post(
527
            url,
528
            json="This is an {Invalid:} JSON",
529
        )
530
        current_data = response.json()
531
        assert 400 == response.status_code
532
        assert "This is an {Invalid:" in current_data["description"]
533
534 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
535 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
536 1
    async def test_create_a_circuit_case_4(
537
        self,
538
        mongo_controller_upsert_mock,
539
        uni_from_dict_mock
540
    ):
541
        """Test create a new circuit trying to send request with an
542
        invalid value."""
543 1
        self.napp.controller.loop = asyncio.get_running_loop()
544
        # pylint: disable=too-many-locals
545 1
        uni_from_dict_mock.side_effect = ValueError("Could not instantiate")
546 1
        mongo_controller_upsert_mock.return_value = True
547 1
        url = f"{self.base_endpoint}/v2/evc/"
548
549 1
        payload = {
550
            "name": "my evc1",
551
            "frequency": "* * * * *",
552
            "uni_a": {
553
                "interface_id": "00:00:00:00:00:00:00:01:76",
554
                "tag": {"tag_type": 'vlan', "value": 80},
555
            },
556
            "uni_z": {
557
                "interface_id": "00:00:00:00:00:00:00:02:2",
558
                "tag": {"tag_type": 'vlan', "value": 1},
559
            },
560
        }
561
562 1
        response = await self.api_client.post(url, json=payload)
563
        current_data = response.json()
564
        expected_data = "Error creating UNI: Invalid value"
565
        assert 400 == response.status_code
566
        assert current_data["description"] == expected_data
567
568
        payload["name"] = 1
569
        response = await self.api_client.post(url, json=payload)
570
        assert 400 == response.status_code, response.data
571
572 1 View Code Duplication
    @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
573 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
574 1
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
575 1
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
576 1
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
577 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
578 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
579 1
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
580 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
581 1
    async def test_create_a_circuit_case_5(
582
        self,
583
        validate_mock,
584
        evc_as_dict_mock,
585
        mongo_controller_upsert_mock,
586
        uni_from_dict_mock,
587
        sched_add_mock,
588
        evc_deploy_mock,
589
        mock_use_uni_tags,
590
        mock_tags_equal,
591
        mock_check_duplicate
592
    ):
593
        """Test create a new intra circuit with a disabled switch"""
594
        # pylint: disable=too-many-locals
595 1
        self.napp.controller.loop = asyncio.get_running_loop()
596 1
        validate_mock.return_value = True
597 1
        mongo_controller_upsert_mock.return_value = True
598 1
        evc_deploy_mock.return_value = True
599 1
        mock_use_uni_tags.return_value = True
600 1
        mock_tags_equal.return_value = True
601 1
        mock_check_duplicate.return_value = True
602 1
        uni1 = create_autospec(UNI)
603 1
        uni2 = create_autospec(UNI)
604 1
        uni1.interface = create_autospec(Interface)
605 1
        uni2.interface = create_autospec(Interface)
606 1
        switch = MagicMock()
607 1
        switch.status = EntityStatus.DISABLED
608 1
        switch.return_value = "00:00:00:00:00:00:00:01"
609 1
        uni1.interface.switch = switch
610 1
        uni2.interface.switch = switch
611 1
        uni_from_dict_mock.side_effect = [uni1, uni2]
612 1
        evc_as_dict_mock.return_value = {}
613 1
        sched_add_mock.return_value = True
614 1
        self.napp.mongo_controller.get_circuits.return_value = {}
615
616 1
        url = f"{self.base_endpoint}/v2/evc/"
617 1
        payload = {
618
            "name": "my evc1",
619
            "frequency": "* * * * *",
620
            "uni_a": {
621
                "interface_id": "00:00:00:00:00:00:00:01:1",
622
                "tag": {"tag_type": 'vlan', "value": 80},
623
            },
624
            "uni_z": {
625
                "interface_id": "00:00:00:00:00:00:00:01:2",
626
                "tag": {"tag_type": 'vlan', "value": 1},
627
            },
628
            "dynamic_backup_path": True,
629
            "primary_constraints": {
630
                "spf_max_path_cost": 8,
631
                "mandatory_metrics": {
632
                    "ownership": "red"
633
                }
634
            },
635
            "secondary_constraints": {
636
                "spf_attribute": "priority",
637
                "mandatory_metrics": {
638
                    "ownership": "blue"
639
                }
640
            }
641
        }
642
643 1
        response = await self.api_client.post(url, json=payload)
644 1
        current_data = response.json()
645
646
        # verify expected result from request
647 1
        assert 201 == response.status_code
648 1
        assert "circuit_id" in current_data
649
650
        # verify uni called
651 1
        uni_from_dict_mock.called_twice()
652 1
        uni_from_dict_mock.assert_any_call(payload["uni_z"])
653 1
        uni_from_dict_mock.assert_any_call(payload["uni_a"])
654
655
        # verify validation called
656 1
        validate_mock.assert_called_once()
657 1
        validate_mock.assert_called_with(
658
            table_group={'evpl': 0, 'epl': 0},
659
            frequency="* * * * *",
660
            name="my evc1",
661
            uni_a=uni1,
662
            uni_z=uni2,
663
            dynamic_backup_path=True,
664
            primary_constraints=payload["primary_constraints"],
665
            secondary_constraints=payload["secondary_constraints"],
666
        )
667
        # verify save method is called
668 1
        mongo_controller_upsert_mock.assert_called_once()
669
670
        # verify evc as dict is called to save in the box
671 1
        evc_as_dict_mock.assert_called()
672
        # verify add circuit in sched
673 1
        sched_add_mock.assert_called_once()
674
675 1 View Code Duplication
    @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
676 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
677 1
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
678 1
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
679 1
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
680 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
681 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
682 1
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
683 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
684 1
    async def test_create_a_circuit_case_6(
685
        self,
686
        validate_mock,
687
        evc_as_dict_mock,
688
        mongo_controller_upsert_mock,
689
        uni_from_dict_mock,
690
        sched_add_mock,
691
        evc_deploy_mock,
692
        mock_use_uni_tags,
693
        mock_tags_equal,
694
        mock_check_duplicate
695
    ):
696
        """Test create a new intra circuit with a disabled interface"""
697
        # pylint: disable=too-many-locals
698 1
        self.napp.controller.loop = asyncio.get_running_loop()
699 1
        validate_mock.return_value = True
700 1
        mongo_controller_upsert_mock.return_value = True
701 1
        evc_deploy_mock.return_value = True
702 1
        mock_use_uni_tags.return_value = True
703 1
        mock_tags_equal.return_value = True
704 1
        mock_check_duplicate.return_value = True
705 1
        uni1 = create_autospec(UNI)
706 1
        uni2 = create_autospec(UNI)
707 1
        uni1.interface = create_autospec(Interface)
708 1
        uni2.interface = create_autospec(Interface)
709 1
        switch = MagicMock()
710 1
        switch.status = EntityStatus.UP
711 1
        switch.return_value = "00:00:00:00:00:00:00:01"
712 1
        uni1.interface.switch = switch
713 1
        uni1.interface.status = EntityStatus.DISABLED
714 1
        uni2.interface.switch = switch
715 1
        uni_from_dict_mock.side_effect = [uni1, uni2]
716 1
        evc_as_dict_mock.return_value = {}
717 1
        sched_add_mock.return_value = True
718 1
        self.napp.mongo_controller.get_circuits.return_value = {}
719
720 1
        url = f"{self.base_endpoint}/v2/evc/"
721 1
        payload = {
722
            "name": "my evc1",
723
            "frequency": "* * * * *",
724
            "uni_a": {
725
                "interface_id": "00:00:00:00:00:00:00:01:1",
726
                "tag": {"tag_type": 'vlan', "value": 80},
727
            },
728
            "uni_z": {
729
                "interface_id": "00:00:00:00:00:00:00:01:2",
730
                "tag": {"tag_type": 'vlan', "value": 1},
731
            },
732
            "dynamic_backup_path": True,
733
            "primary_constraints": {
734
                "spf_max_path_cost": 8,
735
                "mandatory_metrics": {
736
                    "ownership": "red"
737
                }
738
            },
739
            "secondary_constraints": {
740
                "spf_attribute": "priority",
741
                "mandatory_metrics": {
742
                    "ownership": "blue"
743
                }
744
            }
745
        }
746
747 1
        response = await self.api_client.post(url, json=payload)
748 1
        current_data = response.json()
749
750
        # verify expected result from request
751 1
        assert 201 == response.status_code
752 1
        assert "circuit_id" in current_data
753
754
        # verify uni called
755 1
        uni_from_dict_mock.called_twice()
756 1
        uni_from_dict_mock.assert_any_call(payload["uni_z"])
757 1
        uni_from_dict_mock.assert_any_call(payload["uni_a"])
758
759
        # verify validation called
760 1
        validate_mock.assert_called_once()
761 1
        validate_mock.assert_called_with(
762
            table_group={'evpl': 0, 'epl': 0},
763
            frequency="* * * * *",
764
            name="my evc1",
765
            uni_a=uni1,
766
            uni_z=uni2,
767
            dynamic_backup_path=True,
768
            primary_constraints=payload["primary_constraints"],
769
            secondary_constraints=payload["secondary_constraints"],
770
        )
771
        # verify save method is called
772 1
        mongo_controller_upsert_mock.assert_called_once()
773
774
        # verify evc as dict is called to save in the box
775 1
        evc_as_dict_mock.assert_called()
776
        # verify add circuit in sched
777 1
        sched_add_mock.assert_called_once()
778
779 1
    async def test_create_a_circuit_invalid_queue_id(self):
780
        """Test create a new circuit with invalid queue_id."""
781 1
        self.napp.controller.loop = asyncio.get_running_loop()
782 1
        url = f"{self.base_endpoint}/v2/evc/"
783
784 1
        payload = {
785
            "name": "my evc1",
786
            "queue_id": 8,
787
            "uni_a": {
788
                "interface_id": "00:00:00:00:00:00:00:01:76",
789
                "tag": {"tag_type": 'vlan', "value": 80},
790
            },
791
            "uni_z": {
792
                "interface_id": "00:00:00:00:00:00:00:02:2",
793
                "tag": {"tag_type": 'vlan', "value": 1},
794
            },
795
        }
796 1
        response = await self.api_client.post(url, json=payload)
797
        current_data = response.json()
798
        expected_data = "8 is greater than the maximum of 7"
799
800
        assert response.status_code == 400
801
        assert expected_data in current_data["description"]
802
803 1
    @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
804 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
805 1
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
806 1
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
807 1
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
808 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
809 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
810 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
811 1
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
812 1
    async def test_create_circuit_already_enabled(
813
        self,
814
        evc_as_dict_mock,
815
        validate_mock,
816
        mongo_controller_upsert_mock,
817
        uni_from_dict_mock,
818
        sched_add_mock,
819
        evc_deploy_mock,
820
        mock_use_uni_tags,
821
        mock_tags_equal,
822
        mock_check_duplicate
823
    ):
824
        """Test create an already created circuit."""
825
        # pylint: disable=too-many-locals
826 1
        self.napp.controller.loop = asyncio.get_running_loop()
827 1
        validate_mock.return_value = True
828 1
        mongo_controller_upsert_mock.return_value = True
829 1
        sched_add_mock.return_value = True
830 1
        evc_deploy_mock.return_value = True
831 1
        mock_tags_equal.return_value = True
832 1
        mock_check_duplicate.return_value = True
833 1
        mock_use_uni_tags.side_effect = [
834
            None, KytosTagError("The EVC already exists.")
835
        ]
836 1
        uni1 = create_autospec(UNI)
837 1
        uni2 = create_autospec(UNI)
838 1
        uni1.interface = create_autospec(Interface)
839 1
        uni2.interface = create_autospec(Interface)
840 1
        uni1.interface.switch = "00:00:00:00:00:00:00:01"
841 1
        uni2.interface.switch = "00:00:00:00:00:00:00:02"
842 1
        uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2]
843
844 1
        payload = {
845
            "name": "my evc1",
846
            "uni_a": {
847
                "interface_id": "00:00:00:00:00:00:00:01:1",
848
                "tag": {"tag_type": 'vlan', "value": 80},
849
            },
850
            "uni_z": {
851
                "interface_id": "00:00:00:00:00:00:00:02:2",
852
                "tag": {"tag_type": 'vlan', "value": 1},
853
            },
854
            "dynamic_backup_path": True,
855
        }
856
857 1
        evc_as_dict_mock.return_value = payload
858 1
        response = await self.api_client.post(
859
            f"{self.base_endpoint}/v2/evc/",
860
            json=payload
861
        )
862 1
        assert 201 == response.status_code
863
864 1
        response = await self.api_client.post(
865
            f"{self.base_endpoint}/v2/evc/",
866
            json=payload
867
        )
868
        current_data = response.json()
869
        expected_data = "The EVC already exists."
870
        assert current_data["description"] == expected_data
871
        assert 400 == response.status_code
872
873 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
874 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
875 1
    async def test_create_circuit_case_5(
876
        self,
877
        uni_from_dict_mock,
878
        mock_tags_equal
879
    ):
880
        """Test when neither primary path nor dynamic_backup_path is set."""
881 1
        self.napp.controller.loop = asyncio.get_running_loop()
882 1
        mock_tags_equal.return_value = True
883 1
        url = f"{self.base_endpoint}/v2/evc/"
884 1
        uni1 = create_autospec(UNI)
885 1
        uni2 = create_autospec(UNI)
886 1
        uni1.interface = create_autospec(Interface)
887 1
        uni2.interface = create_autospec(Interface)
888 1
        uni1.interface.switch = "00:00:00:00:00:00:00:01"
889 1
        uni2.interface.switch = "00:00:00:00:00:00:00:02"
890 1
        uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2]
891
892 1
        payload = {
893
            "name": "my evc1",
894
            "frequency": "* * * * *",
895
            "uni_a": {
896
                "interface_id": "00:00:00:00:00:00:00:01:1",
897
                "tag": {"tag_type": 'vlan', "value": 80},
898
            },
899
            "uni_z": {
900
                "interface_id": "00:00:00:00:00:00:00:02:2",
901
                "tag": {"tag_type": 'vlan', "value": 1},
902
            },
903
        }
904
905 1
        response = await self.api_client.post(url, json=payload)
906
        current_data = response.json()
907
        expected_data = "The EVC must have a primary path "
908
        expected_data += "or allow dynamic paths."
909
        assert 400 == response.status_code, response.data
910
        assert current_data["description"] == expected_data
911
912 1
    @patch("napps.kytos.mef_eline.main.Main._evc_from_dict")
913 1
    async def test_create_circuit_case_6(self, mock_evc):
914
        """Test create_circuit with KytosTagError"""
915 1
        self.napp.controller.loop = asyncio.get_running_loop()
916 1
        url = f"{self.base_endpoint}/v2/evc/"
917 1
        mock_evc.side_effect = KytosTagError("")
918 1
        payload = {
919
            "name": "my evc1",
920
            "uni_a": {
921
                "interface_id": "00:00:00:00:00:00:00:01:1",
922
            },
923
            "uni_z": {
924
                "interface_id": "00:00:00:00:00:00:00:02:2",
925
            },
926
        }
927 1
        response = await self.api_client.post(url, json=payload)
928
        assert response.status_code == 400, response.data
929
930 1
    async def test_redeploy_evc(self):
931
        """Test endpoint to redeploy an EVC."""
932 1
        evc1 = MagicMock()
933 1
        evc1.is_enabled.return_value = True
934 1
        self.napp.circuits = {"1": evc1, "2": MagicMock()}
935 1
        url = f"{self.base_endpoint}/v2/evc/1/redeploy"
936 1
        response = await self.api_client.patch(url)
937 1
        evc1.remove_failover_flows.assert_called()
938 1
        evc1.remove_current_flows.assert_called_with(
939
            sync=False, return_path=True
940
        )
941 1
        assert response.status_code == 202, response.data
942
943 1
        url = f"{self.base_endpoint}/v2/evc/1/redeploy"
944 1
        url = url + "?try_avoid_same_s_vlan=false"
945 1
        response = await self.api_client.patch(url)
946 1
        evc1.remove_current_flows.assert_called_with(
947
            sync=False, return_path=False
948
        )
949
950 1
        url = f"{self.base_endpoint}/v2/evc/1/redeploy"
951 1
        url = url + "?try_avoid_same_s_vlan=True"
952 1
        response = await self.api_client.patch(url)
953 1
        evc1.remove_current_flows.assert_called_with(
954
            sync=False, return_path=True
955
        )
956
957 1
    async def test_redeploy_evc_disabled(self):
958
        """Test endpoint to redeploy an EVC."""
959 1
        evc1 = MagicMock()
960 1
        evc1.is_enabled.return_value = False
961 1
        self.napp.circuits = {"1": evc1, "2": MagicMock()}
962 1
        url = f"{self.base_endpoint}/v2/evc/1/redeploy"
963 1
        response = await self.api_client.patch(url)
964 1
        evc1.remove_failover_flows.assert_not_called()
965 1
        evc1.remove_current_flows.assert_not_called()
966 1
        assert response.status_code == 409, response.data
967
968 1
    async def test_redeploy_evc_deleted(self):
969
        """Test endpoint to redeploy an EVC."""
970 1
        evc1 = MagicMock()
971 1
        evc1.is_enabled.return_value = True
972 1
        self.napp.circuits = {"1": evc1, "2": MagicMock()}
973 1
        url = f"{self.base_endpoint}/v2/evc/3/redeploy"
974 1
        response = await self.api_client.patch(url)
975
        assert response.status_code == 404, response.data
976
977 1
    async def test_list_schedules__no_data_stored(self):
978
        """Test if list circuits return all circuits stored."""
979 1
        self.napp.mongo_controller.get_circuits.return_value = {"circuits": {}}
980
981 1
        url = f"{self.base_endpoint}/v2/evc/schedule"
982
983 1
        response = await self.api_client.get(url)
984 1
        assert response.status_code == 200
985 1
        assert not response.json()
986
987 1
    def _add_mongodb_schedule_data(self, data_mock):
988
        """Add schedule data to mongodb mock object."""
989 1
        circuits = {"circuits": {}}
990 1
        payload_1 = {
991
            "id": "aa:aa:aa",
992
            "name": "my evc1",
993
            "uni_a": {
994
                "interface_id": "00:00:00:00:00:00:00:01:1",
995
                "tag": {"tag_type": 'vlan', "value": 80},
996
            },
997
            "uni_z": {
998
                "interface_id": "00:00:00:00:00:00:00:02:2",
999
                "tag": {"tag_type": 'vlan', "value": 1},
1000
            },
1001
            "circuit_scheduler": [
1002
                {"id": "1", "frequency": "* * * * *", "action": "create"},
1003
                {"id": "2", "frequency": "1 * * * *", "action": "remove"},
1004
            ],
1005
        }
1006 1
        circuits["circuits"].update({"aa:aa:aa": payload_1})
1007 1
        payload_2 = {
1008
            "id": "bb:bb:bb",
1009
            "name": "my second evc2",
1010
            "uni_a": {
1011
                "interface_id": "00:00:00:00:00:00:00:01:2",
1012
                "tag": {"tag_type": 'vlan', "value": 90},
1013
            },
1014
            "uni_z": {
1015
                "interface_id": "00:00:00:00:00:00:00:03:2",
1016
                "tag": {"tag_type": 'vlan', "value": 100},
1017
            },
1018
            "circuit_scheduler": [
1019
                {"id": "3", "frequency": "1 * * * *", "action": "create"},
1020
                {"id": "4", "frequency": "2 * * * *", "action": "remove"},
1021
            ],
1022
        }
1023 1
        circuits["circuits"].update({"bb:bb:bb": payload_2})
1024 1
        payload_3 = {
1025
            "id": "cc:cc:cc",
1026
            "name": "my third evc3",
1027
            "uni_a": {
1028
                "interface_id": "00:00:00:00:00:00:00:03:1",
1029
                "tag": {"tag_type": 'vlan', "value": 90},
1030
            },
1031
            "uni_z": {
1032
                "interface_id": "00:00:00:00:00:00:00:04:2",
1033
                "tag": {"tag_type": 'vlan', "value": 100},
1034
            },
1035
        }
1036 1
        circuits["circuits"].update({"cc:cc:cc": payload_3})
1037
        # Add one circuit to the mongodb.
1038 1
        data_mock.return_value = circuits
1039
1040 1
    async def test_list_schedules_from_mongodb(self):
1041
        """Test if list circuits return specific circuits stored."""
1042 1
        self._add_mongodb_schedule_data(
1043
            self.napp.mongo_controller.get_circuits
1044
        )
1045
1046 1
        url = f"{self.base_endpoint}/v2/evc/schedule"
1047
1048
        # Call URL
1049 1
        response = await self.api_client.get(url)
1050
        # Expected JSON data from response
1051 1
        expected = [
1052
            {
1053
                "circuit_id": "aa:aa:aa",
1054
                "schedule": {
1055
                    "action": "create",
1056
                    "frequency": "* * * * *",
1057
                    "id": "1",
1058
                },
1059
                "schedule_id": "1",
1060
            },
1061
            {
1062
                "circuit_id": "aa:aa:aa",
1063
                "schedule": {
1064
                    "action": "remove",
1065
                    "frequency": "1 * * * *",
1066
                    "id": "2",
1067
                },
1068
                "schedule_id": "2",
1069
            },
1070
            {
1071
                "circuit_id": "bb:bb:bb",
1072
                "schedule": {
1073
                    "action": "create",
1074
                    "frequency": "1 * * * *",
1075
                    "id": "3",
1076
                },
1077
                "schedule_id": "3",
1078
            },
1079
            {
1080
                "circuit_id": "bb:bb:bb",
1081
                "schedule": {
1082
                    "action": "remove",
1083
                    "frequency": "2 * * * *",
1084
                    "id": "4",
1085
                },
1086
                "schedule_id": "4",
1087
            },
1088
        ]
1089
1090 1
        assert response.status_code == 200
1091 1
        assert expected == response.json()
1092
1093 1
    async def test_get_specific_schedule_from_mongodb(self):
1094
        """Test get schedules from a circuit."""
1095 1
        self._add_mongodb_schedule_data(
1096
            self.napp.mongo_controller.get_circuits
1097
        )
1098
1099 1
        requested_circuit_id = "bb:bb:bb"
1100 1
        evc = self.napp.mongo_controller.get_circuits()
1101 1
        evc = evc["circuits"][requested_circuit_id]
1102 1
        self.napp.mongo_controller.get_circuit.return_value = evc
1103 1
        url = f"{self.base_endpoint}/v2/evc/{requested_circuit_id}"
1104
1105
        # Call URL
1106 1
        response = await self.api_client.get(url)
1107
1108
        # Expected JSON data from response
1109 1
        expected = [
1110
            {"action": "create", "frequency": "1 * * * *", "id": "3"},
1111
            {"action": "remove", "frequency": "2 * * * *", "id": "4"},
1112
        ]
1113
1114 1
        assert response.status_code == 200
1115 1
        assert expected == response.json()["circuit_scheduler"]
1116
1117 1
    async def test_get_specific_schedules_from_mongodb_not_found(self):
1118
        """Test get specific schedule ID that does not exist."""
1119 1
        requested_id = "blah"
1120 1
        self.napp.mongo_controller.get_circuit.return_value = None
1121 1
        url = f"{self.base_endpoint}/v2/evc/{requested_id}"
1122
1123
        # Call URL
1124 1
        response = await self.api_client.get(url)
1125
1126
        expected = "circuit_id blah not found"
1127
        # Assert response not found
1128
        assert response.status_code == 404
1129
        assert expected == response.json()["description"]
1130
1131 1
    def _uni_from_dict_side_effect(self, uni_dict):
1132 1
        interface_id = uni_dict.get("interface_id")
1133 1
        tag_dict = uni_dict.get("tag")
1134 1
        interface = Interface(interface_id, "0", MagicMock(id="1"))
1135 1
        return UNI(interface, tag_dict)
1136
1137 1
    @patch("apscheduler.schedulers.background.BackgroundScheduler.add_job")
1138 1
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
1139 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
1140 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
1141 1
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
1142 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
1143 1
    async def test_create_schedule(
1144
        self,
1145
        validate_mock,
1146
        evc_as_dict_mock,
1147
        mongo_controller_upsert_mock,
1148
        uni_from_dict_mock,
1149
        sched_add_mock,
1150
        scheduler_add_job_mock
1151
    ):
1152
        """Test create a circuit schedule."""
1153 1
        self.napp.controller.loop = asyncio.get_running_loop()
1154 1
        validate_mock.return_value = True
1155 1
        mongo_controller_upsert_mock.return_value = True
1156 1
        uni_from_dict_mock.side_effect = self._uni_from_dict_side_effect
1157 1
        evc_as_dict_mock.return_value = {}
1158 1
        sched_add_mock.return_value = True
1159
1160 1
        self._add_mongodb_schedule_data(
1161
            self.napp.mongo_controller.get_circuits
1162
        )
1163
1164 1
        requested_id = "bb:bb:bb"
1165 1
        url = f"{self.base_endpoint}/v2/evc/schedule/"
1166
1167 1
        payload = {
1168
            "circuit_id": requested_id,
1169
            "schedule": {"frequency": "1 * * * *", "action": "create"},
1170
            "metadata": {"metadata1": "test_data"},
1171
        }
1172
1173
        # Call URL
1174 1
        response = await self.api_client.post(url, json=payload)
1175 1
        response_json = response.json()
1176
1177 1
        assert response.status_code == 201
1178 1
        scheduler_add_job_mock.assert_called_once()
1179 1
        mongo_controller_upsert_mock.assert_called_once()
1180 1
        assert payload["schedule"]["frequency"] == response_json["frequency"]
1181 1
        assert payload["schedule"]["action"] == response_json["action"]
1182 1
        assert response_json["id"] is not None
1183
1184
        # Case 2: there is no schedule
1185 1
        payload = {
1186
              "circuit_id": "cc:cc:cc",
1187
              "schedule": {
1188
                "frequency": "1 * * * *",
1189
                "action": "create"
1190
              }
1191
            }
1192 1
        response = await self.api_client.post(url, json=payload)
1193 1
        assert response.status_code == 201
1194
1195 1
    async def test_create_schedule_invalid_request(self):
1196
        """Test create schedule API with invalid request."""
1197 1
        self.napp.controller.loop = asyncio.get_running_loop()
1198 1
        evc1 = MagicMock()
1199 1
        self.napp.circuits = {'bb:bb:bb': evc1}
1200 1
        url = f'{self.base_endpoint}/v2/evc/schedule/'
1201
1202
        # case 1: empty post
1203 1
        response = await self.api_client.post(url, json={})
1204
        assert response.status_code == 400
1205
1206
        # case 2: not a dictionary
1207
        payload = []
1208
        response = await self.api_client.post(url, json=payload)
1209
        assert response.status_code == 400
1210
1211
        # case 3: missing circuit id
1212
        payload = {
1213
            "schedule": {
1214
                "frequency": "1 * * * *",
1215
                "action": "create"
1216
            }
1217
        }
1218
        response = await self.api_client.post(url, json=payload)
1219
        assert response.status_code == 400
1220
1221
        # case 4: missing schedule
1222
        payload = {
1223
            "circuit_id": "bb:bb:bb"
1224
        }
1225
        response = await self.api_client.post(url, json=payload)
1226
        assert response.status_code == 400
1227
1228
        # case 5: invalid circuit
1229
        payload = {
1230
            "circuit_id": "xx:xx:xx",
1231
            "schedule": {
1232
                "frequency": "1 * * * *",
1233
                "action": "create"
1234
            }
1235
        }
1236
        response = await self.api_client.post(url, json=payload)
1237
        assert response.status_code == 404
1238
1239
        # case 6: invalid json
1240
        response = await self.api_client.post(url, json="test")
1241
        assert response.status_code == 400
1242
1243 1
    @patch('apscheduler.schedulers.background.BackgroundScheduler.remove_job')
1244 1
    @patch('napps.kytos.mef_eline.scheduler.Scheduler.add')
1245 1
    @patch('napps.kytos.mef_eline.main.Main._uni_from_dict')
1246 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
1247 1
    @patch('napps.kytos.mef_eline.main.EVC.as_dict')
1248 1
    @patch('napps.kytos.mef_eline.models.evc.EVC._validate')
1249 1
    async def test_update_schedule(
1250
        self,
1251
        validate_mock,
1252
        evc_as_dict_mock,
1253
        mongo_controller_upsert_mock,
1254
        uni_from_dict_mock,
1255
        sched_add_mock,
1256
        scheduler_remove_job_mock
1257
    ):
1258
        """Test create a circuit schedule."""
1259 1
        self.napp.controller.loop = asyncio.get_running_loop()
1260 1
        mongo_payload_1 = {
1261
            "circuits": {
1262
                "aa:aa:aa": {
1263
                    "id": "aa:aa:aa",
1264
                    "name": "my evc1",
1265
                    "uni_a": {
1266
                        "interface_id": "00:00:00:00:00:00:00:01:1",
1267
                        "tag": {"tag_type": 'vlan', "value": 80},
1268
                    },
1269
                    "uni_z": {
1270
                        "interface_id": "00:00:00:00:00:00:00:02:2",
1271
                        "tag": {"tag_type": 'vlan', "value": 1},
1272
                    },
1273
                    "circuit_scheduler": [
1274
                        {
1275
                            "id": "1",
1276
                            "frequency": "* * * * *",
1277
                            "action": "create"
1278
                        }
1279
                    ],
1280
                }
1281
            }
1282
        }
1283
1284 1
        validate_mock.return_value = True
1285 1
        mongo_controller_upsert_mock.return_value = True
1286 1
        sched_add_mock.return_value = True
1287 1
        uni_from_dict_mock.side_effect = ["uni_a", "uni_z"]
1288 1
        evc_as_dict_mock.return_value = {}
1289 1
        self.napp.mongo_controller.get_circuits.return_value = mongo_payload_1
1290 1
        scheduler_remove_job_mock.return_value = True
1291
1292 1
        requested_schedule_id = "1"
1293 1
        url = f"{self.base_endpoint}/v2/evc/schedule/{requested_schedule_id}"
1294
1295 1
        payload = {"frequency": "*/1 * * * *", "action": "create"}
1296
1297
        # Call URL
1298 1
        response = await self.api_client.patch(url, json=payload)
1299 1
        response_json = response.json()
1300
1301 1
        assert response.status_code == 200
1302 1
        scheduler_remove_job_mock.assert_called_once()
1303 1
        mongo_controller_upsert_mock.assert_called_once()
1304 1
        assert payload["frequency"] == response_json["frequency"]
1305 1
        assert payload["action"] == response_json["action"]
1306 1
        assert response_json["id"] is not None
1307
1308 1
    @patch('napps.kytos.mef_eline.main.Main._find_evc_by_schedule_id')
1309 1
    async def test_update_no_schedule(
1310
        self, find_evc_by_schedule_id_mock
1311
    ):
1312
        """Test update a circuit schedule."""
1313 1
        self.napp.controller.loop = asyncio.get_running_loop()
1314 1
        url = f"{self.base_endpoint}/v2/evc/schedule/1"
1315 1
        payload = {"frequency": "*/1 * * * *", "action": "create"}
1316
1317 1
        find_evc_by_schedule_id_mock.return_value = None, None
1318
1319 1
        response = await self.api_client.patch(url, json=payload)
1320
        assert response.status_code == 404
1321
1322 1
    @patch("apscheduler.schedulers.background.BackgroundScheduler.remove_job")
1323 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
1324 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
1325 1
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
1326 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
1327 1
    async def test_delete_schedule(self, *args):
1328
        """Test create a circuit schedule."""
1329 1
        (
1330
            validate_mock,
1331
            evc_as_dict_mock,
1332
            mongo_controller_upsert_mock,
1333
            uni_from_dict_mock,
1334
            scheduler_remove_job_mock,
1335
        ) = args
1336
1337 1
        mongo_payload_1 = {
1338
            "circuits": {
1339
                "2": {
1340
                    "id": "2",
1341
                    "name": "my evc1",
1342
                    "uni_a": {
1343
                        "interface_id": "00:00:00:00:00:00:00:01:1",
1344
                        "tag": {"tag_type": 'vlan', "value": 80},
1345
                    },
1346
                    "uni_z": {
1347
                        "interface_id": "00:00:00:00:00:00:00:02:2",
1348
                        "tag": {"tag_type": 'vlan', "value": 1},
1349
                    },
1350
                    "circuit_scheduler": [
1351
                        {
1352
                            "id": "1",
1353
                            "frequency": "* * * * *",
1354
                            "action": "create"
1355
                        }
1356
                    ],
1357
                }
1358
            }
1359
        }
1360 1
        validate_mock.return_value = True
1361 1
        mongo_controller_upsert_mock.return_value = True
1362 1
        uni_from_dict_mock.side_effect = ["uni_a", "uni_z"]
1363 1
        evc_as_dict_mock.return_value = {}
1364 1
        self.napp.mongo_controller.get_circuits.return_value = mongo_payload_1
1365 1
        scheduler_remove_job_mock.return_value = True
1366
1367 1
        requested_schedule_id = "1"
1368 1
        url = f"{self.base_endpoint}/v2/evc/schedule/{requested_schedule_id}"
1369
1370
        # Call URL
1371 1
        response = await self.api_client.delete(url)
1372
1373 1
        assert response.status_code == 200
1374 1
        scheduler_remove_job_mock.assert_called_once()
1375 1
        mongo_controller_upsert_mock.assert_called_once()
1376 1
        assert "Schedule removed" in f"{response.json()}"
1377
1378 1
    @patch('napps.kytos.mef_eline.main.Main._find_evc_by_schedule_id')
1379 1
    async def test_delete_schedule_not_found(self, mock_find_evc_by_sched):
1380
        """Test delete a circuit schedule - unexisting."""
1381 1
        mock_find_evc_by_sched.return_value = (None, False)
1382 1
        url = f'{self.base_endpoint}/v2/evc/schedule/1'
1383 1
        response = await self.api_client.delete(url)
1384
        assert response.status_code == 404
1385
1386 1
    def test_get_evcs_by_svc_level(self) -> None:
1387
        """Test get_evcs_by_svc_level."""
1388 1
        levels = [1, 2, 4, 2, 7]
1389 1
        evcs = {i: MagicMock(service_level=v, creation_time=1)
1390
                for i, v in enumerate(levels)}
1391 1
        self.napp.circuits = evcs
1392 1
        expected_levels = sorted(levels, reverse=True)
1393 1
        evcs_by_level = self.napp.get_evcs_by_svc_level()
1394 1
        assert evcs_by_level
1395
1396 1
        for evc, exp_level in zip(evcs_by_level, expected_levels):
1397 1
            assert evc.service_level == exp_level
1398
1399 1
        evcs = {i: MagicMock(service_level=1, creation_time=i)
1400
                for i in reversed(range(2))}
1401 1
        self.napp.circuits = evcs
1402 1
        evcs_by_level = self.napp.get_evcs_by_svc_level()
1403 1
        for i in range(2):
1404 1
            assert evcs_by_level[i].creation_time == i
1405
1406 1
        self.napp.circuits[1].is_enabled = lambda: False
1407 1
        evcs_by_level = self.napp.get_evcs_by_svc_level()
1408 1
        assert len(evcs_by_level) == 1
1409
1410 1
        self.napp.circuits[1].is_enabled = lambda: False
1411 1
        evcs_by_level = self.napp.get_evcs_by_svc_level(enable_filter=False)
1412 1
        assert len(evcs_by_level) == 2
1413
1414 1
    async def test_get_circuit_not_found(self):
1415
        """Test /v2/evc/<circuit_id> 404."""
1416 1
        self.napp.mongo_controller.get_circuit.return_value = None
1417 1
        url = f'{self.base_endpoint}/v2/evc/1234'
1418 1
        response = await self.api_client.get(url)
1419
        assert response.status_code == 404
1420
1421 1
    @patch('httpx.post')
1422 1
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
1423 1
    @patch('napps.kytos.mef_eline.scheduler.Scheduler.add')
1424 1
    @patch('napps.kytos.mef_eline.controllers.ELineController.update_evc')
1425 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
1426 1
    @patch('napps.kytos.mef_eline.models.evc.EVC._validate')
1427 1
    @patch('kytos.core.Controller.get_interface_by_id')
1428 1
    @patch('napps.kytos.mef_eline.models.path.Path.is_valid')
1429 1
    @patch('napps.kytos.mef_eline.models.evc.EVCDeploy.deploy')
1430 1
    @patch('napps.kytos.mef_eline.main.Main._uni_from_dict')
1431 1
    @patch('napps.kytos.mef_eline.main.EVC.as_dict')
1432 1
    async def test_update_circuit(
1433
        self,
1434
        evc_as_dict_mock,
1435
        uni_from_dict_mock,
1436
        evc_deploy,
1437
        _is_valid_mock,
1438
        interface_by_id_mock,
1439
        _mock_validate,
1440
        _mongo_controller_upsert_mock,
1441
        _mongo_controller_update_mock,
1442
        _sched_add_mock,
1443
        mock_use_uni_tags,
1444
        httpx_mock
1445
    ):
1446
        """Test update a circuit circuit."""
1447 1
        self.napp.controller.loop = asyncio.get_running_loop()
1448 1
        mock_use_uni_tags.return_value = True
1449 1
        evc_deploy.return_value = True
1450 1
        interface_by_id_mock.return_value = get_uni_mocked().interface
1451 1
        unis = [
1452
            get_uni_mocked(switch_dpid="00:00:00:00:00:00:00:01"),
1453
            get_uni_mocked(switch_dpid="00:00:00:00:00:00:00:02"),
1454
        ]
1455 1
        uni_from_dict_mock.side_effect = 2 * unis
1456
1457 1
        response = MagicMock()
1458 1
        response.status_code = 201
1459 1
        httpx_mock.return_value = response
1460
1461 1
        payloads = [
1462
            {
1463
                "name": "my evc1",
1464
                "uni_a": {
1465
                    "interface_id": "00:00:00:00:00:00:00:01:1",
1466
                    "tag": {"tag_type": 'vlan', "value": 80},
1467
                },
1468
                "uni_z": {
1469
                    "interface_id": "00:00:00:00:00:00:00:02:2",
1470
                    "tag": {"tag_type": 'vlan', "value": 1},
1471
                },
1472
                "dynamic_backup_path": True,
1473
            },
1474
            {
1475
                "primary_path": [
1476
                    {
1477
                        "endpoint_a": {"id": "00:00:00:00:00:00:00:01:1"},
1478
                        "endpoint_b": {"id": "00:00:00:00:00:00:00:02:2"},
1479
                    }
1480
                ]
1481
            },
1482
            {
1483
                "sb_priority": 3
1484
            },
1485
            {
1486
                # It works only with 'enabled' and not with 'enable'
1487
                "enabled": True
1488
            },
1489
            {
1490
                "sb_priority": 100
1491
            }
1492
        ]
1493
1494 1
        evc_as_dict_mock.return_value = payloads[0]
1495 1
        response = await self.api_client.post(
1496
            f"{self.base_endpoint}/v2/evc/",
1497
            json=payloads[0],
1498
        )
1499 1
        assert 201 == response.status_code
1500
1501 1
        evc_deploy.reset_mock()
1502 1
        evc_as_dict_mock.return_value = payloads[1]
1503 1
        current_data = response.json()
1504 1
        circuit_id = current_data["circuit_id"]
1505 1
        response = await self.api_client.patch(
1506
            f"{self.base_endpoint}/v2/evc/{circuit_id}",
1507
            json=payloads[1],
1508
        )
1509
        # evc_deploy.assert_called_once()
1510 1
        assert 200 == response.status_code
1511
1512 1
        evc_deploy.reset_mock()
1513 1
        evc_as_dict_mock.return_value = payloads[2]
1514 1
        response = await self.api_client.patch(
1515
            f"{self.base_endpoint}/v2/evc/{circuit_id}",
1516
            json=payloads[2],
1517
        )
1518 1
        evc_deploy.assert_not_called()
1519 1
        assert 200 == response.status_code
1520
1521 1
        evc_deploy.reset_mock()
1522 1
        evc_as_dict_mock.return_value = payloads[3]
1523 1
        response = await self.api_client.patch(
1524
            f"{self.base_endpoint}/v2/evc/{circuit_id}",
1525
            json=payloads[3],
1526
        )
1527 1
        evc_deploy.assert_called_once()
1528 1
        assert 200 == response.status_code
1529
1530 1
        evc_deploy.reset_mock()
1531 1
        response = await self.api_client.patch(
1532
            f"{self.base_endpoint}/v2/evc/{circuit_id}",
1533
            content=b'{"priority":5,}',
1534
            headers={"Content-Type": "application/json"}
1535
        )
1536
        evc_deploy.assert_not_called()
1537
        assert 400 == response.status_code
1538
1539
        response = await self.api_client.patch(
1540
            f"{self.base_endpoint}/v2/evc/1234",
1541
            json=payloads[1],
1542
        )
1543
        current_data = response.json()
1544
        expected_data = "circuit_id 1234 not found"
1545
        assert current_data["description"] == expected_data
1546
        assert 404 == response.status_code
1547
1548
        self.napp.circuits[circuit_id]._active = False
1549
        evc_deploy.reset_mock()
1550
        response = await self.api_client.patch(
1551
            f"{self.base_endpoint}/v2/evc/{circuit_id}",
1552
            json=payloads[4]
1553
        )
1554 1
        assert 200 == response.status_code
1555 1
        evc_deploy.assert_called_once()
1556
1557 1 View Code Duplication
    @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
1558 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
1559 1
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
1560 1
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
1561 1
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
1562 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
1563 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
1564 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
1565 1
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
1566 1
    async def test_update_circuit_invalid_json(
1567
        self,
1568
        evc_as_dict_mock,
1569
        validate_mock,
1570
        mongo_controller_upsert_mock,
1571
        uni_from_dict_mock,
1572
        sched_add_mock,
1573
        evc_deploy_mock,
1574
        mock_use_uni_tags,
1575
        mock_tags_equal,
1576
        mock_check_duplicate
1577
    ):
1578
        """Test update a circuit circuit."""
1579 1
        self.napp.controller.loop = asyncio.get_running_loop()
1580 1
        validate_mock.return_value = True
1581 1
        mongo_controller_upsert_mock.return_value = True
1582 1
        sched_add_mock.return_value = True
1583 1
        evc_deploy_mock.return_value = True
1584 1
        mock_use_uni_tags.return_value = True
1585 1
        mock_tags_equal.return_value = True
1586 1
        mock_check_duplicate.return_value = True
1587 1
        uni1 = create_autospec(UNI)
1588 1
        uni2 = create_autospec(UNI)
1589 1
        uni1.interface = create_autospec(Interface)
1590 1
        uni2.interface = create_autospec(Interface)
1591 1
        uni1.interface.switch = "00:00:00:00:00:00:00:01"
1592 1
        uni2.interface.switch = "00:00:00:00:00:00:00:02"
1593 1
        uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2]
1594
1595 1
        payload1 = {
1596
            "name": "my evc1",
1597
            "uni_a": {
1598
                "interface_id": "00:00:00:00:00:00:00:01:1",
1599
                "tag": {"tag_type": 'vlan', "value": 80},
1600
            },
1601
            "uni_z": {
1602
                "interface_id": "00:00:00:00:00:00:00:02:2",
1603
                "tag": {"tag_type": 'vlan', "value": 1},
1604
            },
1605
            "dynamic_backup_path": True,
1606
        }
1607
1608 1
        payload2 = {
1609
            "dynamic_backup_path": False,
1610
        }
1611
1612 1
        evc_as_dict_mock.return_value = payload1
1613 1
        response = await self.api_client.post(
1614
            f"{self.base_endpoint}/v2/evc/",
1615
            json=payload1
1616
        )
1617 1
        assert 201 == response.status_code
1618
1619 1
        evc_as_dict_mock.return_value = payload2
1620 1
        current_data = response.json()
1621 1
        circuit_id = current_data["circuit_id"]
1622 1
        response = await self.api_client.patch(
1623
            f"{self.base_endpoint}/v2/evc/{circuit_id}",
1624
            json=payload2
1625
        )
1626
        current_data = response.json()
1627
        assert 400 == response.status_code
1628
        assert "must have a primary path or" in current_data["description"]
1629
1630 1
    @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
1631 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
1632 1
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
1633 1
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
1634 1
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
1635 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
1636 1
    @patch("napps.kytos.mef_eline.main.Main._link_from_dict")
1637 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
1638 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
1639 1
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
1640 1
    @patch("napps.kytos.mef_eline.models.path.Path.is_valid")
1641 1
    async def test_update_circuit_invalid_path(
1642
        self,
1643
        is_valid_mock,
1644
        evc_as_dict_mock,
1645
        validate_mock,
1646
        mongo_controller_upsert_mock,
1647
        link_from_dict_mock,
1648
        uni_from_dict_mock,
1649
        sched_add_mock,
1650
        evc_deploy_mock,
1651
        mock_use_uni_tags,
1652
        mock_tags_equal,
1653
        mock_check_duplicate
1654
    ):
1655
        """Test update a circuit circuit."""
1656 1
        self.napp.controller.loop = asyncio.get_running_loop()
1657 1
        is_valid_mock.side_effect = InvalidPath("error")
1658 1
        validate_mock.return_value = True
1659 1
        mongo_controller_upsert_mock.return_value = True
1660 1
        sched_add_mock.return_value = True
1661 1
        evc_deploy_mock.return_value = True
1662 1
        mock_use_uni_tags.return_value = True
1663 1
        link_from_dict_mock.return_value = 1
1664 1
        mock_tags_equal.return_value = True
1665 1
        mock_check_duplicate.return_value = True
1666 1
        uni1 = create_autospec(UNI)
1667 1
        uni2 = create_autospec(UNI)
1668 1
        uni1.interface = create_autospec(Interface)
1669 1
        uni2.interface = create_autospec(Interface)
1670 1
        uni1.interface.switch = "00:00:00:00:00:00:00:01"
1671 1
        uni2.interface.switch = "00:00:00:00:00:00:00:02"
1672 1
        uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2]
1673
1674 1
        payload1 = {
1675
            "name": "my evc1",
1676
            "uni_a": {
1677
                "interface_id": "00:00:00:00:00:00:00:01:1",
1678
                "tag": {"tag_type": 'vlan', "value": 80},
1679
            },
1680
            "uni_z": {
1681
                "interface_id": "00:00:00:00:00:00:00:02:2",
1682
                "tag": {"tag_type": 'vlan', "value": 1},
1683
            },
1684
            "dynamic_backup_path": True,
1685
        }
1686
1687 1
        payload2 = {
1688
            "primary_path": [
1689
                {
1690
                    "endpoint_a": {"id": "00:00:00:00:00:00:00:01:1"},
1691
                    "endpoint_b": {"id": "00:00:00:00:00:00:00:02:2"},
1692
                }
1693
            ]
1694
        }
1695
1696 1
        evc_as_dict_mock.return_value = payload1
1697 1
        response = await self.api_client.post(
1698
            f"{self.base_endpoint}/v2/evc/",
1699
            json=payload1,
1700
        )
1701 1
        assert 201 == response.status_code
1702
1703 1
        evc_as_dict_mock.return_value = payload2
1704 1
        current_data = response.json()
1705 1
        circuit_id = current_data["circuit_id"]
1706 1
        response = await self.api_client.patch(
1707
            f"{self.base_endpoint}/v2/evc/{circuit_id}",
1708
            json=payload2,
1709
        )
1710
        current_data = response.json()
1711
        expected_data = "primary_path is not a valid path: error"
1712
        assert 400 == response.status_code
1713
        assert current_data["description"] == expected_data
1714
1715 1
    def test_link_from_dict_non_existent_intf(self):
1716
        """Test _link_from_dict non existent intf."""
1717 1
        self.napp.controller.get_interface_by_id = MagicMock(return_value=None)
1718 1
        link_dict = {
1719
            "endpoint_a": {"id": "a"},
1720
            "endpoint_b": {"id": "b"}
1721
        }
1722 1
        with pytest.raises(ValueError):
1723 1
            self.napp._link_from_dict(link_dict, "current_path")
1724
1725 1
    def test_link_from_dict_vlan_metadata(self):
1726
        """Test that link_from_dict only accepts vlans for current_path
1727
         and failover_path."""
1728 1
        intf = MagicMock(id="01:1")
1729 1
        self.napp.controller.get_interface_by_id = MagicMock(return_value=intf)
1730 1
        link_dict = {
1731
            'id': 'mock_link',
1732
            'endpoint_a': {'id': '00:00:00:00:00:00:00:01:4'},
1733
            'endpoint_b': {'id': '00:00:00:00:00:00:00:05:2'},
1734
            'metadata': {'s_vlan': {'tag_type': 'vlan', 'value': 1}}
1735
        }
1736 1
        link = self.napp._link_from_dict(link_dict, "current_path")
1737 1
        assert link.metadata.get('s_vlan', None)
1738
1739 1
        link = self.napp._link_from_dict(link_dict, "failover_path")
1740 1
        assert link.metadata.get('s_vlan', None)
1741
1742 1
        link = self.napp._link_from_dict(link_dict, "primary_path")
1743 1
        assert link.metadata.get('s_vlan', None) is None
1744
1745 1
    def test_uni_from_dict_non_existent_intf(self):
1746
        """Test _link_from_dict non existent intf."""
1747 1
        self.napp.controller.get_interface_by_id = MagicMock(return_value=None)
1748 1
        uni_dict = {
1749
            "interface_id": "aaa",
1750
        }
1751 1
        with pytest.raises(ValueError):
1752 1
            self.napp._uni_from_dict(uni_dict)
1753
1754 1 View Code Duplication
    @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
1755 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
1756 1
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
1757 1
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
1758 1
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
1759 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
1760 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
1761 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
1762 1
    async def test_update_evc_no_json_mime(
1763
        self,
1764
        mongo_controller_upsert_mock,
1765
        validate_mock,
1766
        uni_from_dict_mock,
1767
        sched_add_mock,
1768
        evc_deploy_mock,
1769
        mock_use_uni_tags,
1770
        mock_tags_equal,
1771
        mock_check_duplicate
1772
    ):
1773
        """Test update a circuit with wrong mimetype."""
1774 1
        self.napp.controller.loop = asyncio.get_running_loop()
1775 1
        validate_mock.return_value = True
1776 1
        sched_add_mock.return_value = True
1777 1
        evc_deploy_mock.return_value = True
1778 1
        mock_use_uni_tags.return_value = True
1779 1
        mock_tags_equal.return_value = True
1780 1
        mock_check_duplicate.return_value = True
1781 1
        uni1 = create_autospec(UNI)
1782 1
        uni2 = create_autospec(UNI)
1783 1
        uni1.interface = create_autospec(Interface)
1784 1
        uni2.interface = create_autospec(Interface)
1785 1
        uni1.interface.switch = "00:00:00:00:00:00:00:01"
1786 1
        uni2.interface.switch = "00:00:00:00:00:00:00:02"
1787 1
        uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2]
1788 1
        mongo_controller_upsert_mock.return_value = True
1789
1790 1
        payload1 = {
1791
            "name": "my evc1",
1792
            "uni_a": {
1793
                "interface_id": "00:00:00:00:00:00:00:01:1",
1794
                "tag": {"tag_type": 'vlan', "value": 80},
1795
            },
1796
            "uni_z": {
1797
                "interface_id": "00:00:00:00:00:00:00:02:2",
1798
                "tag": {"tag_type": 'vlan', "value": 1},
1799
            },
1800
            "dynamic_backup_path": True,
1801
        }
1802
1803 1
        payload2 = {"dynamic_backup_path": False}
1804
1805 1
        response = await self.api_client.post(
1806
            f"{self.base_endpoint}/v2/evc/",
1807
            json=payload1,
1808
        )
1809 1
        assert 201 == response.status_code
1810
1811 1
        current_data = response.json()
1812 1
        circuit_id = current_data["circuit_id"]
1813 1
        response = await self.api_client.patch(
1814
            f"{self.base_endpoint}/v2/evc/{circuit_id}", data=payload2
1815
        )
1816
        current_data = response.json()
1817
        assert 415 == response.status_code
1818
        assert "application/json" in current_data["description"]
1819
1820 1
    async def test_delete_no_evc(self):
1821
        """Test delete when EVC does not exist."""
1822 1
        url = f"{self.base_endpoint}/v2/evc/123"
1823 1
        response = await self.api_client.delete(url)
1824
        current_data = response.json()
1825
        expected_data = "circuit_id 123 not found"
1826
        assert current_data["description"] == expected_data
1827
        assert 404 == response.status_code
1828
1829 1
    @patch("napps.kytos.mef_eline.main.Main._check_no_tag_duplication")
1830 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._tag_lists_equal")
1831 1
    @patch("napps.kytos.mef_eline.models.evc.EVC.remove_uni_tags")
1832 1
    @patch("napps.kytos.mef_eline.main.Main._use_uni_tags")
1833 1
    @patch("napps.kytos.mef_eline.models.evc.EVC.remove_current_flows")
1834 1
    @patch("napps.kytos.mef_eline.models.evc.EVC.deploy")
1835 1
    @patch("napps.kytos.mef_eline.scheduler.Scheduler.add")
1836 1
    @patch("napps.kytos.mef_eline.main.Main._uni_from_dict")
1837 1
    @patch("napps.kytos.mef_eline.controllers.ELineController.upsert_evc")
1838 1
    @patch("napps.kytos.mef_eline.models.evc.EVC._validate")
1839 1
    @patch("napps.kytos.mef_eline.main.EVC.as_dict")
1840 1
    async def test_delete_archived_evc(
1841
        self,
1842
        evc_as_dict_mock,
1843
        validate_mock,
1844
        mongo_controller_upsert_mock,
1845
        uni_from_dict_mock,
1846
        sched_add_mock,
1847
        evc_deploy_mock,
1848
        remove_current_flows_mock,
1849
        mock_use_uni,
1850
        mock_remove_tags,
1851
        mock_tags_equal,
1852
        mock_check_duplicate
1853
    ):
1854
        """Try to delete an archived EVC"""
1855 1
        self.napp.controller.loop = asyncio.get_running_loop()
1856 1
        validate_mock.return_value = True
1857 1
        mongo_controller_upsert_mock.return_value = True
1858 1
        sched_add_mock.return_value = True
1859 1
        evc_deploy_mock.return_value = True
1860 1
        mock_use_uni.return_value = True
1861 1
        mock_tags_equal.return_value = True
1862 1
        mock_check_duplicate.return_value = True
1863 1
        uni1 = create_autospec(UNI)
1864 1
        uni2 = create_autospec(UNI)
1865 1
        uni1.interface = create_autospec(Interface)
1866 1
        uni2.interface = create_autospec(Interface)
1867 1
        uni1.interface.switch = "00:00:00:00:00:00:00:01"
1868 1
        uni2.interface.switch = "00:00:00:00:00:00:00:02"
1869 1
        uni_from_dict_mock.side_effect = [uni1, uni2, uni1, uni2]
1870
1871 1
        payload1 = {
1872
            "name": "my evc1",
1873
            "uni_a": {
1874
                "interface_id": "00:00:00:00:00:00:00:01:1",
1875
                "tag": {"tag_type": 'vlan', "value": 80},
1876
            },
1877
            "uni_z": {
1878
                "interface_id": "00:00:00:00:00:00:00:02:2",
1879
                "tag": {"tag_type": 'vlan', "value": 1},
1880
            },
1881
            "dynamic_backup_path": True,
1882
        }
1883
1884 1
        evc_as_dict_mock.return_value = payload1
1885 1
        response = await self.api_client.post(
1886
            f"{self.base_endpoint}/v2/evc/",
1887
            json=payload1
1888
        )
1889 1
        assert 201 == response.status_code
1890 1
        assert len(self.napp.circuits) == 1
1891 1
        current_data = response.json()
1892 1
        circuit_id = current_data["circuit_id"]
1893 1
        self.napp.circuits[circuit_id].archive()
1894
1895 1
        response = await self.api_client.delete(
1896
            f"{self.base_endpoint}/v2/evc/{circuit_id}"
1897
        )
1898 1
        assert 200 == response.status_code
1899 1
        assert mock_remove_tags.call_count == 0
1900 1
        assert remove_current_flows_mock.call_count == 0
1901 1
        assert len(self.napp.circuits) == 0
1902
1903 1
        response = await self.api_client.delete(
1904
            f"{self.base_endpoint}/v2/evc/{circuit_id}"
1905
        )
1906
        current_data = response.json()
1907
        expected_data = f"circuit_id {circuit_id} not found"
1908
        assert current_data["description"] == expected_data
1909
        assert 404 == response.status_code
1910
        assert len(self.napp.circuits) == 0
1911
1912 1
    @patch("napps.kytos.mef_eline.main.emit_event")
1913 1
    async def test_delete_circuit(self, emit_event_mock):
1914
        """Test delete_circuit"""
1915 1
        evc = MagicMock()
1916 1
        evc.archived = False
1917 1
        circuit_id = '1'
1918 1
        self.napp.circuits[circuit_id] = evc
1919 1
        response = await self.api_client.delete(
1920
            f"{self.base_endpoint}/v2/evc/{circuit_id}"
1921
        )
1922 1
        assert 200 == response.status_code
1923 1
        assert evc.deactivate.call_count == 1
1924 1
        assert evc.disable.call_count == 1
1925 1
        evc.remove_current_flows.assert_called_once_with(
1926
            sync=False
1927
        )
1928 1
        evc.remove_failover_flows.assert_called_once_with(
1929
            sync=False
1930
        )
1931 1
        assert evc.archive.call_count == 1
1932 1
        assert evc.remove_uni_tags.call_count == 1
1933 1
        assert evc.sync.call_count == 1
1934 1
        assert not self.napp.circuits
1935 1
        assert emit_event_mock.call_count == 1
1936
1937 1
    def test_handle_link_up(self):
1938
        """Test handle_link_up method."""
1939 1
        evc_mock = create_autospec(EVC)
1940 1
        evc_mock.service_level, evc_mock.creation_time = 0, 1
1941 1
        evc_mock.is_enabled = MagicMock(side_effect=[
1942
            True, False, True, True, True
1943
        ])
1944 1
        evc_mock.lock = MagicMock()
1945 1
        evc_mock.archived = False
1946 1
        evcs = [evc_mock, evc_mock, evc_mock]
1947 1
        event = KytosEvent(name="test", content={"link": "abc"})
1948 1
        self.napp.circuits = dict(zip(["1", "2", "3"], evcs))
1949 1
        self.napp.handle_link_up(event)
1950 1
        assert evc_mock.handle_link_up.call_count == 2
1951 1
        evc_mock.handle_link_up.assert_called_with("abc")
1952
1953 1
    def test_handle_link_down(
1954
        self
1955
    ):
1956
        """Test handle_link_down method."""
1957 1
        evc1 = MagicMock(id="1")
1958 1
        evc1.is_affected_by_link.return_value = True
1959 1
        evc1.is_failover_path_affected_by_link.return_value = True
1960
1961 1
        evc2 = MagicMock(id="2")
1962 1
        evc2.is_affected_by_link.return_value = True
1963 1
        evc2.failover_path = Path([])
1964
1965 1
        default_undeploy = [evc1, evc2]
1966
1967 1
        evc3 = MagicMock(id="3")
1968 1
        evc3.is_affected_by_link.return_value = True
1969 1
        evc3.is_failover_path_affected_by_link.return_value = False
1970
1971 1
        evc4 = MagicMock(id="4")
1972 1
        evc4.is_affected_by_link.return_value = True
1973 1
        evc4.is_failover_path_affected_by_link.return_value = False
1974
1975 1
        default_swap_to_failover = [evc3, evc4]
1976
1977 1
        evc5 = MagicMock(id="5")
1978 1
        evc5.is_affected_by_link.return_value = False
1979 1
        evc5.is_failover_path_affected_by_link.return_value = True
1980
1981 1
        evc6 = MagicMock(id="6")
1982 1
        evc6.is_affected_by_link.return_value = False
1983 1
        evc6.is_failover_path_affected_by_link.return_value = True
1984
1985 1
        default_clear_failover = [evc5, evc6]
1986
1987 1
        self.napp.get_evcs_by_svc_level = MagicMock()
1988
1989 1
        self.napp.get_evcs_by_svc_level.return_value = [
1990
            evc1,
1991
            evc2,
1992
            evc3,
1993
            evc4,
1994
            evc5,
1995
            evc6,
1996
        ]
1997
1998 1
        self.napp.execute_swap_to_failover = MagicMock()
1999
2000 1
        swap_to_failover_success = [evc3]
2001 1
        swap_to_failover_failure = [evc4]
2002
2003 1
        self.napp.execute_swap_to_failover.return_value =\
2004
            swap_to_failover_success, swap_to_failover_failure
2005
2006 1
        self.napp.execute_clear_failover = MagicMock()
2007
2008 1
        clear_failover_success = [evc5, evc3]
2009 1
        clear_failover_failure = [evc6]
2010
2011 1
        self.napp.execute_clear_failover.return_value =\
2012
            clear_failover_success, clear_failover_failure
2013
2014 1
        self.napp.execute_undeploy = MagicMock()
2015
2016 1
        undeploy_success = [evc1, evc4, evc6]
2017 1
        undeploy_failure = [evc2]
2018
2019 1
        self.napp.execute_undeploy.return_value =\
2020
            undeploy_success, undeploy_failure
2021
2022 1
        link = MagicMock(id="123")
2023 1
        event = KytosEvent(name="test", content={"link": link})
2024
2025 1
        self.napp.handle_link_down(event)
2026
2027 1
        self.napp.execute_swap_to_failover.assert_called_with(
2028
            default_swap_to_failover
2029
        )
2030
2031 1
        self.napp.execute_clear_failover.assert_called_with(
2032
            [*default_clear_failover, *swap_to_failover_success]
2033
        )
2034
2035 1
        self.napp.execute_undeploy.assert_called_with([
2036
            *default_undeploy,
2037
            *swap_to_failover_failure,
2038
            *clear_failover_failure
2039
        ])
2040
2041 1
        self.napp.mongo_controller.update_evcs.assert_called_with([
2042
            evc3.as_dict(),
2043
            evc5.as_dict(),
2044
            evc1.as_dict(),
2045
            evc4.as_dict(),
2046
            evc6.as_dict(),
2047
        ])
2048
2049 1 View Code Duplication
    @patch("napps.kytos.mef_eline.main.emit_event")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
2050 1
    @patch("napps.kytos.mef_eline.main.send_flow_mods_http")
2051 1
    def test_execute_swap_to_failover(
2052
        self,
2053
        send_flow_mods_mock: MagicMock,
2054
        emit_main_mock: MagicMock,
2055
    ):
2056
        """Test execute_swap_to_failover method."""
2057 1
        evc1 = MagicMock(id="1")
2058 1
        good_path = MagicMock(id="GoodPath")
2059 1
        bad_path = MagicMock(id="BadPath")
2060 1
        evc1.current_path = bad_path
2061 1
        evc1.failover_path = good_path
2062 1
        evc2 = MagicMock(id="2")
2063
2064 1
        self.napp.prepare_swap_to_failover_flow = {
2065
            evc1: {"1": ["Flow1"]},
2066
            evc2: None
2067
        }.get
2068
2069 1
        self.napp.prepare_swap_to_failover_event = {
2070
            evc1: "FailoverEvent1",
2071
        }.get
2072
2073 1
        success, failure = self.napp.execute_swap_to_failover([evc1, evc2])
2074
2075 1
        assert success == [evc1]
2076 1
        assert failure == [evc2]
2077
2078 1
        send_flow_mods_mock.assert_called_with(
2079
            {"1": ["Flow1"]},
2080
            "install"
2081
        )
2082
2083 1
        assert evc1.current_path == good_path
2084 1
        assert evc1.failover_path == bad_path
2085
2086 1
        emit_main_mock.assert_called_with(
2087
            self.napp.controller,
2088
            "failover_link_down",
2089
            content={"1": "FailoverEvent1"}
2090
        )
2091
2092 1 View Code Duplication
    @patch("napps.kytos.mef_eline.main.emit_event")
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
2093 1
    @patch("napps.kytos.mef_eline.main.send_flow_mods_http")
2094 1
    def test_execute_swap_to_failover_exception(
2095
        self,
2096
        send_flow_mods_mock: MagicMock,
2097
        emit_main_mock: MagicMock,
2098
    ):
2099
        """Test handle_link_down method when an exception occurs."""
2100 1
        evc1 = MagicMock(id="1")
2101 1
        good_path = MagicMock(id="GoodPath")
2102 1
        bad_path = MagicMock(id="BadPath")
2103 1
        evc1.current_path = bad_path
2104 1
        evc1.failover_path = good_path
2105 1
        evc2 = MagicMock(id="2")
2106
2107 1
        self.napp.prepare_swap_to_failover_flow = {
2108
            evc1: {"1": ["Flow1"]},
2109
            evc2: None
2110
        }.get
2111
2112 1
        self.napp.prepare_swap_to_failover_event = {
2113
            evc1: "FailoverEvent1",
2114
        }.get
2115
2116 1
        send_flow_mods_mock.side_effect = FlowModException(
2117
            "Flowmod failed to send"
2118
        )
2119
2120 1
        success, failure = self.napp.execute_swap_to_failover([evc1, evc2])
2121
2122 1
        assert success == []
2123 1
        assert failure == [evc1, evc2]
2124
2125 1
        send_flow_mods_mock.assert_called_with(
2126
            {"1": ["Flow1"]},
2127
            "install"
2128
        )
2129
2130 1
        assert evc1.current_path == bad_path
2131 1
        assert evc1.failover_path == good_path
2132
2133 1
        emit_main_mock.assert_not_called()
2134
2135 1
    @patch("napps.kytos.mef_eline.main.emit_event")
2136 1
    @patch("napps.kytos.mef_eline.main.send_flow_mods_http")
2137 1
    def test_execute_clear_failover(
2138
        self,
2139
        send_flow_mods_mock: MagicMock,
2140
        emit_main_mock: MagicMock,
2141
    ):
2142
        """Test execute_clear_failover method."""
2143 1
        evc1 = MagicMock(id="1")
2144 1
        good_path = MagicMock(id="GoodPath")
2145 1
        bad_path = MagicMock(id="BadPath")
2146 1
        evc1.current_path = good_path
2147 1
        evc1.failover_path = bad_path
2148 1
        evc2 = MagicMock(id="2")
2149
2150 1
        self.napp.prepare_clear_failover_flow = {
2151
            evc1: {"1": ["Flow1"]},
2152
            evc2: None
2153
        }.get
2154
2155 1
        self.napp.prepare_clear_failover_event = {
2156
            evc1: "FailoverEvent1",
2157
        }.get
2158
2159 1
        success, failure = self.napp.execute_clear_failover([evc1, evc2])
2160
2161 1
        assert success == [evc1]
2162 1
        assert failure == [evc2]
2163
2164 1
        send_flow_mods_mock.assert_called_with(
2165
            {"1": ["Flow1"]},
2166
            "delete"
2167
        )
2168
2169 1
        assert evc1.current_path == good_path
2170 1
        assert not evc1.failover_path
2171
2172 1
        bad_path.make_vlans_available.assert_called_once_with(
2173
            self.napp.controller
2174
        )
2175
2176 1
        emit_main_mock.assert_called_with(
2177
            self.napp.controller,
2178
            "failover_old_path",
2179
            content={"1": "FailoverEvent1"}
2180
        )
2181
2182 1
    @patch("napps.kytos.mef_eline.main.emit_event")
2183 1
    @patch("napps.kytos.mef_eline.main.send_flow_mods_http")
2184 1
    def test_execute_clear_failover_exception(
2185
        self,
2186
        send_flow_mods_mock: MagicMock,
2187
        emit_main_mock: MagicMock,
2188
    ):
2189
        """Test execute_clear_failover method when an exception occurs."""
2190 1
        evc1 = MagicMock(id="1")
2191 1
        good_path = MagicMock(id="GoodPath")
2192 1
        bad_path = MagicMock(id="BadPath")
2193 1
        evc1.current_path = good_path
2194 1
        evc1.failover_path = bad_path
2195 1
        evc2 = MagicMock(id="2")
2196
2197 1
        self.napp.prepare_clear_failover_flow = {
2198
            evc1: {"1": ["Flow1"]},
2199
            evc2: None
2200
        }.get
2201
2202 1
        self.napp.prepare_clear_failover_event = {
2203
            evc1: "FailoverEvent1",
2204
        }.get
2205
2206 1
        send_flow_mods_mock.side_effect = FlowModException(
2207
            "Flowmod failed to send"
2208
        )
2209
2210 1
        success, failure = self.napp.execute_clear_failover([evc1, evc2])
2211
2212 1
        assert success == []
2213 1
        assert failure == [evc1, evc2]
2214
2215 1
        send_flow_mods_mock.assert_called_with(
2216
            {"1": ["Flow1"]},
2217
            "delete"
2218
        )
2219
2220 1
        assert evc1.current_path == good_path
2221 1
        assert evc1.failover_path == bad_path
2222
2223 1
        emit_main_mock.assert_not_called()
2224
2225 1
    @patch("napps.kytos.mef_eline.main.emit_event")
2226 1
    @patch("napps.kytos.mef_eline.main.send_flow_mods_http")
2227 1
    def test_execute_undeploy(
2228
        self,
2229
        send_flow_mods_mock: MagicMock,
2230
        emit_main_mock: MagicMock,
2231
    ):
2232
        """Test execute_undeploy method."""
2233 1
        evc1 = MagicMock(id="1")
2234 1
        bad_path1 = MagicMock(id="GoodPath")
2235 1
        bad_path2 = MagicMock(id="BadPath")
2236 1
        evc1.current_path = bad_path1
2237 1
        evc1.failover_path = bad_path2
2238 1
        evc2 = MagicMock(id="2")
2239
2240 1
        self.napp.prepare_undeploy_flow = {
2241
            evc1: {"1": ["Flow1"]},
2242
            evc2: None
2243
        }.get
2244
2245 1
        success, failure = self.napp.execute_undeploy([evc1, evc2])
2246
2247 1
        assert success == [evc1]
2248 1
        assert failure == [evc2]
2249
2250 1
        send_flow_mods_mock.assert_called_with(
2251
            {"1": ["Flow1"]},
2252
            "delete"
2253
        )
2254
2255 1
        assert not evc1.current_path
2256 1
        assert not evc1.failover_path
2257
2258 1
        assert evc2.current_path
2259 1
        assert evc2.failover_path
2260
2261 1
        bad_path1.make_vlans_available.assert_called_once_with(
2262
            self.napp.controller
2263
        )
2264 1
        bad_path2.make_vlans_available.assert_called_once_with(
2265
            self.napp.controller
2266
        )
2267
2268 1
        evc1.deactivate.assert_called()
2269 1
        evc2.deactivate.assert_not_called()
2270
2271 1
        emit_main_mock.assert_called_with(
2272
            self.napp.controller,
2273
            "need_redeploy",
2274
            content={"evc_id": "1"}
2275
        )
2276
2277 1
    @patch("napps.kytos.mef_eline.main.emit_event")
2278 1
    @patch("napps.kytos.mef_eline.main.send_flow_mods_http")
2279 1
    def test_execute_undeploy_exception(
2280
        self,
2281
        send_flow_mods_mock: MagicMock,
2282
        emit_main_mock: MagicMock,
2283
    ):
2284
        """Test execute_undeploy method method when an exception occurs."""
2285 1
        evc1 = MagicMock(id="1")
2286 1
        bad_path1 = MagicMock(id="GoodPath")
2287 1
        bad_path2 = MagicMock(id="BadPath")
2288 1
        evc1.current_path = bad_path1
2289 1
        evc1.failover_path = bad_path2
2290 1
        evc2 = MagicMock(id="2")
2291
2292 1
        self.napp.prepare_undeploy_flow = {
2293
            evc1: {"1": ["Flow1"]},
2294
            evc2: None
2295
        }.get
2296
2297 1
        send_flow_mods_mock.side_effect = FlowModException(
2298
            "Flowmod failed to send"
2299
        )
2300
2301 1
        success, failure = self.napp.execute_undeploy([evc1, evc2])
2302
2303 1
        assert success == []
2304 1
        assert failure == [evc1, evc2]
2305
2306 1
        send_flow_mods_mock.assert_called_with(
2307
            {"1": ["Flow1"]},
2308
            "delete"
2309
        )
2310
2311 1
        assert evc1.current_path
2312 1
        assert evc1.failover_path
2313
2314 1
        assert evc2.current_path
2315 1
        assert evc2.failover_path
2316
2317 1
        bad_path1.make_vlans_available.assert_not_called()
2318 1
        bad_path2.make_vlans_available.assert_not_called()
2319
2320 1
        evc1.deactivate.assert_not_called()
2321 1
        evc2.deactivate.assert_not_called()
2322
2323 1
        emit_main_mock.assert_not_called()
2324
2325 1
    @patch("napps.kytos.mef_eline.main.emit_event")
2326 1
    def test_handle_evc_affected_by_link_down(self, emit_event_mock):
2327
        """Test handle_evc_affected_by_link_down method."""
2328 1
        uni = create_autospec(UNI)
2329 1
        evc1 = MagicMock(
2330
            id="1",
2331
            metadata="data_mocked",
2332
            _active="true",
2333
            _enabled="false",
2334
            uni_a=uni,
2335
            uni_z=uni,
2336
        )
2337 1
        evc1.name = "name_mocked"
2338 1
        evc1.handle_link_down.return_value = True
2339 1
        evc2 = MagicMock(
2340
            id="2",
2341
            metadata="mocked_data",
2342
            _active="false",
2343
            _enabled="true",
2344
            uni_a=uni,
2345
            uni_z=uni,
2346
        )
2347 1
        evc2.name = "mocked_name"
2348 1
        evc2.handle_link_down.return_value = False
2349 1
        self.napp.circuits = {"1": evc1, "2": evc2}
2350
2351 1
        event = KytosEvent(name="e1", content={
2352
            "evc_id": "3",
2353
            "link": MagicMock(),
2354
        })
2355 1
        self.napp.handle_evc_affected_by_link_down(event)
2356 1
        emit_event_mock.assert_not_called()
2357 1
        event.content["evc_id"] = "1"
2358 1
        self.napp.handle_evc_affected_by_link_down(event)
2359 1
        emit_event_mock.assert_called_with(
2360
            self.napp.controller, "redeployed_link_down", content={
2361
                "id": "1",
2362
                "evc_id": "1",
2363
                "name": "name_mocked",
2364
                "metadata": "data_mocked",
2365
                "active": "true",
2366
                "enabled": "false",
2367
                "uni_a": uni.as_dict(),
2368
                "uni_z": uni.as_dict(),
2369
            }
2370
        )
2371
2372 1
        event.content["evc_id"] = "2"
2373 1
        self.napp.handle_evc_affected_by_link_down(event)
2374 1
        emit_event_mock.assert_called_with(
2375
            self.napp.controller, "error_redeploy_link_down", content={
2376
                "evc_id": "2",
2377
                "id": "2",
2378
                "name": "mocked_name",
2379
                "metadata": "mocked_data",
2380
                "active": "false",
2381
                "enabled": "true",
2382
                "uni_a": uni.as_dict(),
2383
                "uni_z": uni.as_dict(),
2384
            }
2385
        )
2386
2387 1
    async def test_add_metadata(self):
2388
        """Test method to add metadata"""
2389 1
        self.napp.controller.loop = asyncio.get_running_loop()
2390 1
        evc_mock = create_autospec(EVC)
2391 1
        evc_mock.metadata = {}
2392 1
        evc_mock.id = 1234
2393 1
        self.napp.circuits = {"1234": evc_mock}
2394
2395 1
        payload = {"metadata1": 1, "metadata2": 2}
2396 1
        response = await self.api_client.post(
2397
            f"{self.base_endpoint}/v2/evc/1234/metadata",
2398
            json=payload
2399
        )
2400
2401 1
        assert response.status_code == 201
2402 1
        evc_mock.extend_metadata.assert_called_with(payload)
2403
2404 1
    async def test_add_metadata_malformed_json(self):
2405
        """Test method to add metadata with a malformed json"""
2406 1
        self.napp.controller.loop = asyncio.get_running_loop()
2407 1
        payload = b'{"metadata1": 1, "metadata2": 2,}'
2408 1
        response = await self.api_client.post(
2409
            f"{self.base_endpoint}/v2/evc/1234/metadata",
2410
            content=payload,
2411
            headers={"Content-Type": "application/json"}
2412
        )
2413
2414
        assert response.status_code == 400
2415
        assert "body contains invalid API" in response.json()["description"]
2416
2417 1
    async def test_add_metadata_no_body(self):
2418
        """Test method to add metadata with no body"""
2419 1
        self.napp.controller.loop = asyncio.get_running_loop()
2420 1
        response = await self.api_client.post(
2421
            f"{self.base_endpoint}/v2/evc/1234/metadata"
2422
        )
2423
        assert response.status_code == 400
2424
        assert response.json()["description"] == \
2425
            "Missing required request body"
2426
2427 1
    async def test_add_metadata_no_evc(self):
2428
        """Test method to add metadata with no evc"""
2429 1
        self.napp.controller.loop = asyncio.get_running_loop()
2430 1
        payload = {"metadata1": 1, "metadata2": 2}
2431 1
        response = await self.api_client.post(
2432
            f"{self.base_endpoint}/v2/evc/1234/metadata",
2433
            json=payload,
2434
        )
2435
        assert response.status_code == 404
2436
        assert response.json()["description"] == \
2437
            "circuit_id 1234 not found."
2438
2439 1
    async def test_add_metadata_wrong_content_type(self):
2440
        """Test method to add metadata with wrong content type"""
2441 1
        self.napp.controller.loop = asyncio.get_running_loop()
2442 1
        payload = {"metadata1": 1, "metadata2": 2}
2443 1
        response = await self.api_client.post(
2444
            f"{self.base_endpoint}/v2/evc/1234/metadata",
2445
            data=payload,
2446
            headers={"Content-Type": "application/xml"}
2447
        )
2448
        assert response.status_code == 415
2449
        assert "application/xml" in response.json()["description"]
2450
2451 1
    async def test_get_metadata(self):
2452
        """Test method to get metadata"""
2453 1
        evc_mock = create_autospec(EVC)
2454 1
        evc_mock.metadata = {'metadata1': 1, 'metadata2': 2}
2455 1
        evc_mock.id = 1234
2456 1
        self.napp.circuits = {"1234": evc_mock}
2457
2458 1
        response = await self.api_client.get(
2459
            f"{self.base_endpoint}/v2/evc/1234/metadata",
2460
        )
2461 1
        assert response.status_code == 200
2462 1
        assert response.json() == {"metadata": evc_mock.metadata}
2463
2464 1
    async def test_delete_metadata(self):
2465
        """Test method to delete metadata"""
2466 1
        evc_mock = create_autospec(EVC)
2467 1
        evc_mock.metadata = {'metadata1': 1, 'metadata2': 2}
2468 1
        evc_mock.id = 1234
2469 1
        self.napp.circuits = {"1234": evc_mock}
2470
2471 1
        response = await self.api_client.delete(
2472
            f"{self.base_endpoint}/v2/evc/1234/metadata/metadata1",
2473
        )
2474 1
        assert response.status_code == 200
2475
2476 1
    async def test_delete_metadata_no_evc(self):
2477
        """Test method to delete metadata with no evc"""
2478 1
        response = await self.api_client.delete(
2479
            f"{self.base_endpoint}/v2/evc/1234/metadata/metadata1",
2480
        )
2481
        assert response.status_code == 404
2482
        assert response.json()["description"] == \
2483
            "circuit_id 1234 not found."
2484
2485 1
    @patch('napps.kytos.mef_eline.main.Main._load_evc')
2486 1
    def test_load_all_evcs(self, load_evc_mock):
2487
        """Test load_evcs method"""
2488 1
        mock_circuits = {
2489
            'circuits': {
2490
                1: 'circuit_1',
2491
                2: 'circuit_2',
2492
                3: 'circuit_3',
2493
                4: 'circuit_4'
2494
            }
2495
        }
2496 1
        self.napp.mongo_controller.get_circuits.return_value = mock_circuits
2497 1
        self.napp.circuits = {2: 'circuit_2', 3: 'circuit_3'}
2498 1
        self.napp.load_all_evcs()
2499 1
        load_evc_mock.assert_has_calls([call('circuit_1'), call('circuit_4')])
2500 1
        assert self.napp.controller.buffers.app.put.call_count > 1
2501 1
        call_args = self.napp.controller.buffers.app.put.call_args[0]
2502 1
        assert call_args[0].name == "kytos/mef_eline.evcs_loaded"
2503 1
        assert dict(call_args[0].content) == mock_circuits["circuits"]
2504 1
        timeout_d = {"timeout": 1}
2505 1
        assert self.napp.controller.buffers.app.put.call_args[1] == timeout_d
2506
2507 1
    @patch('napps.kytos.mef_eline.main.Main._evc_from_dict')
2508 1
    def test_load_evc(self, evc_from_dict_mock):
2509
        """Test _load_evc method"""
2510
        # pylint: disable=protected-access
2511
        # case 1: early return with ValueError exception
2512 1
        evc_from_dict_mock.side_effect = ValueError("err")
2513 1
        evc_dict = MagicMock()
2514 1
        assert not self.napp._load_evc(evc_dict)
2515
2516
        # case 2: early return with KytosTagError exception
2517 1
        evc_from_dict_mock.side_effect = KytosTagError("")
2518 1
        assert not self.napp._load_evc(evc_dict)
2519
2520
        # case 3: archived evc
2521 1
        evc = MagicMock()
2522 1
        evc.archived = True
2523 1
        evc_from_dict_mock.side_effect = None
2524 1
        evc_from_dict_mock.return_value = evc
2525 1
        assert not self.napp._load_evc(evc_dict)
2526
2527
        # case 4: success creating
2528 1
        evc.archived = False
2529 1
        evc.id = 1
2530 1
        self.napp.sched = MagicMock()
2531
2532 1
        result = self.napp._load_evc(evc_dict)
2533 1
        assert result == evc
2534 1
        self.napp.sched.add.assert_called_with(evc)
2535 1
        assert self.napp.circuits[1] == evc
2536
2537 1
    def test_handle_flow_mod_error(self):
2538
        """Test handle_flow_mod_error method"""
2539 1
        flow = MagicMock()
2540 1
        flow.cookie = 0xaa00000000000011
2541 1
        event = MagicMock()
2542 1
        event.content = {'flow': flow, 'error_command': 'add'}
2543 1
        evc = create_autospec(EVC)
2544 1
        evc.archived = False
2545 1
        evc.remove_current_flows = MagicMock()
2546 1
        evc.lock = MagicMock()
2547 1
        self.napp.circuits = {"00000000000011": evc}
2548 1
        self.napp.handle_flow_mod_error(event)
2549 1
        evc.remove_current_flows.assert_called_once()
2550
2551 1
    def test_handle_flow_mod_error_return(self):
2552
        """Test handle_flow_mod_error method with early return."""
2553 1
        flow = MagicMock()
2554 1
        flow.cookie = 0xaa00000000000011
2555 1
        event = MagicMock()
2556 1
        event.content = {'flow': flow, 'error_command': 'delete'}
2557
2558 1
        evc = create_autospec(EVC)
2559 1
        evc.archived = False
2560 1
        evc.is_enabled.return_value = True
2561
2562
        # Flow command is not 'add'
2563 1
        self.napp.circuits = {"00000000000011": evc}
2564 1
        self.napp.handle_flow_mod_error(event)
2565 1
        assert not evc.remove_current_flows.call_count
2566
2567
        # EVC is not enabled
2568 1
        event.content["error_command"] = "add"
2569 1
        evc.is_enabled.return_value = False
2570 1
        self.napp.handle_flow_mod_error(event)
2571 1
        assert not evc.remove_current_flows.call_count
2572
2573
        # EVC is archived
2574 1
        evc.is_enabled.return_value = True
2575 1
        evc.archived = True
2576 1
        self.napp.handle_flow_mod_error(event)
2577 1
        assert not evc.remove_current_flows.call_count
2578
2579
        # EVC does not exist in self.circuits
2580 1
        evc.archived = False
2581 1
        self.napp.circuits = {}
2582 1
        self.napp.handle_flow_mod_error(event)
2583 1
        assert not evc.remove_current_flows.call_count
2584
2585 1
    @patch("kytos.core.Controller.get_interface_by_id")
2586 1
    def test_uni_from_dict(self, _get_interface_by_id_mock):
2587
        """Test _uni_from_dict method."""
2588
        # pylint: disable=protected-access
2589
        # case1: early return on empty dict
2590 1
        assert not self.napp._uni_from_dict(None)
2591
2592
        # case2: invalid interface raises ValueError
2593 1
        _get_interface_by_id_mock.return_value = None
2594 1
        uni_dict = {
2595
            "interface_id": "00:01:1",
2596
            "tag": {"tag_type": 'vlan', "value": 81},
2597
        }
2598 1
        with pytest.raises(ValueError):
2599 1
            self.napp._uni_from_dict(uni_dict)
2600
2601
        # case3: success creation
2602 1
        uni_mock = get_uni_mocked(switch_id="00:01")
2603 1
        _get_interface_by_id_mock.return_value = uni_mock.interface
2604 1
        uni = self.napp._uni_from_dict(uni_dict)
2605 1
        assert uni == uni_mock
2606
2607
        # case4: success creation of tag list
2608 1
        uni_dict["tag"]["value"] = [[1, 10]]
2609 1
        uni = self.napp._uni_from_dict(uni_dict)
2610 1
        assert isinstance(uni.user_tag, TAGRange)
2611
2612
        # case5: success creation without tag
2613 1
        uni_mock.user_tag = None
2614 1
        del uni_dict["tag"]
2615 1
        uni = self.napp._uni_from_dict(uni_dict)
2616 1
        assert uni == uni_mock
2617
2618 1
    def test_handle_flow_delete(self):
2619
        """Test handle_flow_delete method"""
2620 1
        flow = MagicMock()
2621 1
        flow.cookie = 0xaa00000000000011
2622 1
        event = MagicMock()
2623 1
        event.content = {'flow': flow}
2624 1
        evc = create_autospec(EVC)
2625 1
        evc.set_flow_removed_at = MagicMock()
2626 1
        self.napp.circuits = {"00000000000011": evc}
2627 1
        self.napp.handle_flow_delete(event)
2628 1
        evc.set_flow_removed_at.assert_called_once()
2629
2630 1 View Code Duplication
    async def test_add_bulk_metadata(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
2631
        """Test add_bulk_metadata method"""
2632 1
        self.napp.controller.loop = asyncio.get_running_loop()
2633 1
        evc_mock = create_autospec(EVC)
2634 1
        evc_mock.id = 1234
2635 1
        self.napp.circuits = {"1234": evc_mock}
2636 1
        payload = {
2637
            "circuit_ids": ["1234"],
2638
            "metadata1": 1,
2639
            "metadata2": 2
2640
        }
2641 1
        response = await self.api_client.post(
2642
            f"{self.base_endpoint}/v2/evc/metadata",
2643
            json=payload
2644
        )
2645 1
        assert response.status_code == 201
2646 1
        args = self.napp.mongo_controller.update_evcs_metadata.call_args[0]
2647 1
        ids = payload.pop("circuit_ids")
2648 1
        assert args[0] == ids
2649 1
        assert args[1] == payload
2650 1
        assert args[2] == "add"
2651 1
        calls = self.napp.mongo_controller.update_evcs_metadata.call_count
2652 1
        assert calls == 1
2653 1
        evc_mock.extend_metadata.assert_called_with(payload)
2654
2655 1
    async def test_add_bulk_metadata_empty_list(self):
2656
        """Test add_bulk_metadata method empty list"""
2657 1
        self.napp.controller.loop = asyncio.get_running_loop()
2658 1
        evc_mock = create_autospec(EVC)
2659 1
        evc_mock.id = 1234
2660 1
        self.napp.circuits = {"1234": evc_mock}
2661 1
        payload = {
2662
            "circuit_ids": [],
2663
            "metadata1": 1,
2664
            "metadata2": 2
2665
        }
2666 1
        response = await self.api_client.post(
2667
            f"{self.base_endpoint}/v2/evc/metadata",
2668
            json=payload
2669
        )
2670
        assert response.status_code == 400
2671
        assert "invalid" in response.json()["description"]
2672
2673 1
    async def test_add_bulk_metadata_no_id(self):
2674
        """Test add_bulk_metadata with unknown evc id"""
2675 1
        self.napp.controller.loop = asyncio.get_running_loop()
2676 1
        evc_mock = create_autospec(EVC)
2677 1
        evc_mock.id = 1234
2678 1
        self.napp.circuits = {"1234": evc_mock}
2679 1
        payload = {
2680
            "circuit_ids": ["1234", "4567"]
2681
        }
2682 1
        response = await self.api_client.post(
2683
            f"{self.base_endpoint}/v2/evc/metadata",
2684
            json=payload
2685
        )
2686
        assert response.status_code == 404
2687
2688 1
    async def test_add_bulk_metadata_no_circuits(self):
2689
        """Test add_bulk_metadata without circuit_ids"""
2690 1
        self.napp.controller.loop = asyncio.get_running_loop()
2691 1
        evc_mock = create_autospec(EVC)
2692 1
        evc_mock.id = 1234
2693 1
        self.napp.circuits = {"1234": evc_mock}
2694 1
        payload = {
2695
            "metadata": "data"
2696
        }
2697 1
        response = await self.api_client.post(
2698
            f"{self.base_endpoint}/v2/evc/metadata",
2699
            json=payload
2700
        )
2701
        assert response.status_code == 400
2702
2703 1 View Code Duplication
    async def test_delete_bulk_metadata(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
2704
        """Test delete_metadata method"""
2705 1
        self.napp.controller.loop = asyncio.get_running_loop()
2706 1
        evc_mock = create_autospec(EVC)
2707 1
        evc_mock.id = 1234
2708 1
        self.napp.circuits = {"1234": evc_mock}
2709 1
        payload = {
2710
            "circuit_ids": ["1234"]
2711
        }
2712 1
        response = await self.api_client.request(
2713
            "DELETE",
2714
            f"{self.base_endpoint}/v2/evc/metadata/metadata1",
2715
            json=payload
2716
        )
2717 1
        assert response.status_code == 200
2718 1
        args = self.napp.mongo_controller.update_evcs_metadata.call_args[0]
2719 1
        assert args[0] == payload["circuit_ids"]
2720 1
        assert args[1] == {"metadata1": ""}
2721 1
        assert args[2] == "del"
2722 1
        calls = self.napp.mongo_controller.update_evcs_metadata.call_count
2723 1
        assert calls == 1
2724 1
        assert evc_mock.remove_metadata.call_count == 1
2725
2726 1
    async def test_delete_bulk_metadata_error(self):
2727
        """Test bulk_delete_metadata with ciruit erroring"""
2728 1
        self.napp.controller.loop = asyncio.get_running_loop()
2729 1
        evc_mock = create_autospec(EVC)
2730 1
        evcs = [evc_mock, evc_mock]
2731 1
        self.napp.circuits = dict(zip(["1", "2"], evcs))
2732 1
        payload = {"circuit_ids": ["1", "2", "3"]}
2733 1
        response = await self.api_client.request(
2734
            "DELETE",
2735
            f"{self.base_endpoint}/v2/evc/metadata/metadata1",
2736
            json=payload
2737
        )
2738
        assert response.status_code == 404, response.data
2739
        assert response.json()["description"] == ["3"]
2740
2741 1
    async def test_use_uni_tags(self):
2742
        """Test _use_uni_tags"""
2743 1
        self.napp.controller.loop = asyncio.get_running_loop()
2744 1
        evc_mock = create_autospec(EVC)
2745 1
        evc_mock.uni_a = "uni_a_mock"
2746 1
        evc_mock.uni_z = "uni_z_mock"
2747 1
        self.napp._use_uni_tags(evc_mock)
2748 1
        assert evc_mock._use_uni_vlan.call_count == 2
2749 1
        assert evc_mock._use_uni_vlan.call_args[0][0] == evc_mock.uni_z
2750
2751
        # One UNI tag is not available
2752 1
        evc_mock._use_uni_vlan.side_effect = [KytosTagError(""), None]
2753 1
        with pytest.raises(KytosTagError):
2754 1
            self.napp._use_uni_tags(evc_mock)
2755 1
        assert evc_mock._use_uni_vlan.call_count == 3
2756 1
        assert evc_mock.make_uni_vlan_available.call_count == 0
2757
2758 1
        evc_mock._use_uni_vlan.side_effect = [None, KytosTagError("")]
2759 1
        with pytest.raises(KytosTagError):
2760 1
            self.napp._use_uni_tags(evc_mock)
2761 1
        assert evc_mock._use_uni_vlan.call_count == 5
2762 1
        assert evc_mock.make_uni_vlan_available.call_count == 1
2763
2764 1
    def test_check_no_tag_duplication(self):
2765
        """Test _check_no_tag_duplication"""
2766 1
        evc = MagicMock()
2767 1
        evc.check_no_tag_duplicate = MagicMock()
2768 1
        evc.archived = False
2769 1
        evc.id = "1"
2770 1
        self.napp.circuits = {"1": evc}
2771 1
        evc_id = "2"
2772 1
        uni_a = get_uni_mocked(valid=True)
2773 1
        uni_z = get_uni_mocked(valid=True)
2774 1
        self.napp._check_no_tag_duplication(evc_id, uni_a, uni_z)
2775 1
        assert evc.check_no_tag_duplicate.call_count == 0
2776
2777 1
        uni_a.user_tag = None
2778 1
        uni_z.user_tag = None
2779 1
        self.napp._check_no_tag_duplication(evc_id, uni_a, uni_z)
2780 1
        assert evc.check_no_tag_duplicate.call_count == 2
2781
2782 1
        self.napp._check_no_tag_duplication(evc_id, uni_a, None)
2783 1
        assert evc.check_no_tag_duplicate.call_count == 3
2784
2785 1
        self.napp._check_no_tag_duplication(evc_id, None, None)
2786 1
        assert evc.check_no_tag_duplicate.call_count == 3
2787
2788 1
    @patch("napps.kytos.mef_eline.main.time")
2789 1
    @patch("napps.kytos.mef_eline.main.Main.handle_interface_link_up")
2790 1
    @patch("napps.kytos.mef_eline.main.Main.handle_interface_link_down")
2791 1
    def test_handle_on_interface_link_change(
2792
        self,
2793
        mock_down,
2794
        mock_up,
2795
        mock_time
2796
    ):
2797
        """Test handle_on_interface_link_change"""
2798 1
        mock_time.sleep.return_value = True
2799 1
        mock_intf = Mock()
2800 1
        mock_intf.id = "mock_intf"
2801
2802
        # Created/link_up/enabled
2803 1
        name = '.*.switch.interface.created'
2804 1
        content = {"interface": mock_intf}
2805 1
        event = KytosEvent(name=name, content=content)
2806 1
        self.napp.handle_on_interface_link_change(event)
2807 1
        assert mock_down.call_count == 0
2808 1
        assert mock_up.call_count == 1
2809
2810 1
        name = '.*.interface.enabled'
2811 1
        content = {"interface": mock_intf}
2812 1
        event = KytosEvent(name=name, content=content)
2813 1
        self.napp.handle_on_interface_link_change(event)
2814 1
        assert mock_down.call_count == 0
2815 1
        assert mock_up.call_count == 2
2816
2817
        # Deleted/link_down/disabled
2818 1
        name = '.*.switch.interface.deleted'
2819 1
        event = KytosEvent(name=name, content=content)
2820 1
        self.napp.handle_on_interface_link_change(event)
2821 1
        assert mock_down.call_count == 1
2822 1
        assert mock_up.call_count == 2
2823
2824 1
        name = '.*.interface.down'
2825 1
        event = KytosEvent(name=name, content=content)
2826 1
        self.napp.handle_on_interface_link_change(event)
2827 1
        assert mock_down.call_count == 2
2828 1
        assert mock_up.call_count == 2
2829
2830
        # Event delay
2831 1
        self.napp._intf_events[mock_intf.id]["last_acquired"] = "mock_time"
2832 1
        for _ in range(1, 6):
2833 1
            self.napp.handle_on_interface_link_change(event)
2834 1
        assert mock_down.call_count == 2
2835 1
        assert mock_up.call_count == 2
2836
2837 1
        self.napp._intf_events[mock_intf.id].pop("last_acquired")
2838 1
        self.napp.handle_on_interface_link_change(event)
2839 1
        assert mock_down.call_count == 3
2840 1
        assert mock_up.call_count == 2
2841
2842
        # Out of order event
2843 1
        event = KytosEvent(name=name, content=content)
2844 1
        self.napp._intf_events[mock_intf.id]["event"] = Mock(timestamp=now())
2845
2846 1
        self.napp.handle_on_interface_link_change(event)
2847 1
        assert mock_down.call_count == 3
2848
        assert mock_up.call_count == 2
2849