Passed
Pull Request — master (#553)
by Aldo
03:53
created

build.tests.unit.models.test_path   A

Complexity

Total Complexity 32

Size/Duplication

Total Lines 926
Duplicated Lines 4 %

Test Coverage

Coverage 99.69%

Importance

Changes 0
Metric Value
eloc 695
dl 37
loc 926
ccs 322
cts 323
cp 0.9969
rs 9.7449
c 0
b 0
f 0
wmc 32

29 Methods

Rating   Name   Duplication   Size   Complexity  
A TestPath.test_link_affected_by_interface_2() 0 10 1
A TestPath.test_link_affected_by_interface_1() 0 10 1
A TestPath.test_is_affected_by_link_1() 0 4 1
A TestPath.test_status_case_1() 0 4 1
A TestPath.test_status_case_2() 0 9 1
A TestPath.test_is_valid_with_loop() 0 17 2
A TestPath.test_status_case_3() 0 5 1
A TestPath.test_is_valid() 18 18 1
A TestPath.test_empty_is_valid() 0 4 1
A TestPath.test_compare_different_paths() 0 22 1
A TestPath.test_status_case_4() 0 9 1
A TestPath.test_status_case_7() 0 9 1
A TestPath.test_as_dict() 0 10 1
A TestPath.test_is_valid_invalid() 19 19 2
A TestPath.test_is_valid_diconnected() 0 14 2
A TestPath.test_compare_same_paths() 0 14 1
A TestPath.test_status_case_6() 0 9 1
A TestPath.test_status_case_5() 0 9 1
A TestPath.test_status_case_8() 0 10 1
A TestPath._mocked_httpx_get_status_case_4() 0 9 1
A TestDynamicPathManager.test_get_best_paths_error() 0 22 1
A TestPath.test_choose_vlans() 0 13 1
B TestDynamicPathManager.test_get_best_paths() 0 100 1
A TestDynamicPathManager.test_get_disjoint_paths_error() 0 33 1
B TestDynamicPathManager.test_get_disjoint_paths_simple_evc() 0 75 1
A TestDynamicPathManager.test_clear_path() 0 22 1
B TestDynamicPathManager.test_get_disjoint_paths() 0 346 1
A TestDynamicPathManager.test_create_path_invalid() 0 12 1
A TestDynamicPathManager.test_get_shared_components() 0 26 1

How to fix   Duplicated Code   

Duplicated Code

Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.

Common duplication problems, and corresponding solutions are:

1
"""Module to test the Path class."""
2 1
import sys
3 1
from unittest.mock import call, patch, Mock, MagicMock
4 1
import pytest
5 1
from napps.kytos.mef_eline import settings
6 1
from httpx import TimeoutException, ConnectError
7 1
from kytos.core.common import EntityStatus
8 1
from kytos.core.link import Link
9 1
from kytos.core.switch import Switch
10
11
# pylint: disable=wrong-import-position,ungrouped-imports,no-member
12
13 1
sys.path.insert(0, "/var/lib/kytos/napps/..")
14
# pylint: enable=wrong-import-position
15 1
from napps.kytos.mef_eline.exceptions import InvalidPath  # NOQA pycodestyle
16 1
from napps.kytos.mef_eline.models import (  # NOQA pycodestyle
17
    DynamicPathManager, Path)
18 1
from napps.kytos.mef_eline.tests.helpers import (  # NOQA pycodestyle
19
    MockResponse, get_link_mocked, id_to_interface_mock)
20
21
22 1
class TestPath():
23
    """Class to test path methods."""
24
25 1
    def test_is_affected_by_link_1(self):
26
        """Test method is affected by link."""
27 1
        path = Path()
28 1
        assert path.is_affected_by_link() is False
29
30 1
    def test_link_affected_by_interface_1(self):
31
        """Test method to get the link using an interface."""
32 1
        link1 = Mock()
33 1
        link1.endpoint_a = "a"
34 1
        link1.endpoint_b = "b"
35 1
        link2 = Mock()
36 1
        link2.endpoint_a = "c"
37 1
        link2.endpoint_b = "d"
38 1
        path = Path([link1, link2])
39 1
        assert path.link_affected_by_interface() is None
40
41 1
    def test_link_affected_by_interface_2(self):
42
        """Test method to get the link using an interface."""
43 1
        link1 = Mock()
44 1
        link1.endpoint_a = "a"
45 1
        link1.endpoint_b = "b"
46 1
        link2 = Mock()
47 1
        link2.endpoint_a = "c"
48 1
        link2.endpoint_b = "d"
49 1
        path = Path([link1, link2])
50 1
        assert path.link_affected_by_interface("a") == link1
51
52 1
    def test_status_case_1(self):
53
        """Test if empty link is DISABLED."""
54 1
        current_path = Path()
55 1
        assert current_path.status == EntityStatus.DISABLED
56
57 1
    def test_status_case_2(self):
58
        """Test if link status is DOWN."""
59 1
        link1 = get_link_mocked()
60 1
        link2 = get_link_mocked()
61 1
        link1.id = "def"
62 1
        link2.id = "abc"
63 1
        links = [link1, link2]
64 1
        current_path = Path(links)
65 1
        assert current_path.status == EntityStatus.DOWN
66
67 1
    def test_status_case_3(self):
68
        """Test if link status is DISABLED."""
69 1
        links = []
70 1
        current_path = Path(links)
71 1
        assert current_path.status == EntityStatus.DISABLED
72
73
    # This method will be used by the mock to replace httpx.get
74 1
    def _mocked_httpx_get_status_case_4(self):
75
        return MockResponse(
76
            {
77
                "links": {
78
                    "abc": {"active": True, "enabled": True},
79
                    "def": {"active": True, "enabled": True},
80
                }
81
            },
82
            200,
83
        )
84
85 1
    def test_status_case_4(self):
86
        """Test if link status is UP."""
87 1
        link1 = get_link_mocked(status=EntityStatus.UP)
88 1
        link2 = get_link_mocked(status=EntityStatus.UP)
89 1
        link1.id = "def"
90 1
        link2.id = "abc"
91 1
        links = [link1, link2]
92 1
        current_path = Path(links)
93 1
        assert current_path.status == EntityStatus.UP
94
95 1
    def test_status_case_5(self):
96
        """Test if link status is UP."""
97 1
        link1 = get_link_mocked(status=EntityStatus.UP)
98 1
        link2 = get_link_mocked(status=EntityStatus.DISABLED)
99 1
        link1.id = "def"
100 1
        link2.id = "abc"
101 1
        links = [link1, link2]
102 1
        current_path = Path(links)
103 1
        assert current_path.status == EntityStatus.DISABLED
104
105 1
    def test_status_case_6(self):
106
        """Test if link status is UP."""
107 1
        link1 = get_link_mocked(status=EntityStatus.DISABLED)
108 1
        link2 = get_link_mocked(status=EntityStatus.UP)
109 1
        link1.id = "def"
110 1
        link2.id = "abc"
111 1
        links = [link1, link2]
112 1
        current_path = Path(links)
113 1
        assert current_path.status == EntityStatus.DISABLED
114
115 1
    def test_status_case_7(self):
116
        """Test if link status is DOWN if has one link down."""
117 1
        link1 = get_link_mocked(status=EntityStatus.UP)
118 1
        link2 = get_link_mocked(status=EntityStatus.DOWN)
119 1
        link1.id = "def"
120 1
        link2.id = "abc"
121 1
        links = [link1, link2]
122 1
        current_path = Path(links)
123 1
        assert current_path.status == EntityStatus.DOWN
124
125 1
    def test_status_case_8(self):
126
        """Test if status is DOWN if has one endpoint link is mismatched."""
127 1
        link1 = get_link_mocked(status=EntityStatus.UP)
128 1
        link1.endpoint_a.link = None
129 1
        link2 = get_link_mocked(status=EntityStatus.UP)
130 1
        link1.id = "def"
131 1
        link2.id = "abc"
132 1
        links = [link1, link2]
133 1
        current_path = Path(links)
134 1
        assert current_path.status == EntityStatus.DOWN
135
136 1
    def test_compare_same_paths(self):
137
        """Test compare paths with same links."""
138 1
        links = [
139
            get_link_mocked(
140
                endpoint_a_port=9, endpoint_b_port=10, metadata={"s_vlan": 5}
141
            ),
142
            get_link_mocked(
143
                endpoint_a_port=11, endpoint_b_port=12, metadata={"s_vlan": 6}
144
            ),
145
        ]
146
147 1
        path_1 = Path(links)
148 1
        path_2 = Path(links)
149 1
        assert path_1 == path_2
150
151 1
    def test_compare_different_paths(self):
152
        """Test compare paths with different links."""
153 1
        links_1 = [
154
            get_link_mocked(
155
                endpoint_a_port=9, endpoint_b_port=10, metadata={"s_vlan": 5}
156
            ),
157
            get_link_mocked(
158
                endpoint_a_port=11, endpoint_b_port=12, metadata={"s_vlan": 6}
159
            ),
160
        ]
161 1
        links_2 = [
162
            get_link_mocked(
163
                endpoint_a_port=12, endpoint_b_port=11, metadata={"s_vlan": 5}
164
            ),
165
            get_link_mocked(
166
                endpoint_a_port=14, endpoint_b_port=16, metadata={"s_vlan": 11}
167
            ),
168
        ]
169
170 1
        path_1 = Path(links_1)
171 1
        path_2 = Path(links_2)
172 1
        assert path_1 != path_2
173
174 1
    def test_as_dict(self):
175
        """Test path as dict."""
176 1
        links = [
177
            get_link_mocked(link_dict={"id": 3}),
178
            get_link_mocked(link_dict={"id": 2}),
179
        ]
180
181 1
        current_path = Path(links)
182 1
        expected_dict = [{"id": 3}, {"id": 2}]
183 1
        assert expected_dict == current_path.as_dict()
184
185 1
    def test_empty_is_valid(self) -> None:
186
        """Test empty path is valid."""
187 1
        path = Path([])
188 1
        assert path.is_valid(MagicMock(), MagicMock(), False)
189
190 1 View Code Duplication
    def test_is_valid(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
191
        """Test is_valid method."""
192 1
        switch1 = Switch("00:00:00:00:00:00:00:01")
193 1
        switch2 = Switch("00:00:00:00:00:00:00:02")
194 1
        switch3 = Switch("00:00:00:00:00:00:00:03")
195 1
        switch4 = Switch("00:00:00:00:00:00:00:04")
196 1
        switch5 = Switch("00:00:00:00:00:00:00:05")
197 1
        switch6 = Switch("00:00:00:00:00:00:00:06")
198
        # Links connected
199 1
        links = [
200
            get_link_mocked(switch_a=switch5, switch_b=switch6),
201
            get_link_mocked(switch_a=switch4, switch_b=switch5),
202
            get_link_mocked(switch_a=switch3, switch_b=switch4),
203
            get_link_mocked(switch_a=switch2, switch_b=switch3),
204
            get_link_mocked(switch_a=switch1, switch_b=switch2),
205
        ]
206 1
        path = Path(links)
207 1
        assert path.is_valid(switch6, switch1) is True
208
209 1
    def test_is_valid_diconnected(self):
210
        """Test is_valid with disconnected path"""
211 1
        switch1 = Switch("00:00:00:00:00:00:00:01")
212 1
        switch2 = Switch("00:00:00:00:00:00:00:02")
213 1
        switch3 = Switch("00:00:00:00:00:00:00:03")
214 1
        switch4 = Switch("00:00:00:00:00:00:00:04")
215 1
        links = [
216
            get_link_mocked(switch_a=switch1, switch_b=switch2),
217
            get_link_mocked(switch_a=switch3, switch_b=switch4),
218
            get_link_mocked(switch_a=switch2, switch_b=switch4),
219
        ]
220 1
        path = Path(links)
221 1
        with pytest.raises(InvalidPath):
222 1
            path.is_valid(switch1, switch4)
223
224 1
    def test_is_valid_with_loop(self):
225
        """Test is_valid with a loop"""
226 1
        switch1 = Switch("00:00:00:00:00:00:00:01")
227 1
        switch2 = Switch("00:00:00:00:00:00:00:02")
228 1
        switch3 = Switch("00:00:00:00:00:00:00:03")
229 1
        switch4 = Switch("00:00:00:00:00:00:00:04")
230 1
        switch5 = Switch("00:00:00:00:00:00:00:05")
231 1
        links = [
232
            get_link_mocked(switch_a=switch1, switch_b=switch2),
233
            get_link_mocked(switch_a=switch2, switch_b=switch3),
234
            get_link_mocked(switch_a=switch3, switch_b=switch4),
235
            get_link_mocked(switch_a=switch2, switch_b=switch4),
236
            get_link_mocked(switch_a=switch2, switch_b=switch5),
237
        ]
238 1
        path = Path(links)
239 1
        with pytest.raises(InvalidPath):
240 1
            path.is_valid(switch1, switch5)
241
242 1 View Code Duplication
    def test_is_valid_invalid(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
243
        """Test is_valid when path is invalid
244
        UNI_Z is not connected"""
245 1
        switch1 = Switch("00:00:00:00:00:00:00:01")
246 1
        switch2 = Switch("00:00:00:00:00:00:00:02")
247 1
        switch3 = Switch("00:00:00:00:00:00:00:03")
248 1
        switch4 = Switch("00:00:00:00:00:00:00:04")
249 1
        switch5 = Switch("00:00:00:00:00:00:00:05")
250 1
        switch6 = Switch("00:00:00:00:00:00:00:06")
251 1
        links = [
252
            get_link_mocked(switch_a=switch5, switch_b=switch6),
253
            get_link_mocked(switch_a=switch4, switch_b=switch5),
254
            get_link_mocked(switch_a=switch3, switch_b=switch4),
255
            get_link_mocked(switch_a=switch2, switch_b=switch3),
256
            get_link_mocked(switch_a=switch1, switch_b=switch2),
257
        ]
258 1
        path = Path(links)
259 1
        with pytest.raises(InvalidPath):
260 1
            path.is_valid(switch3, switch6)
261
262 1
    def test_choose_vlans(self):
263
        """Test choose_vlans"""
264 1
        link1 = get_link_mocked()
265 1
        link2 = get_link_mocked()
266 1
        link1.id = "def"
267 1
        link2.id = "abc"
268 1
        links = [link1, link2]
269 1
        current_path = Path(links)
270 1
        current_path.choose_vlans(MagicMock())
271 1
        assert link1.remove_metadata.call_count == 1
272 1
        assert link1.add_metadata.call_count == 1
273 1
        assert link2.remove_metadata.call_count == 1
274 1
        assert link2.add_metadata.call_count == 1
275
276
277 1
class TestDynamicPathManager():
278
    """Tests for the DynamicPathManager class"""
279
280 1
    def test_clear_path(self):
281
        """Test _clear_path method"""
282 1
        path = [
283
            '00:00:00:00:00:00:00:01:1',
284
            '00:00:00:00:00:00:00:02:3',
285
            '00:00:00:00:00:00:00:02',
286
            '00:00:00:00:00:00:00:02:4',
287
            '00:00:00:00:00:00:00:03:2',
288
            '00:00:00:00:00:00:00:03',
289
            '00:00:00:00:00:00:00:03:1',
290
            '00:00:00:00:00:00:00:04:1'
291
        ]
292 1
        expected_path = [
293
            '00:00:00:00:00:00:00:01:1',
294
            '00:00:00:00:00:00:00:02:3',
295
            '00:00:00:00:00:00:00:02:4',
296
            '00:00:00:00:00:00:00:03:2',
297
            '00:00:00:00:00:00:00:03:1',
298
            '00:00:00:00:00:00:00:04:1'
299
        ]
300
        # pylint: disable=protected-access
301 1
        assert DynamicPathManager._clear_path(path) == expected_path
302
303 1
    def test_create_path_invalid(self):
304
        """Test create_path method"""
305 1
        path = [
306
            '00:00:00:00:00:00:00:01:1',
307
            '00:00:00:00:00:00:00:02:3',
308
            '00:00:00:00:00:00:00:02',
309
            '00:00:00:00:00:00:00:02:4',
310
            '00:00:00:00:00:00:00:03:2',
311
            '00:00:00:00:00:00:00:03',
312
            '00:00:00:00:00:00:00:03:1',
313
        ]
314 1
        assert DynamicPathManager.create_path(path) is None
315
316 1
    @patch("httpx.post")
317 1
    def test_get_best_paths(self, mock_httpx_post):
318
        """Test get_best_paths method."""
319 1
        controller = MagicMock()
320 1
        controller.get_interface_by_id.side_effect = id_to_interface_mock
321 1
        DynamicPathManager.set_controller(controller)
322
323 1
        paths1 = {
324
            "paths": [
325
                {
326
                    "cost": 5,
327
                    "hops": [
328
                        "00:00:00:00:00:00:00:01:1",
329
                        "00:00:00:00:00:00:00:01",
330
                        "00:00:00:00:00:00:00:01:2",
331
                        "00:00:00:00:00:00:00:02:2",
332
                        "00:00:00:00:00:00:00:02",
333
                        "00:00:00:00:00:00:00:02:1"
334
                        ]
335
                },
336
                {
337
                    "cost": 11,
338
                    "hops": [
339
                        "00:00:00:00:00:00:00:01:1",
340
                        "00:00:00:00:00:00:00:01",
341
                        "00:00:00:00:00:00:00:01:2",
342
                        "00:00:00:00:00:00:00:02:2",
343
                        "00:00:00:00:00:00:00:02",
344
                        "00:00:00:00:00:00:00:02:3",
345
                        "00:00:00:00:00:00:00:03:3",
346
                        "00:00:00:00:00:00:00:03",
347
                        "00:00:00:00:00:00:00:03:4",
348
                        "00:00:00:00:00:00:00:04:4",
349
                        "00:00:00:00:00:00:00:04",
350
                        "00:00:00:00:00:00:00:04:1"
351
                        ]
352
                },
353
            ]
354
        }
355
356 1
        expected_paths_0 = [
357
            Link(
358
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
359
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
360
            ),
361
        ]
362
363 1
        expected_paths_1 = [
364
            Link(
365
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
366
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
367
            ),
368
            Link(
369
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
370
                id_to_interface_mock("00:00:00:00:00:00:00:03:3")
371
            ),
372
            Link(
373
                id_to_interface_mock("00:00:00:00:00:00:00:03:4"),
374
                id_to_interface_mock("00:00:00:00:00:00:00:04:4")
375
            ),
376
        ]
377
378 1
        mock_response = MagicMock()
379 1
        mock_response.status_code = 200
380 1
        mock_response.json.return_value = paths1
381 1
        mock_httpx_post.return_value = mock_response
382 1
        kwargs = {
383
            "spf_attribute": settings.SPF_ATTRIBUTE,
384
            "spf_max_path_cost": 8,
385
            "mandatory_metrics": {
386
                "ownership": "red"
387
            }
388
        }
389 1
        circuit = MagicMock()
390 1
        circuit.uni_a.interface.id = "1"
391 1
        circuit.uni_z.interface.id = "2"
392 1
        max_paths = 2
393 1
        res_paths = list(DynamicPathManager.get_best_paths(circuit,
394
                         max_paths=max_paths, **kwargs))
395 1
        assert (
396
            [link.id for link in res_paths[0]] ==
397
            [link.id for link in expected_paths_0]
398
        )
399 1
        assert (
400
            [link.id for link in res_paths[1]] ==
401
            [link.id for link in expected_paths_1]
402
        )
403 1
        expected_call = call(
404
            "http://localhost:8181/api/kytos/pathfinder/v3/",
405
            json={
406
                **{
407
                    "source": circuit.uni_a.interface.id,
408
                    "destination": circuit.uni_z.interface.id,
409
                    "spf_max_paths": max_paths,
410
                },
411
                **kwargs
412
            },
413
            timeout=10,
414
        )
415 1
        mock_httpx_post.assert_has_calls([expected_call])
416
417 1
    @patch('time.sleep')
418 1
    @patch("napps.kytos.mef_eline.models.path.log")
419 1
    @patch("httpx.post")
420 1
    def test_get_best_paths_error(self, mock_httpx_post, mock_log, _):
421
        """Test get_best_paths method."""
422 1
        controller = MagicMock()
423 1
        DynamicPathManager.set_controller(controller)
424 1
        circuit = MagicMock()
425 1
        circuit.id = "id"
426 1
        circuit.uni_a.interface.id = "1"
427 1
        circuit.uni_z.interface.id = "2"
428 1
        circuit.name = "some_name"
429
430 1
        mock_response = MagicMock()
431 1
        mock_response.status_code = 400
432 1
        mock_response.json.return_value = {}
433 1
        mock_httpx_post.return_value = mock_response
434
435 1
        res_paths = list(DynamicPathManager.get_best_paths(circuit))
436 1
        assert not res_paths
437 1
        assert isinstance(res_paths, list)
438 1
        assert mock_log.error.call_count == 1
439
440
    # pylint: disable=too-many-statements, too-many-locals
441 1
    @patch.object(
442
        DynamicPathManager,
443
        "get_shared_components",
444
        side_effect=DynamicPathManager.get_shared_components
445
    )
446 1
    @patch("httpx.post")
447 1
    def test_get_disjoint_paths(self, mock_httpx_post, mock_shared):
448
        """Test get_disjoint_paths method."""
449
450 1
        controller = MagicMock()
451 1
        controller.get_interface_by_id.side_effect = id_to_interface_mock
452 1
        DynamicPathManager.set_controller(controller)
453
454 1
        evc = MagicMock()
455 1
        evc.secondary_constraints = {
456
            "spf_attribute": "hop",
457
            "spf_max_path_cost": 20,
458
            "mandatory_metrics": {
459
                "ownership": "red"
460
            }
461
        }
462 1
        evc.uni_a.interface.id = "1"
463 1
        evc.uni_z.interface.id = "2"
464 1
        evc.uni_a.interface.switch.id = "00:00:00:00:00:00:00:01"
465 1
        evc.uni_z.interface.switch.id = "00:00:00:00:00:00:00:05"
466
467
        # Topo0
468 1
        paths1 = {
469
            "paths": [
470
                {
471
                    "cost": 11,
472
                    "hops": [
473
                        "00:00:00:00:00:00:00:01:1",
474
                        "00:00:00:00:00:00:00:01",
475
                        "00:00:00:00:00:00:00:01:2",
476
                        "00:00:00:00:00:00:00:02:2",
477
                        "00:00:00:00:00:00:00:02",
478
                        "00:00:00:00:00:00:00:02:3",
479
                        "00:00:00:00:00:00:00:04:2",
480
                        "00:00:00:00:00:00:00:04",
481
                        "00:00:00:00:00:00:00:04:3",
482
                        "00:00:00:00:00:00:00:05:2",
483
                        "00:00:00:00:00:00:00:05",
484
                        "00:00:00:00:00:00:00:05:1"
485
                        ]
486
                },
487
                {
488
                    "cost": 11,
489
                    "hops": [
490
                        "00:00:00:00:00:00:00:01:1",
491
                        "00:00:00:00:00:00:00:01",
492
                        "00:00:00:00:00:00:00:01:3",
493
                        "00:00:00:00:00:00:00:03:2",
494
                        "00:00:00:00:00:00:00:03",
495
                        "00:00:00:00:00:00:00:03:3",
496
                        "00:00:00:00:00:00:00:04:4",
497
                        "00:00:00:00:00:00:00:04",
498
                        "00:00:00:00:00:00:00:04:3",
499
                        "00:00:00:00:00:00:00:05:2",
500
                        "00:00:00:00:00:00:00:05",
501
                        "00:00:00:00:00:00:00:05:1"
502
                        ]
503
                },
504
                {
505
                    "cost": 14,
506
                    "hops": [
507
                        "00:00:00:00:00:00:00:01:1",
508
                        "00:00:00:00:00:00:00:01",
509
                        "00:00:00:00:00:00:00:01:2",
510
                        "00:00:00:00:00:00:00:02:2",
511
                        "00:00:00:00:00:00:00:02",
512
                        "00:00:00:00:00:00:00:02:3",
513
                        "00:00:00:00:00:00:00:04:2",
514
                        "00:00:00:00:00:00:00:04",
515
                        "00:00:00:00:00:00:00:04:5",
516
                        "00:00:00:00:00:00:00:06:2",
517
                        "00:00:00:00:00:00:00:06",
518
                        "00:00:00:00:00:00:00:06:3",
519
                        "00:00:00:00:00:00:00:05:3",
520
                        "00:00:00:00:00:00:00:05",
521
                        "00:00:00:00:00:00:00:05:1"
522
                    ]
523
                },
524
                {
525
                    "cost": 14,
526
                    "hops": [
527
                        "00:00:00:00:00:00:00:01:1",
528
                        "00:00:00:00:00:00:00:01",
529
                        "00:00:00:00:00:00:00:01:3",
530
                        "00:00:00:00:00:00:00:03:2",
531
                        "00:00:00:00:00:00:00:03",
532
                        "00:00:00:00:00:00:00:03:3",
533
                        "00:00:00:00:00:00:00:04:4",
534
                        "00:00:00:00:00:00:00:04",
535
                        "00:00:00:00:00:00:00:04:5",
536
                        "00:00:00:00:00:00:00:06:2",
537
                        "00:00:00:00:00:00:00:06",
538
                        "00:00:00:00:00:00:00:06:3",
539
                        "00:00:00:00:00:00:00:05:3",
540
                        "00:00:00:00:00:00:00:05",
541
                        "00:00:00:00:00:00:00:05:1"
542
                    ]
543
                },
544
                {
545
                    "cost": 17,
546
                    "hops": [
547
                        "00:00:00:00:00:00:00:01:1",
548
                        "00:00:00:00:00:00:00:01",
549
                        "00:00:00:00:00:00:00:01:3",
550
                        "00:00:00:00:00:00:00:03:2",
551
                        "00:00:00:00:00:00:00:03",
552
                        "00:00:00:00:00:00:00:03:3",
553
                        "00:00:00:00:00:00:00:04:4",
554
                        "00:00:00:00:00:00:00:04",
555
                        "00:00:00:00:00:00:00:04:5",
556
                        "00:00:00:00:00:00:00:06:2",
557
                        "00:00:00:00:00:00:00:06",
558
                        "00:00:00:00:00:00:00:06:4",
559
                        "00:00:00:00:00:00:00:07:2",
560
                        "00:00:00:00:00:00:00:07",
561
                        "00:00:00:00:00:00:00:07:3",
562
                        "00:00:00:00:00:00:00:05:4",
563
                        "00:00:00:00:00:00:00:05",
564
                        "00:00:00:00:00:00:00:05:1"
565
                    ]
566
                },
567
            ]
568
        }
569
570 1
        mock_response = MagicMock()
571 1
        mock_response.status_code = 200
572 1
        mock_response.json.return_value = paths1
573
574
        # when we dont have the current_path
575 1
        mock_httpx_post.return_value = mock_response
576 1
        disjoint_paths = list(DynamicPathManager.get_disjoint_paths(evc, []))
577 1
        assert not disjoint_paths
578
579 1
        current_path = [
580
            Link(
581
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
582
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
583
            ),
584
            Link(
585
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
586
                id_to_interface_mock("00:00:00:00:00:00:00:04:2")
587
            ),
588
            Link(
589
                id_to_interface_mock("00:00:00:00:00:00:00:04:3"),
590
                id_to_interface_mock("00:00:00:00:00:00:00:05:2")
591
            ),
592
        ]
593 1
        path_links = [
594
            ("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:02:2"),
595
            ("00:00:00:00:00:00:00:02:3", "00:00:00:00:00:00:00:04:2"),
596
            ("00:00:00:00:00:00:00:04:3", "00:00:00:00:00:00:00:05:2")
597
        ]
598 1
        path_switches = {
599
            "00:00:00:00:00:00:00:04",
600
            "00:00:00:00:00:00:00:02"
601
        }
602
603
        # only one path available from pathfinder (precesilly the
604
        # current_path), so the maximum disjoint path will be empty
605 1
        mock_response.json.return_value = {"paths": paths1["paths"][0:1]}
606 1
        mock_httpx_post.return_value = mock_response
607 1
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
608 1
        args = mock_shared.call_args[0]
609 1
        assert args[0] == paths1["paths"][0]
610 1
        assert args[1] == path_links
611 1
        assert args[2] == path_switches
612 1
        assert len(paths) == 0
613
614 1
        expected_disjoint_path = [
615
            Link(
616
                id_to_interface_mock("00:00:00:00:00:00:00:01:3"),
617
                id_to_interface_mock("00:00:00:00:00:00:00:03:2")
618
            ),
619
            Link(
620
                id_to_interface_mock("00:00:00:00:00:00:00:03:3"),
621
                id_to_interface_mock("00:00:00:00:00:00:00:04:4")
622
            ),
623
            Link(
624
                id_to_interface_mock("00:00:00:00:00:00:00:04:5"),
625
                id_to_interface_mock("00:00:00:00:00:00:00:06:2")
626
            ),
627
            Link(
628
                id_to_interface_mock("00:00:00:00:00:00:00:06:3"),
629
                id_to_interface_mock("00:00:00:00:00:00:00:05:3")
630
            ),
631
        ]
632
633
        # there are one alternative path
634 1
        mock_response.json.return_value = paths1
635 1
        mock_httpx_post.return_value = mock_response
636 1
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
637 1
        args = mock_shared.call_args[0]
638 1
        assert args[0] == paths1["paths"][-1]
639 1
        assert args[1] == path_links
640 1
        assert args[2] == {
641
            "00:00:00:00:00:00:00:04",
642
            "00:00:00:00:00:00:00:02"
643
        }
644 1
        assert len(paths) == 4
645
        # for more information on the paths please refer to EP029
646 1
        assert len(paths[0]) == 4  # path S-Z-W-I-D
647 1
        assert len(paths[1]) == 5  # path S-Z-W-I-J-D
648 1
        assert len(paths[2]) == 3  # path S-Z-W-D
649 1
        assert len(paths[3]) == 4  # path S-X-W-I-D
650 1
        assert (
651
            [link.id for link in paths[0]] ==
652
            [link.id for link in expected_disjoint_path]
653
        )
654
655 1
        max_paths = 10
656 1
        expected_call = call(
657
            "http://localhost:8181/api/kytos/pathfinder/v3/",
658
            json={
659
                **{
660
                    "source": evc.uni_a.interface.id,
661
                    "destination": evc.uni_z.interface.id,
662
                    "spf_max_paths": max_paths,
663
                },
664
                **evc.secondary_constraints
665
            },
666
            timeout=10,
667
        )
668 1
        assert mock_httpx_post.call_count >= 1
669
        # If secondary_constraints are set they are expected to be parametrized
670 1
        mock_httpx_post.assert_has_calls([expected_call])
671
672
        # EP029 Topo2
673 1
        evc.uni_a.interface.switch.id = "00:00:00:00:00:00:00:01"
674 1
        evc.uni_z.interface.switch.id = "00:00:00:00:00:00:00:07"
675 1
        paths2 = {
676
            "paths": [
677
                {
678
                    "cost": 14,
679
                    "hops": [
680
                        "00:00:00:00:00:00:00:01:1",
681
                        "00:00:00:00:00:00:00:01",
682
                        "00:00:00:00:00:00:00:01:2",
683
                        "00:00:00:00:00:00:00:02:1",
684
                        "00:00:00:00:00:00:00:02",
685
                        "00:00:00:00:00:00:00:02:2",
686
                        "00:00:00:00:00:00:00:03:1",
687
                        "00:00:00:00:00:00:00:03",
688
                        "00:00:00:00:00:00:00:03:2",
689
                        "00:00:00:00:00:00:00:04:1",
690
                        "00:00:00:00:00:00:00:04",
691
                        "00:00:00:00:00:00:00:04:2",
692
                        "00:00:00:00:00:00:00:07:2",
693
                        "00:00:00:00:00:00:00:07",
694
                        "00:00:00:00:00:00:00:07:1"
695
                    ]
696
                },
697
                {
698
                    "cost": 17,
699
                    "hops": [
700
                        "00:00:00:00:00:00:00:01:1",
701
                        "00:00:00:00:00:00:00:01",
702
                        "00:00:00:00:00:00:00:01:2",
703
                        "00:00:00:00:00:00:00:02:1",
704
                        "00:00:00:00:00:00:00:02",
705
                        "00:00:00:00:00:00:00:02:3",
706
                        "00:00:00:00:00:00:00:05:1",
707
                        "00:00:00:00:00:00:00:05",
708
                        "00:00:00:00:00:00:00:05:2",
709
                        "00:00:00:00:00:00:00:06:1",
710
                        "00:00:00:00:00:00:00:06",
711
                        "00:00:00:00:00:00:00:06:2",
712
                        "00:00:00:00:00:00:00:04:3",
713
                        "00:00:00:00:00:00:00:04",
714
                        "00:00:00:00:00:00:00:04:2",
715
                        "00:00:00:00:00:00:00:07:2",
716
                        "00:00:00:00:00:00:00:07",
717
                        "00:00:00:00:00:00:00:07:1"
718
                    ]
719
                }
720
            ]
721
        }
722
723 1
        current_path = [
724
            Link(
725
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
726
                id_to_interface_mock("00:00:00:00:00:00:00:02:1")
727
            ),
728
            Link(
729
                id_to_interface_mock("00:00:00:00:00:00:00:02:2"),
730
                id_to_interface_mock("00:00:00:00:00:00:00:03:1")
731
            ),
732
            Link(
733
                id_to_interface_mock("00:00:00:00:00:00:00:03:2"),
734
                id_to_interface_mock("00:00:00:00:00:00:00:04:1")
735
            ),
736
            Link(
737
                id_to_interface_mock("00:00:00:00:00:00:00:04:2"),
738
                id_to_interface_mock("00:00:00:00:00:00:00:07:2")
739
            ),
740
        ]
741 1
        path_interfaces = [
742
            ("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:02:1"),
743
            ("00:00:00:00:00:00:00:02:2", "00:00:00:00:00:00:00:03:1"),
744
            ("00:00:00:00:00:00:00:03:2", "00:00:00:00:00:00:00:04:1"),
745
            ("00:00:00:00:00:00:00:04:2", "00:00:00:00:00:00:00:07:2")
746
        ]
747 1
        path_switches = {
748
            "00:00:00:00:00:00:00:02",
749
            "00:00:00:00:00:00:00:03",
750
            "00:00:00:00:00:00:00:04"
751
        }
752
753 1
        expected_disjoint_path = [
754
            Link(
755
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
756
                id_to_interface_mock("00:00:00:00:00:00:00:02:1")
757
            ),
758
            Link(
759
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
760
                id_to_interface_mock("00:00:00:00:00:00:00:05:1")
761
            ),
762
            Link(
763
                id_to_interface_mock("00:00:00:00:00:00:00:05:2"),
764
                id_to_interface_mock("00:00:00:00:00:00:00:06:1")
765
            ),
766
            Link(
767
                id_to_interface_mock("00:00:00:00:00:00:00:06:2"),
768
                id_to_interface_mock("00:00:00:00:00:00:00:04:3")
769
            ),
770
            Link(
771
                id_to_interface_mock("00:00:00:00:00:00:00:04:2"),
772
                id_to_interface_mock("00:00:00:00:00:00:00:07:2")
773
            ),
774
        ]
775
776 1
        mock_response.json.return_value = {"paths": paths2["paths"]}
777 1
        mock_httpx_post.return_value = mock_response
778 1
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
779 1
        args = mock_shared.call_args[0]
780 1
        assert args[0] == paths2["paths"][-1]
781 1
        assert args[1] == path_interfaces
782 1
        assert args[2] == path_switches
783 1
        assert len(paths) == 1
784 1
        assert (
785
            [link.id for link in paths[0]] ==
786
            [link.id for link in expected_disjoint_path]
787
        )
788
789 1
    @patch("httpx.post")
790 1
    def test_get_disjoint_paths_simple_evc(self, mock_httpx_post):
791
        """Test get_disjoint_paths method for simple EVCs."""
792 1
        controller = MagicMock()
793 1
        controller.get_interface_by_id.side_effect = id_to_interface_mock
794 1
        DynamicPathManager.set_controller(controller)
795
796 1
        evc = MagicMock()
797 1
        evc.secondary_constraints = {
798
            "spf_attribute": "hop",
799
            "spf_max_path_cost": 20,
800
            "mandatory_metrics": {
801
                "ownership": "red"
802
            }
803
        }
804 1
        evc.uni_a.interface.id = "1"
805 1
        evc.uni_z.interface.id = "2"
806 1
        evc.uni_a.interface.switch.id = "00:00:00:00:00:00:00:01"
807 1
        evc.uni_z.interface.switch.id = "00:00:00:00:00:00:00:05"
808
809 1
        mock_paths = {
810
            "paths": [
811
                {
812
                    "cost": 5,
813
                    "hops": [
814
                        "00:00:00:00:00:00:00:01:1",
815
                        "00:00:00:00:00:00:00:01",
816
                        "00:00:00:00:00:00:00:01:3",
817
                        "00:00:00:00:00:00:00:02:2",
818
                        "00:00:00:00:00:00:00:02",
819
                        "00:00:00:00:00:00:00:02:1"
820
                        ]
821
                },
822
                {
823
                    "cost": 8,
824
                    "hops": [
825
                        "00:00:00:00:00:00:00:01:1",
826
                        "00:00:00:00:00:00:00:01",
827
                        "00:00:00:00:00:00:00:01:4",
828
                        "00:00:00:00:00:00:00:03:3",
829
                        "00:00:00:00:00:00:00:03",
830
                        "00:00:00:00:00:00:00:03:2",
831
                        "00:00:00:00:00:00:00:02:3",
832
                        "00:00:00:00:00:00:00:02",
833
                        "00:00:00:00:00:00:00:02:1"
834
                        ]
835
                },
836
            ]
837
        }
838 1
        current_path = [
839
            Link(
840
                id_to_interface_mock("00:00:00:00:00:00:00:01:3"),
841
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
842
            ),
843
        ]
844 1
        expected_disjoint_path = [
845
            Link(
846
                id_to_interface_mock("00:00:00:00:00:00:00:01:4"),
847
                id_to_interface_mock("00:00:00:00:00:00:00:03:3")
848
            ),
849
            Link(
850
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
851
                id_to_interface_mock("00:00:00:00:00:00:00:03:2")
852
            ),
853
        ]
854
855 1
        mock_response = MagicMock()
856 1
        mock_response.status_code = 200
857 1
        mock_response.json.return_value = mock_paths
858 1
        mock_httpx_post.return_value = mock_response
859 1
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
860 1
        assert len(paths) == 1
861 1
        assert (
862
            [link.id for link in paths[0]] ==
863
            [link.id for link in expected_disjoint_path]
864
        )
865
866 1
    @patch("httpx.post")
867 1
    @patch("napps.kytos.mef_eline.models.path.log")
868 1
    @patch("time.sleep")
869 1
    def test_get_disjoint_paths_error(self, _, mock_log, mock_post):
870
        """Test get_disjoint_paths with reported errors. These are caught under
871
        httpx.RequestError."""
872 1
        mock_post.side_effect = TimeoutException('mock')
873 1
        unwanted_path = [
874
            Link(
875
                id_to_interface_mock("00:00:00:00:00:00:00:01:4"),
876
                id_to_interface_mock("00:00:00:00:00:00:00:03:3")
877
            ),
878
        ]
879 1
        evc = MagicMock()
880 1
        evc.secondary_constraints = {
881
            "spf_attribute": "hop",
882
            "spf_max_path_cost": 20,
883
            "mandatory_metrics": {
884
                "ownership": "red"
885
            }
886
        }
887 1
        evc.uni_a.interface.id = "1"
888 1
        evc.uni_z.interface.id = "2"
889 1
        evc.uni_a.interface.switch.id = "00:00:00:00:00:00:00:01"
890 1
        evc.uni_z.interface.switch.id = "00:00:00:00:00:00:00:05"
891 1
        path = DynamicPathManager.get_disjoint_paths(evc, unwanted_path)
892 1
        assert len(list(path)) == 0
893 1
        assert mock_log.error.call_count == 1
894
895 1
        mock_post.side_effect = ConnectError('mock')
896 1
        path = DynamicPathManager.get_disjoint_paths(evc, unwanted_path)
897 1
        assert len(list(path)) == 0
898 1
        assert mock_log.error.call_count == 2
899
900 1
    def test_get_shared_components(self):
901
        """Test get_shared_components"""
902 1
        mock_path = {"hops": [
903
            '00:00:00:00:00:00:00:01:1',
904
            '00:00:00:00:00:00:00:01',
905
            '00:00:00:00:00:00:00:01:4',
906
            '00:00:00:00:00:00:00:05:2',
907
            '00:00:00:00:00:00:00:05',
908
            '00:00:00:00:00:00:00:05:3',
909
            '00:00:00:00:00:00:00:02:4',
910
            '00:00:00:00:00:00:00:02',
911
            '00:00:00:00:00:00:00:02:3',
912
            '00:00:00:00:00:00:00:03:2',
913
            '00:00:00:00:00:00:00:03',
914
            '00:00:00:00:00:00:00:03:1'
915
        ]}
916 1
        mock_links = [
917
            ("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:02:2"),
918
            ("00:00:00:00:00:00:00:02:3", "00:00:00:00:00:00:00:03:2")
919
        ]
920 1
        mock_switches = {"00:00:00:00:00:00:00:02"}
921 1
        actual_lk, actual_sw = DynamicPathManager.get_shared_components(
922
            mock_path, mock_links, mock_switches
923
        )
924 1
        assert actual_lk == 1
925
        assert actual_sw == 1
926