Passed
Pull Request — master (#617)
by Aldo
04:18
created

build.tests.unit.models.test_path   A

Complexity

Total Complexity 32

Size/Duplication

Total Lines 941
Duplicated Lines 3.93 %

Test Coverage

Coverage 99.68%

Importance

Changes 0
Metric Value
eloc 705
dl 37
loc 941
ccs 313
cts 314
cp 0.9968
rs 9.735
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 TestDynamicPathManager.test_clear_path() 0 22 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 TestDynamicPathManager.test_valid_path_invalid() 0 28 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
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
263 1
class TestDynamicPathManager():
264
    """Tests for the DynamicPathManager class"""
265
266 1
    def test_clear_path(self):
267
        """Test _clear_path method"""
268 1
        path = [
269
            '00:00:00:00:00:00:00:01:1',
270
            '00:00:00:00:00:00:00:02:3',
271
            '00:00:00:00:00:00:00:02',
272
            '00:00:00:00:00:00:00:02:4',
273
            '00:00:00:00:00:00:00:03:2',
274
            '00:00:00:00:00:00:00:03',
275
            '00:00:00:00:00:00:00:03:1',
276
            '00:00:00:00:00:00:00:04:1'
277
        ]
278 1
        expected_path = [
279
            '00:00:00:00:00:00:00:01:1',
280
            '00:00:00:00:00:00:00:02:3',
281
            '00:00:00:00:00:00:00:02:4',
282
            '00:00:00:00:00:00:00:03:2',
283
            '00:00:00:00:00:00:00:03:1',
284
            '00:00:00:00:00:00:00:04:1'
285
        ]
286
        # pylint: disable=protected-access
287 1
        assert DynamicPathManager._clear_path(path) == expected_path
288
289 1
    @pytest.mark.parametrize(
290
        "invalid_path",
291
        [
292
            [
293
                "00:00:00:00:00:00:00:02:3",
294
                "00:00:00:00:00:00:00:03:2",
295
                "00:00:00:00:00:00:00:03:3",
296
                "00:00:00:00:00:00:00:04:2"
297
            ],
298
            [
299
                "00:00:00:00:00:00:00:01:2",
300
                "00:00:00:00:00:00:00:02:2",
301
                "00:00:00:00:00:00:00:02:3",
302
                "00:00:00:00:00:00:00:03:2",
303
                "00:00:00:00:00:00:00:03:5",
304
            ],
305
            [
306
                "00:00:00:00:00:00:00:02:2",
307
                "00:00:00:00:00:00:00:02:3",
308
                "00:00:00:00:00:00:00:03:2",
309
                "00:00:00:00:00:00:00:03:3",
310
                "00:00:00:00:00:00:00:04:2"
311
            ],
312
        ]
313
    )
314 1
    def test_valid_path_invalid(self, invalid_path):
315
        """Test _valid_path for invalid paths."""
316 1
        assert DynamicPathManager._valid_path(invalid_path) is False
317
318 1
    def test_create_path_invalid(self):
319
        """Test create_path method"""
320 1
        path = [
321
            '00:00:00:00:00:00:00:01:1',
322
            '00:00:00:00:00:00:00:02:3',
323
            '00:00:00:00:00:00:00:02',
324
            '00:00:00:00:00:00:00:02:4',
325
            '00:00:00:00:00:00:00:03:2',
326
            '00:00:00:00:00:00:00:03',
327
            '00:00:00:00:00:00:00:03:1',
328
        ]
329 1
        assert DynamicPathManager.create_path(path) is None
330
331 1
    @patch("httpx.post")
332 1
    def test_get_best_paths(self, mock_httpx_post):
333
        """Test get_best_paths method."""
334 1
        controller = MagicMock()
335 1
        controller.get_interface_by_id.side_effect = id_to_interface_mock
336 1
        DynamicPathManager.set_controller(controller)
337
338 1
        paths1 = {
339
            "paths": [
340
                {
341
                    "cost": 5,
342
                    "hops": [
343
                        "00:00:00:00:00:00:00:01:1",
344
                        "00:00:00:00:00:00:00:01",
345
                        "00:00:00:00:00:00:00:01:2",
346
                        "00:00:00:00:00:00:00:02:2",
347
                        "00:00:00:00:00:00:00:02",
348
                        "00:00:00:00:00:00:00:02:1"
349
                        ]
350
                },
351
                {
352
                    "cost": 11,
353
                    "hops": [
354
                        "00:00:00:00:00:00:00:01:1",
355
                        "00:00:00:00:00:00:00:01",
356
                        "00:00:00:00:00:00:00:01:2",
357
                        "00:00:00:00:00:00:00:02:2",
358
                        "00:00:00:00:00:00:00:02",
359
                        "00:00:00:00:00:00:00:02:3",
360
                        "00:00:00:00:00:00:00:03:3",
361
                        "00:00:00:00:00:00:00:03",
362
                        "00:00:00:00:00:00:00:03:4",
363
                        "00:00:00:00:00:00:00:04:4",
364
                        "00:00:00:00:00:00:00:04",
365
                        "00:00:00:00:00:00:00:04:1"
366
                        ]
367
                },
368
            ]
369
        }
370
371 1
        expected_paths_0 = [
372
            Link(
373
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
374
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
375
            ),
376
        ]
377
378 1
        expected_paths_1 = [
379
            Link(
380
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
381
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
382
            ),
383
            Link(
384
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
385
                id_to_interface_mock("00:00:00:00:00:00:00:03:3")
386
            ),
387
            Link(
388
                id_to_interface_mock("00:00:00:00:00:00:00:03:4"),
389
                id_to_interface_mock("00:00:00:00:00:00:00:04:4")
390
            ),
391
        ]
392
393 1
        mock_response = MagicMock()
394 1
        mock_response.status_code = 200
395 1
        mock_response.json.return_value = paths1
396 1
        mock_httpx_post.return_value = mock_response
397 1
        kwargs = {
398
            "spf_attribute": settings.SPF_ATTRIBUTE,
399
            "spf_max_path_cost": 8,
400
            "mandatory_metrics": {
401
                "ownership": "red"
402
            }
403
        }
404 1
        circuit = MagicMock()
405 1
        circuit.uni_a.interface.id = "1"
406 1
        circuit.uni_z.interface.id = "2"
407 1
        max_paths = 2
408 1
        res_paths = list(DynamicPathManager.get_best_paths(circuit,
409
                         max_paths=max_paths, **kwargs))
410 1
        assert (
411
            [link.id for link in res_paths[0]] ==
412
            [link.id for link in expected_paths_0]
413
        )
414 1
        assert (
415
            [link.id for link in res_paths[1]] ==
416
            [link.id for link in expected_paths_1]
417
        )
418 1
        expected_call = call(
419
            "http://localhost:8181/api/kytos/pathfinder/v3/",
420
            json={
421
                **{
422
                    "source": circuit.uni_a.interface.id,
423
                    "destination": circuit.uni_z.interface.id,
424
                    "spf_max_paths": max_paths,
425
                },
426
                **kwargs
427
            },
428
            timeout=10,
429
        )
430 1
        mock_httpx_post.assert_has_calls([expected_call])
431
432 1
    @patch('time.sleep')
433 1
    @patch("napps.kytos.mef_eline.models.path.log")
434 1
    @patch("httpx.post")
435 1
    def test_get_best_paths_error(self, mock_httpx_post, mock_log, _):
436
        """Test get_best_paths method."""
437 1
        controller = MagicMock()
438 1
        DynamicPathManager.set_controller(controller)
439 1
        circuit = MagicMock()
440 1
        circuit.id = "id"
441 1
        circuit.uni_a.interface.id = "1"
442 1
        circuit.uni_z.interface.id = "2"
443 1
        circuit.name = "some_name"
444
445 1
        mock_response = MagicMock()
446 1
        mock_response.status_code = 400
447 1
        mock_response.json.return_value = {}
448 1
        mock_httpx_post.return_value = mock_response
449
450 1
        res_paths = list(DynamicPathManager.get_best_paths(circuit))
451 1
        assert not res_paths
452 1
        assert isinstance(res_paths, list)
453 1
        assert mock_log.error.call_count == 1
454
455
    # pylint: disable=too-many-statements, too-many-locals
456 1
    @patch.object(
457
        DynamicPathManager,
458
        "get_shared_components",
459
        side_effect=DynamicPathManager.get_shared_components
460
    )
461 1
    @patch("httpx.post")
462 1
    def test_get_disjoint_paths(self, mock_httpx_post, mock_shared):
463
        """Test get_disjoint_paths method."""
464
465 1
        controller = MagicMock()
466 1
        controller.get_interface_by_id.side_effect = id_to_interface_mock
467 1
        DynamicPathManager.set_controller(controller)
468
469 1
        evc = MagicMock()
470 1
        evc.secondary_constraints = {
471
            "spf_attribute": "hop",
472
            "spf_max_path_cost": 20,
473
            "mandatory_metrics": {
474
                "ownership": "red"
475
            }
476
        }
477 1
        evc.uni_a.interface.id = "1"
478 1
        evc.uni_z.interface.id = "2"
479 1
        evc.uni_a.interface.switch.id = "00:00:00:00:00:00:00:01"
480 1
        evc.uni_z.interface.switch.id = "00:00:00:00:00:00:00:05"
481
482
        # Topo0
483 1
        paths1 = {
484
            "paths": [
485
                {
486
                    "cost": 11,
487
                    "hops": [
488
                        "00:00:00:00:00:00:00:01:1",
489
                        "00:00:00:00:00:00:00:01",
490
                        "00:00:00:00:00:00:00:01:2",
491
                        "00:00:00:00:00:00:00:02:2",
492
                        "00:00:00:00:00:00:00:02",
493
                        "00:00:00:00:00:00:00:02:3",
494
                        "00:00:00:00:00:00:00:04:2",
495
                        "00:00:00:00:00:00:00:04",
496
                        "00:00:00:00:00:00:00:04:3",
497
                        "00:00:00:00:00:00:00:05:2",
498
                        "00:00:00:00:00:00:00:05",
499
                        "00:00:00:00:00:00:00:05:1"
500
                        ]
501
                },
502
                {
503
                    "cost": 11,
504
                    "hops": [
505
                        "00:00:00:00:00:00:00:01:1",
506
                        "00:00:00:00:00:00:00:01",
507
                        "00:00:00:00:00:00:00:01:3",
508
                        "00:00:00:00:00:00:00:03:2",
509
                        "00:00:00:00:00:00:00:03",
510
                        "00:00:00:00:00:00:00:03:3",
511
                        "00:00:00:00:00:00:00:04:4",
512
                        "00:00:00:00:00:00:00:04",
513
                        "00:00:00:00:00:00:00:04:3",
514
                        "00:00:00:00:00:00:00:05:2",
515
                        "00:00:00:00:00:00:00:05",
516
                        "00:00:00:00:00:00:00:05:1"
517
                        ]
518
                },
519
                {
520
                    "cost": 14,
521
                    "hops": [
522
                        "00:00:00:00:00:00:00:01:1",
523
                        "00:00:00:00:00:00:00:01",
524
                        "00:00:00:00:00:00:00:01:2",
525
                        "00:00:00:00:00:00:00:02:2",
526
                        "00:00:00:00:00:00:00:02",
527
                        "00:00:00:00:00:00:00:02:3",
528
                        "00:00:00:00:00:00:00:04:2",
529
                        "00:00:00:00:00:00:00:04",
530
                        "00:00:00:00:00:00:00:04:5",
531
                        "00:00:00:00:00:00:00:06:2",
532
                        "00:00:00:00:00:00:00:06",
533
                        "00:00:00:00:00:00:00:06:3",
534
                        "00:00:00:00:00:00:00:05:3",
535
                        "00:00:00:00:00:00:00:05",
536
                        "00:00:00:00:00:00:00:05:1"
537
                    ]
538
                },
539
                {
540
                    "cost": 14,
541
                    "hops": [
542
                        "00:00:00:00:00:00:00:01:1",
543
                        "00:00:00:00:00:00:00:01",
544
                        "00:00:00:00:00:00:00:01:3",
545
                        "00:00:00:00:00:00:00:03:2",
546
                        "00:00:00:00:00:00:00:03",
547
                        "00:00:00:00:00:00:00:03:3",
548
                        "00:00:00:00:00:00:00:04:4",
549
                        "00:00:00:00:00:00:00:04",
550
                        "00:00:00:00:00:00:00:04:5",
551
                        "00:00:00:00:00:00:00:06:2",
552
                        "00:00:00:00:00:00:00:06",
553
                        "00:00:00:00:00:00:00:06:3",
554
                        "00:00:00:00:00:00:00:05:3",
555
                        "00:00:00:00:00:00:00:05",
556
                        "00:00:00:00:00:00:00:05:1"
557
                    ]
558
                },
559
                {
560
                    "cost": 17,
561
                    "hops": [
562
                        "00:00:00:00:00:00:00:01:1",
563
                        "00:00:00:00:00:00:00:01",
564
                        "00:00:00:00:00:00:00:01:3",
565
                        "00:00:00:00:00:00:00:03:2",
566
                        "00:00:00:00:00:00:00:03",
567
                        "00:00:00:00:00:00:00:03:3",
568
                        "00:00:00:00:00:00:00:04:4",
569
                        "00:00:00:00:00:00:00:04",
570
                        "00:00:00:00:00:00:00:04:5",
571
                        "00:00:00:00:00:00:00:06:2",
572
                        "00:00:00:00:00:00:00:06",
573
                        "00:00:00:00:00:00:00:06:4",
574
                        "00:00:00:00:00:00:00:07:2",
575
                        "00:00:00:00:00:00:00:07",
576
                        "00:00:00:00:00:00:00:07:3",
577
                        "00:00:00:00:00:00:00:05:4",
578
                        "00:00:00:00:00:00:00:05",
579
                        "00:00:00:00:00:00:00:05:1"
580
                    ]
581
                },
582
            ]
583
        }
584
585 1
        mock_response = MagicMock()
586 1
        mock_response.status_code = 200
587 1
        mock_response.json.return_value = paths1
588
589
        # when we dont have the current_path
590 1
        mock_httpx_post.return_value = mock_response
591 1
        disjoint_paths = list(DynamicPathManager.get_disjoint_paths(evc, []))
592 1
        assert not disjoint_paths
593
594 1
        current_path = [
595
            Link(
596
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
597
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
598
            ),
599
            Link(
600
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
601
                id_to_interface_mock("00:00:00:00:00:00:00:04:2")
602
            ),
603
            Link(
604
                id_to_interface_mock("00:00:00:00:00:00:00:04:3"),
605
                id_to_interface_mock("00:00:00:00:00:00:00:05:2")
606
            ),
607
        ]
608 1
        path_links = [
609
            ("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:02:2"),
610
            ("00:00:00:00:00:00:00:02:3", "00:00:00:00:00:00:00:04:2"),
611
            ("00:00:00:00:00:00:00:04:3", "00:00:00:00:00:00:00:05:2")
612
        ]
613 1
        path_switches = {
614
            "00:00:00:00:00:00:00:04",
615
            "00:00:00:00:00:00:00:02"
616
        }
617
618
        # only one path available from pathfinder (precesilly the
619
        # current_path), so the maximum disjoint path will be empty
620 1
        mock_response.json.return_value = {"paths": paths1["paths"][0:1]}
621 1
        mock_httpx_post.return_value = mock_response
622 1
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
623 1
        args = mock_shared.call_args[0]
624 1
        assert args[0] == paths1["paths"][0]
625 1
        assert args[1] == path_links
626 1
        assert args[2] == path_switches
627 1
        assert len(paths) == 0
628
629 1
        expected_disjoint_path = [
630
            Link(
631
                id_to_interface_mock("00:00:00:00:00:00:00:01:3"),
632
                id_to_interface_mock("00:00:00:00:00:00:00:03:2")
633
            ),
634
            Link(
635
                id_to_interface_mock("00:00:00:00:00:00:00:03:3"),
636
                id_to_interface_mock("00:00:00:00:00:00:00:04:4")
637
            ),
638
            Link(
639
                id_to_interface_mock("00:00:00:00:00:00:00:04:5"),
640
                id_to_interface_mock("00:00:00:00:00:00:00:06:2")
641
            ),
642
            Link(
643
                id_to_interface_mock("00:00:00:00:00:00:00:06:3"),
644
                id_to_interface_mock("00:00:00:00:00:00:00:05:3")
645
            ),
646
        ]
647
648
        # there are one alternative path
649 1
        mock_response.json.return_value = paths1
650 1
        mock_httpx_post.return_value = mock_response
651 1
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
652 1
        args = mock_shared.call_args[0]
653 1
        assert args[0] == paths1["paths"][-1]
654 1
        assert args[1] == path_links
655 1
        assert args[2] == {
656
            "00:00:00:00:00:00:00:04",
657
            "00:00:00:00:00:00:00:02"
658
        }
659 1
        assert len(paths) == 4
660
        # for more information on the paths please refer to EP029
661 1
        assert len(paths[0]) == 4  # path S-Z-W-I-D
662 1
        assert len(paths[1]) == 5  # path S-Z-W-I-J-D
663 1
        assert len(paths[2]) == 3  # path S-Z-W-D
664 1
        assert len(paths[3]) == 4  # path S-X-W-I-D
665 1
        assert (
666
            [link.id for link in paths[0]] ==
667
            [link.id for link in expected_disjoint_path]
668
        )
669
670 1
        max_paths = 10
671 1
        expected_call = call(
672
            "http://localhost:8181/api/kytos/pathfinder/v3/",
673
            json={
674
                **{
675
                    "source": evc.uni_a.interface.id,
676
                    "destination": evc.uni_z.interface.id,
677
                    "spf_max_paths": max_paths,
678
                },
679
                **evc.secondary_constraints
680
            },
681
            timeout=10,
682
        )
683 1
        assert mock_httpx_post.call_count >= 1
684
        # If secondary_constraints are set they are expected to be parametrized
685 1
        mock_httpx_post.assert_has_calls([expected_call])
686
687
        # EP029 Topo2
688 1
        evc.uni_a.interface.switch.id = "00:00:00:00:00:00:00:01"
689 1
        evc.uni_z.interface.switch.id = "00:00:00:00:00:00:00:07"
690 1
        paths2 = {
691
            "paths": [
692
                {
693
                    "cost": 14,
694
                    "hops": [
695
                        "00:00:00:00:00:00:00:01:1",
696
                        "00:00:00:00:00:00:00:01",
697
                        "00:00:00:00:00:00:00:01:2",
698
                        "00:00:00:00:00:00:00:02:1",
699
                        "00:00:00:00:00:00:00:02",
700
                        "00:00:00:00:00:00:00:02:2",
701
                        "00:00:00:00:00:00:00:03:1",
702
                        "00:00:00:00:00:00:00:03",
703
                        "00:00:00:00:00:00:00:03:2",
704
                        "00:00:00:00:00:00:00:04:1",
705
                        "00:00:00:00:00:00:00:04",
706
                        "00:00:00:00:00:00:00:04:2",
707
                        "00:00:00:00:00:00:00:07:2",
708
                        "00:00:00:00:00:00:00:07",
709
                        "00:00:00:00:00:00:00:07:1"
710
                    ]
711
                },
712
                {
713
                    "cost": 17,
714
                    "hops": [
715
                        "00:00:00:00:00:00:00:01:1",
716
                        "00:00:00:00:00:00:00:01",
717
                        "00:00:00:00:00:00:00:01:2",
718
                        "00:00:00:00:00:00:00:02:1",
719
                        "00:00:00:00:00:00:00:02",
720
                        "00:00:00:00:00:00:00:02:3",
721
                        "00:00:00:00:00:00:00:05:1",
722
                        "00:00:00:00:00:00:00:05",
723
                        "00:00:00:00:00:00:00:05:2",
724
                        "00:00:00:00:00:00:00:06:1",
725
                        "00:00:00:00:00:00:00:06",
726
                        "00:00:00:00:00:00:00:06:2",
727
                        "00:00:00:00:00:00:00:04:3",
728
                        "00:00:00:00:00:00:00:04",
729
                        "00:00:00:00:00:00:00:04:2",
730
                        "00:00:00:00:00:00:00:07:2",
731
                        "00:00:00:00:00:00:00:07",
732
                        "00:00:00:00:00:00:00:07:1"
733
                    ]
734
                }
735
            ]
736
        }
737
738 1
        current_path = [
739
            Link(
740
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
741
                id_to_interface_mock("00:00:00:00:00:00:00:02:1")
742
            ),
743
            Link(
744
                id_to_interface_mock("00:00:00:00:00:00:00:02:2"),
745
                id_to_interface_mock("00:00:00:00:00:00:00:03:1")
746
            ),
747
            Link(
748
                id_to_interface_mock("00:00:00:00:00:00:00:03:2"),
749
                id_to_interface_mock("00:00:00:00:00:00:00:04:1")
750
            ),
751
            Link(
752
                id_to_interface_mock("00:00:00:00:00:00:00:04:2"),
753
                id_to_interface_mock("00:00:00:00:00:00:00:07:2")
754
            ),
755
        ]
756 1
        path_interfaces = [
757
            ("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:02:1"),
758
            ("00:00:00:00:00:00:00:02:2", "00:00:00:00:00:00:00:03:1"),
759
            ("00:00:00:00:00:00:00:03:2", "00:00:00:00:00:00:00:04:1"),
760
            ("00:00:00:00:00:00:00:04:2", "00:00:00:00:00:00:00:07:2")
761
        ]
762 1
        path_switches = {
763
            "00:00:00:00:00:00:00:02",
764
            "00:00:00:00:00:00:00:03",
765
            "00:00:00:00:00:00:00:04"
766
        }
767
768 1
        expected_disjoint_path = [
769
            Link(
770
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
771
                id_to_interface_mock("00:00:00:00:00:00:00:02:1")
772
            ),
773
            Link(
774
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
775
                id_to_interface_mock("00:00:00:00:00:00:00:05:1")
776
            ),
777
            Link(
778
                id_to_interface_mock("00:00:00:00:00:00:00:05:2"),
779
                id_to_interface_mock("00:00:00:00:00:00:00:06:1")
780
            ),
781
            Link(
782
                id_to_interface_mock("00:00:00:00:00:00:00:06:2"),
783
                id_to_interface_mock("00:00:00:00:00:00:00:04:3")
784
            ),
785
            Link(
786
                id_to_interface_mock("00:00:00:00:00:00:00:04:2"),
787
                id_to_interface_mock("00:00:00:00:00:00:00:07:2")
788
            ),
789
        ]
790
791 1
        mock_response.json.return_value = {"paths": paths2["paths"]}
792 1
        mock_httpx_post.return_value = mock_response
793 1
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
794 1
        args = mock_shared.call_args[0]
795 1
        assert args[0] == paths2["paths"][-1]
796 1
        assert args[1] == path_interfaces
797 1
        assert args[2] == path_switches
798 1
        assert len(paths) == 1
799 1
        assert (
800
            [link.id for link in paths[0]] ==
801
            [link.id for link in expected_disjoint_path]
802
        )
803
804 1
    @patch("httpx.post")
805 1
    def test_get_disjoint_paths_simple_evc(self, mock_httpx_post):
806
        """Test get_disjoint_paths method for simple EVCs."""
807 1
        controller = MagicMock()
808 1
        controller.get_interface_by_id.side_effect = id_to_interface_mock
809 1
        DynamicPathManager.set_controller(controller)
810
811 1
        evc = MagicMock()
812 1
        evc.secondary_constraints = {
813
            "spf_attribute": "hop",
814
            "spf_max_path_cost": 20,
815
            "mandatory_metrics": {
816
                "ownership": "red"
817
            }
818
        }
819 1
        evc.uni_a.interface.id = "1"
820 1
        evc.uni_z.interface.id = "2"
821 1
        evc.uni_a.interface.switch.id = "00:00:00:00:00:00:00:01"
822 1
        evc.uni_z.interface.switch.id = "00:00:00:00:00:00:00:05"
823
824 1
        mock_paths = {
825
            "paths": [
826
                {
827
                    "cost": 5,
828
                    "hops": [
829
                        "00:00:00:00:00:00:00:01:1",
830
                        "00:00:00:00:00:00:00:01",
831
                        "00:00:00:00:00:00:00:01:3",
832
                        "00:00:00:00:00:00:00:02:2",
833
                        "00:00:00:00:00:00:00:02",
834
                        "00:00:00:00:00:00:00:02:1"
835
                        ]
836
                },
837
                {
838
                    "cost": 8,
839
                    "hops": [
840
                        "00:00:00:00:00:00:00:01:1",
841
                        "00:00:00:00:00:00:00:01",
842
                        "00:00:00:00:00:00:00:01:4",
843
                        "00:00:00:00:00:00:00:03:3",
844
                        "00:00:00:00:00:00:00:03",
845
                        "00:00:00:00:00:00:00:03:2",
846
                        "00:00:00:00:00:00:00:02:3",
847
                        "00:00:00:00:00:00:00:02",
848
                        "00:00:00:00:00:00:00:02:1"
849
                        ]
850
                },
851
            ]
852
        }
853 1
        current_path = [
854
            Link(
855
                id_to_interface_mock("00:00:00:00:00:00:00:01:3"),
856
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
857
            ),
858
        ]
859 1
        expected_disjoint_path = [
860
            Link(
861
                id_to_interface_mock("00:00:00:00:00:00:00:01:4"),
862
                id_to_interface_mock("00:00:00:00:00:00:00:03:3")
863
            ),
864
            Link(
865
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
866
                id_to_interface_mock("00:00:00:00:00:00:00:03:2")
867
            ),
868
        ]
869
870 1
        mock_response = MagicMock()
871 1
        mock_response.status_code = 200
872 1
        mock_response.json.return_value = mock_paths
873 1
        mock_httpx_post.return_value = mock_response
874 1
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
875 1
        assert len(paths) == 1
876 1
        assert (
877
            [link.id for link in paths[0]] ==
878
            [link.id for link in expected_disjoint_path]
879
        )
880
881 1
    @patch("httpx.post")
882 1
    @patch("napps.kytos.mef_eline.models.path.log")
883 1
    @patch("time.sleep")
884 1
    def test_get_disjoint_paths_error(self, _, mock_log, mock_post):
885
        """Test get_disjoint_paths with reported errors. These are caught under
886
        httpx.RequestError."""
887 1
        mock_post.side_effect = TimeoutException('mock')
888 1
        unwanted_path = [
889
            Link(
890
                id_to_interface_mock("00:00:00:00:00:00:00:01:4"),
891
                id_to_interface_mock("00:00:00:00:00:00:00:03:3")
892
            ),
893
        ]
894 1
        evc = MagicMock()
895 1
        evc.secondary_constraints = {
896
            "spf_attribute": "hop",
897
            "spf_max_path_cost": 20,
898
            "mandatory_metrics": {
899
                "ownership": "red"
900
            }
901
        }
902 1
        evc.uni_a.interface.id = "1"
903 1
        evc.uni_z.interface.id = "2"
904 1
        evc.uni_a.interface.switch.id = "00:00:00:00:00:00:00:01"
905 1
        evc.uni_z.interface.switch.id = "00:00:00:00:00:00:00:05"
906 1
        path = DynamicPathManager.get_disjoint_paths(evc, unwanted_path)
907 1
        assert len(list(path)) == 0
908 1
        assert mock_log.error.call_count == 1
909
910 1
        mock_post.side_effect = ConnectError('mock')
911 1
        path = DynamicPathManager.get_disjoint_paths(evc, unwanted_path)
912 1
        assert len(list(path)) == 0
913 1
        assert mock_log.error.call_count == 2
914
915 1
    def test_get_shared_components(self):
916
        """Test get_shared_components"""
917 1
        mock_path = {"hops": [
918
            '00:00:00:00:00:00:00:01:1',
919
            '00:00:00:00:00:00:00:01',
920
            '00:00:00:00:00:00:00:01:4',
921
            '00:00:00:00:00:00:00:05:2',
922
            '00:00:00:00:00:00:00:05',
923
            '00:00:00:00:00:00:00:05:3',
924
            '00:00:00:00:00:00:00:02:4',
925
            '00:00:00:00:00:00:00:02',
926
            '00:00:00:00:00:00:00:02:3',
927
            '00:00:00:00:00:00:00:03:2',
928
            '00:00:00:00:00:00:00:03',
929
            '00:00:00:00:00:00:00:03:1'
930
        ]}
931 1
        mock_links = [
932
            ("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:02:2"),
933
            ("00:00:00:00:00:00:00:02:3", "00:00:00:00:00:00:00:03:2")
934
        ]
935 1
        mock_switches = {"00:00:00:00:00:00:00:02"}
936 1
        actual_lk, actual_sw = DynamicPathManager.get_shared_components(
937
            mock_path, mock_links, mock_switches
938
        )
939 1
        assert actual_lk == 1
940
        assert actual_sw == 1
941