Passed
Pull Request — master (#491)
by Aldo
04:10
created

build.tests.unit.models.test_path   A

Complexity

Total Complexity 30

Size/Duplication

Total Lines 878
Duplicated Lines 4.21 %

Test Coverage

Coverage 99.66%

Importance

Changes 0
Metric Value
eloc 656
dl 37
loc 878
ccs 290
cts 291
cp 0.9966
rs 9.944
c 0
b 0
f 0
wmc 30

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