build.tests.unit.models.test_path.TestDynamicPathManager.test_get_disjoint_paths() - Code Metrics - Inspection of "[fix] update `primary_path` or `backup_path` empty" - kytos-ng/mef_eline - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#233)

by Vinicius

created 2022-12-21 16:51 UTC

TestDynamicPathManager.test_get_disjoint_paths() B

↳ Parent: build.tests.unit.models.test_path

Complexity

Conditions

Size

Total Lines	300
Code Lines	226

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	226
dl	0
loc	300
rs	7
c	0
b	0
f	0
cc	1
nop	2

How to fix Long Method

"""Module to test the Path class."""
import sys
from unittest import TestCase
from unittest.mock import call, patch, Mock, MagicMock

from kytos.core.common import EntityStatus
from kytos.core.link import Link
from kytos.core.switch import Switch

# pylint: disable=wrong-import-position

sys.path.insert(0, "/var/lib/kytos/napps/..")
# pylint: enable=wrong-import-position
from napps.kytos.mef_eline.exceptions import InvalidPath  # NOQA pycodestyle
from napps.kytos.mef_eline.models import Path, DynamicPathManager  # NOQA pycodestyle
from napps.kytos.mef_eline.tests.helpers import (
    MockResponse,
    id_to_interface_mock,
    get_link_mocked,
    get_mocked_requests,
)  # NOQA pycodestyle


class TestPath(TestCase):
    """Class to test path methods."""

    def test_is_affected_by_link_1(self):
        """Test method is affected by link."""
        path = Path()
        self.assertIs(path.is_affected_by_link(), False)

    def test_link_affected_by_interface_1(self):
        """Test method to get the link using an interface."""
        link1 = Mock()
        link1.endpoint_a = "a"
        link1.endpoint_b = "b"
        link2 = Mock()
        link2.endpoint_a = "c"
        link2.endpoint_b = "d"
        path = Path([link1, link2])
        self.assertIsNone(path.link_affected_by_interface())

    def test_link_affected_by_interface_2(self):
        """Test method to get the link using an interface."""
        link1 = Mock()
        link1.endpoint_a = "a"
        link1.endpoint_b = "b"
        link2 = Mock()
        link2.endpoint_a = "c"
        link2.endpoint_b = "d"
        path = Path([link1, link2])
        self.assertEqual(path.link_affected_by_interface("a"), link1)

    def test_status_case_1(self):
        """Test if empty link is DISABLED."""
        current_path = Path()
        self.assertEqual(current_path.status, EntityStatus.DISABLED)

    @patch("requests.get", side_effect=get_mocked_requests)
    def test_status_case_2(self, requests_mocked):
        # pylint: disable=unused-argument
        """Test if link status is DOWN."""
        link1 = get_link_mocked()
        link2 = get_link_mocked()
        link1.id = "def"
        link2.id = "abc"
        links = [link1, link2]
        current_path = Path(links)
        self.assertEqual(current_path.status, EntityStatus.DOWN)

    def test_status_case_3(self):
        """Test if link status is DISABLED."""
        links = []
        current_path = Path(links)
        self.assertEqual(current_path.status, EntityStatus.DISABLED)

    # This method will be used by the mock to replace requests.get
    def _mocked_requests_get_status_case_4(self):
        return MockResponse(
            {
                "links": {
                    "abc": {"active": True, "enabled": True},
                    "def": {"active": True, "enabled": True},
                }
            },
            200,
        )

    @patch("requests.get", side_effect=_mocked_requests_get_status_case_4)
    def test_status_case_4(self, requests_mocked):
        # pylint: disable=unused-argument
        """Test if link status is UP."""
        link1 = get_link_mocked()
        link2 = get_link_mocked()
        link1.id = "def"
        link2.id = "abc"
        links = [link1, link2]
        current_path = Path(links)
        self.assertEqual(current_path.status, EntityStatus.UP)

    # This method will be used by the mock to replace requests.get
    def _mocked_requests_get_status_case_5(self):
        return MockResponse(
            {
                "links": {
                    "abc": {"active": True, "enabled": True},
                    "def": {"active": False, "enabled": False},
                }
            },
            200,
        )

    @patch("requests.get", side_effect=_mocked_requests_get_status_case_5)
    def test_status_case_5(self, requests_mocked):
        # pylint: disable=unused-argument
        """Test if link status is UP."""
        link1 = get_link_mocked()
        link2 = get_link_mocked()
        link1.id = "def"
        link2.id = "abc"
        links = [link1, link2]
        current_path = Path(links)
        self.assertEqual(current_path.status, EntityStatus.DISABLED)

    # This method will be used by the mock to replace requests.get
    def _mocked_requests_get_status_case_6(self):
        return MockResponse(
            {
                "links": {
                    "abc": {"active": False, "enabled": False},
                    "def": {"active": False, "enabled": True},
                }
            },
            200,
        )

    @patch("requests.get", side_effect=_mocked_requests_get_status_case_6)
    def test_status_case_6(self, requests_mocked):
        # pylint: disable=unused-argument
        """Test if link status is UP."""
        link1 = get_link_mocked()
        link2 = get_link_mocked()
        link1.id = "def"
        link2.id = "abc"
        links = [link1, link2]
        current_path = Path(links)
        self.assertEqual(current_path.status, EntityStatus.DISABLED)

    def test_compare_same_paths(self):
        """Test compare paths with same links."""
        links = [
            get_link_mocked(
                endpoint_a_port=9, endpoint_b_port=10, metadata={"s_vlan": 5}
            ),
            get_link_mocked(
                endpoint_a_port=11, endpoint_b_port=12, metadata={"s_vlan": 6}
            ),
        ]

        path_1 = Path(links)
        path_2 = Path(links)
        self.assertEqual(path_1, path_2)

    def test_compare_different_paths(self):
        """Test compare paths with different links."""
        links_1 = [
            get_link_mocked(
                endpoint_a_port=9, endpoint_b_port=10, metadata={"s_vlan": 5}
            ),
            get_link_mocked(
                endpoint_a_port=11, endpoint_b_port=12, metadata={"s_vlan": 6}
            ),
        ]
        links_2 = [
            get_link_mocked(
                endpoint_a_port=12, endpoint_b_port=11, metadata={"s_vlan": 5}
            ),
            get_link_mocked(
                endpoint_a_port=14, endpoint_b_port=16, metadata={"s_vlan": 11}
            ),
        ]

        path_1 = Path(links_1)
        path_2 = Path(links_2)
        self.assertNotEqual(path_1, path_2)

    def test_as_dict(self):
        """Test path as dict."""
        links = [
            get_link_mocked(link_dict={"id": 3}),
            get_link_mocked(link_dict={"id": 2}),
        ]

        current_path = Path(links)
        expected_dict = [{"id": 3}, {"id": 2}]
        self.assertEqual(expected_dict, current_path.as_dict())

    def test_empty_is_valid(self) -> None:
        """Test empty path is valid."""
        path = Path([])
        self.assertEqual(path.is_valid(MagicMock(), MagicMock(), False), True)

    def test_is_valid(self):
        """Test is_valid method."""
        switch1 = Switch("00:00:00:00:00:00:00:01")
        switch2 = Switch("00:00:00:00:00:00:00:02")
        switch3 = Switch("00:00:00:00:00:00:00:03")
        switch4 = Switch("00:00:00:00:00:00:00:04")
        switch5 = Switch("00:00:00:00:00:00:00:05")
        switch6 = Switch("00:00:00:00:00:00:00:06")

        links1 = [
            get_link_mocked(switch_a=switch1, switch_b=switch2),
            get_link_mocked(switch_a=switch2, switch_b=switch3),
            get_link_mocked(switch_a=switch3, switch_b=switch4),
            get_link_mocked(switch_a=switch4, switch_b=switch5),
            get_link_mocked(switch_a=switch5, switch_b=switch6),
        ]

        links2 = [
            get_link_mocked(switch_a=switch1, switch_b=switch2),
            get_link_mocked(switch_a=switch3, switch_b=switch2),
            get_link_mocked(switch_a=switch3, switch_b=switch4),
        ]

        for links, switch_a, switch_z, expected in (
            (links1, switch1, switch6, True),
            (links2, switch1, switch4, False),
            (links1, switch2, switch6, False),
        ):
            with self.subTest(
                links=links,
                switch_a=switch_a,
                switch_z=switch_z,
                expected=expected,
            ):
                path = Path(links)
                if expected:
                    self.assertEqual(
                        path.is_valid(switch_a, switch_z), expected
                    )
                else:
                    with self.assertRaises(InvalidPath):
                        path.is_valid(switch_a, switch_z)


class TestDynamicPathManager(TestCase):
    """Tests for the DynamicPathManager class"""

    def test_clear_path(self):
        """Test _clear_path method"""
        path = [
            '00:00:00:00:00:00:00:01:1',
            '00:00:00:00:00:00:00:02:3',
            '00:00:00:00:00:00:00:02',
            '00:00:00:00:00:00:00:02:4',
            '00:00:00:00:00:00:00:03:2',
            '00:00:00:00:00:00:00:03',
            '00:00:00:00:00:00:00:03:1',
            '00:00:00:00:00:00:00:04:1'
        ]
        expected_path = [
            '00:00:00:00:00:00:00:01:1',
            '00:00:00:00:00:00:00:02:3',
            '00:00:00:00:00:00:00:02:4',
            '00:00:00:00:00:00:00:03:2',
            '00:00:00:00:00:00:00:03:1',
            '00:00:00:00:00:00:00:04:1'
        ]
        # pylint: disable=protected-access
        self.assertEqual(DynamicPathManager._clear_path(path), expected_path)

    def test_create_path_invalid(self):
        """Test create_path method"""
        path = [
            '00:00:00:00:00:00:00:01:1',
            '00:00:00:00:00:00:00:02:3',
            '00:00:00:00:00:00:00:02',
            '00:00:00:00:00:00:00:02:4',
            '00:00:00:00:00:00:00:03:2',
            '00:00:00:00:00:00:00:03',
            '00:00:00:00:00:00:00:03:1',
        ]
        self.assertIsNone(DynamicPathManager.create_path(path))

    @patch("requests.post")
    def test_get_best_path(self, mock_requests_post):
        """Test get_best_path method."""
        controller = MagicMock()
        controller.get_interface_by_id.side_effect = id_to_interface_mock
        DynamicPathManager.set_controller(controller)

        paths1 = {
            "paths": [
                {
                    "cost": 5,
                    "hops": [
                        "00:00:00:00:00:00:00:01:1",
                        "00:00:00:00:00:00:00:01",
                        "00:00:00:00:00:00:00:01:2",
                        "00:00:00:00:00:00:00:02:2",
                        "00:00:00:00:00:00:00:02",
                        "00:00:00:00:00:00:00:02:1"
                        ]
                },
            ]
        }

        expected_path = [
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
            ),
        ]

        # test success case
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = paths1
        mock_requests_post.return_value = mock_response

        res_paths = list(DynamicPathManager.get_best_path(MagicMock()))
        self.assertEqual(
            [link.id for link in res_paths],
            [link.id for link in expected_path]
        )

        # test failure when controller dont find the interface on create_path
        controller.get_interface_by_id.side_effect = [
            id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
            None
        ]
        self.assertIsNone(DynamicPathManager.get_best_path(MagicMock()))

        mock_response.status_code = 400
        mock_response.json.return_value = {}
        mock_requests_post.return_value = mock_response

        res_paths = DynamicPathManager.get_best_path(MagicMock())
        self.assertIsNone(res_paths)

    @patch("requests.post")
    def test_get_best_paths(self, mock_requests_post):
        """Test get_best_paths method."""
        controller = MagicMock()
        controller.get_interface_by_id.side_effect = id_to_interface_mock
        DynamicPathManager.set_controller(controller)

        paths1 = {
            "paths": [
                {
                    "cost": 5,
                    "hops": [
                        "00:00:00:00:00:00:00:01:1",
                        "00:00:00:00:00:00:00:01",
                        "00:00:00:00:00:00:00:01:2",
                        "00:00:00:00:00:00:00:02:2",
                        "00:00:00:00:00:00:00:02",
                        "00:00:00:00:00:00:00:02:1"
                        ]
                },
                {
                    "cost": 11,
                    "hops": [
                        "00:00:00:00:00:00:00:01:1",
                        "00:00:00:00:00:00:00:01",
                        "00:00:00:00:00:00:00:01:2",
                        "00:00:00:00:00:00:00:02:2",
                        "00:00:00:00:00:00:00:02",
                        "00:00:00:00:00:00:00:02:3",
                        "00:00:00:00:00:00:00:03:3",
                        "00:00:00:00:00:00:00:03",
                        "00:00:00:00:00:00:00:03:4",
                        "00:00:00:00:00:00:00:04:4",
                        "00:00:00:00:00:00:00:04",
                        "00:00:00:00:00:00:00:04:1"
                        ]
                },
            ]
        }

        expected_paths_0 = [
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
            ),
        ]

        expected_paths_1 = [
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
                id_to_interface_mock("00:00:00:00:00:00:00:03:3")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:03:4"),
                id_to_interface_mock("00:00:00:00:00:00:00:04:4")
            ),
        ]

        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = paths1
        mock_requests_post.return_value = mock_response
        kwargs = {
            "spf_max_path_cost": 8,
            "mandatory_metrics": {
                "ownership": "red"
            }
        }
        circuit = MagicMock()
        circuit.uni_a.interface.id = "1"
        circuit.uni_z.interface.id = "2"
        max_paths = 2
        res_paths = list(DynamicPathManager.get_best_paths(circuit,
                         max_paths=max_paths, **kwargs))
        self.assertEqual(
            [link.id for link in res_paths[0]],
            [link.id for link in expected_paths_0]
        )
        self.assertEqual(
            [link.id for link in res_paths[1]],
            [link.id for link in expected_paths_1]
        )
        expected_call = call(
            "http://localhost:8181/api/kytos/pathfinder/v2/",
            json={
                **{
                    "source": circuit.uni_a.interface.id,
                    "destination": circuit.uni_z.interface.id,
                    "spf_max_paths": max_paths,
                },
                **kwargs
            },
        )
        mock_requests_post.assert_has_calls([expected_call])

    @patch("requests.post")
    def test_get_disjoint_paths(self, mock_requests_post):
        """Test get_disjoint_paths method."""

        controller = MagicMock()
        controller.get_interface_by_id.side_effect = id_to_interface_mock
        DynamicPathManager.set_controller(controller)

        evc = MagicMock()
        evc.secondary_constraints = {
            "spf_max_path_cost": 20,
            "mandatory_metrics": {
                "ownership": "red"
            }
        }
        evc.uni_a.interface.id = "1"
        evc.uni_z.interface.id = "2"

        # Topo0
        paths1 = {
            "paths": [
                {
                    "cost": 11,
                    "hops": [
                        "00:00:00:00:00:00:00:01:1",
                        "00:00:00:00:00:00:00:01",
                        "00:00:00:00:00:00:00:01:2",
                        "00:00:00:00:00:00:00:02:2",
                        "00:00:00:00:00:00:00:02",
                        "00:00:00:00:00:00:00:02:3",
                        "00:00:00:00:00:00:00:04:2",
                        "00:00:00:00:00:00:00:04",
                        "00:00:00:00:00:00:00:04:3",
                        "00:00:00:00:00:00:00:05:2",
                        "00:00:00:00:00:00:00:05",
                        "00:00:00:00:00:00:00:05:1"
                        ]
                },
                {
                    "cost": 11,
                    "hops": [
                        "00:00:00:00:00:00:00:01:1",
                        "00:00:00:00:00:00:00:01",
                        "00:00:00:00:00:00:00:01:3",
                        "00:00:00:00:00:00:00:03:2",
                        "00:00:00:00:00:00:00:03",
                        "00:00:00:00:00:00:00:03:3",
                        "00:00:00:00:00:00:00:04:4",
                        "00:00:00:00:00:00:00:04",
                        "00:00:00:00:00:00:00:04:3",
                        "00:00:00:00:00:00:00:05:2",
                        "00:00:00:00:00:00:00:05",
                        "00:00:00:00:00:00:00:05:1"
                        ]
                },
                {
                    "cost": 14,
                    "hops": [
                        "00:00:00:00:00:00:00:01:1",
                        "00:00:00:00:00:00:00:01",
                        "00:00:00:00:00:00:00:01:2",
                        "00:00:00:00:00:00:00:02:2",
                        "00:00:00:00:00:00:00:02",
                        "00:00:00:00:00:00:00:02:3",
                        "00:00:00:00:00:00:00:04:2",
                        "00:00:00:00:00:00:00:04",
                        "00:00:00:00:00:00:00:04:5",
                        "00:00:00:00:00:00:00:06:2",
                        "00:00:00:00:00:00:00:06",
                        "00:00:00:00:00:00:00:06:3",
                        "00:00:00:00:00:00:00:05:3",
                        "00:00:00:00:00:00:00:05",
                        "00:00:00:00:00:00:00:05:1"
                    ]
                },
                {
                    "cost": 14,
                    "hops": [
                        "00:00:00:00:00:00:00:01:1",
                        "00:00:00:00:00:00:00:01",
                        "00:00:00:00:00:00:00:01:3",
                        "00:00:00:00:00:00:00:03:2",
                        "00:00:00:00:00:00:00:03",
                        "00:00:00:00:00:00:00:03:3",
                        "00:00:00:00:00:00:00:04:4",
                        "00:00:00:00:00:00:00:04",
                        "00:00:00:00:00:00:00:04:5",
                        "00:00:00:00:00:00:00:06:2",
                        "00:00:00:00:00:00:00:06",
                        "00:00:00:00:00:00:00:06:3",
                        "00:00:00:00:00:00:00:05:3",
                        "00:00:00:00:00:00:00:05",
                        "00:00:00:00:00:00:00:05:1"
                    ]
                },
                {
                    "cost": 17,
                    "hops": [
                        "00:00:00:00:00:00:00:01:1",
                        "00:00:00:00:00:00:00:01",
                        "00:00:00:00:00:00:00:01:3",
                        "00:00:00:00:00:00:00:03:2",
                        "00:00:00:00:00:00:00:03",
                        "00:00:00:00:00:00:00:03:3",
                        "00:00:00:00:00:00:00:04:4",
                        "00:00:00:00:00:00:00:04",
                        "00:00:00:00:00:00:00:04:5",
                        "00:00:00:00:00:00:00:06:2",
                        "00:00:00:00:00:00:00:06",
                        "00:00:00:00:00:00:00:06:4",
                        "00:00:00:00:00:00:00:07:2",
                        "00:00:00:00:00:00:00:07",
                        "00:00:00:00:00:00:00:07:3",
                        "00:00:00:00:00:00:00:05:4",
                        "00:00:00:00:00:00:00:05",
                        "00:00:00:00:00:00:00:05:1"
                    ]
                },
            ]
        }

        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = paths1

        # when we dont have the current_path
        mock_requests_post.return_value = mock_response
        disjoint_paths = list(DynamicPathManager.get_disjoint_paths(evc, []))
        self.assertEqual(disjoint_paths, [])

        current_path = [
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:02:2")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
                id_to_interface_mock("00:00:00:00:00:00:00:04:2")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:04:3"),
                id_to_interface_mock("00:00:00:00:00:00:00:05:2")
            ),
        ]

        # only one path available from pathfinder (precesilly the
        # current_path), so the maximum disjoint path will be empty
        mock_response.json.return_value = {"paths": paths1["paths"][0:1]}
        mock_requests_post.return_value = mock_response
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
        self.assertEqual(len(paths), 0)

        expected_disjoint_path = [
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:01:3"),
                id_to_interface_mock("00:00:00:00:00:00:00:03:2")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:03:3"),
                id_to_interface_mock("00:00:00:00:00:00:00:04:4")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:04:5"),
                id_to_interface_mock("00:00:00:00:00:00:00:06:2")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:06:3"),
                id_to_interface_mock("00:00:00:00:00:00:00:05:3")
            ),
        ]

        # there are one alternative path
        mock_response.json.return_value = paths1
        mock_requests_post.return_value = mock_response
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
        self.assertEqual(len(paths), 4)
        # for more information on the paths please refer to EP029
        self.assertEqual(len(paths[0]), 4)  # path S-Z-W-I-D
        self.assertEqual(len(paths[1]), 5)  # path S-Z-W-I-J-D
        self.assertEqual(len(paths[2]), 3)  # path S-Z-W-D
        self.assertEqual(len(paths[3]), 4)  # path S-X-W-I-D
        self.assertEqual(
            [link.id for link in paths[0]],
            [link.id for link in expected_disjoint_path]
        )

        max_paths = 10
        expected_call = call(
            "http://localhost:8181/api/kytos/pathfinder/v2/",
            json={
                **{
                    "source": evc.uni_a.interface.id,
                    "destination": evc.uni_z.interface.id,
                    "spf_max_paths": max_paths,
                },
                **evc.secondary_constraints
            },
        )
        assert mock_requests_post.call_count >= 1
        # If secondary_constraints are set they are expected to be parametrized
        mock_requests_post.assert_has_calls([expected_call])

        # EP029 Topo2
        paths2 = {
            "paths": [
                {
                    "cost": 14,
                    "hops": [
                        "00:00:00:00:00:00:00:01:1",
                        "00:00:00:00:00:00:00:01",
                        "00:00:00:00:00:00:00:01:2",
                        "00:00:00:00:00:00:00:02:1",
                        "00:00:00:00:00:00:00:02",
                        "00:00:00:00:00:00:00:02:2",
                        "00:00:00:00:00:00:00:03:1",
                        "00:00:00:00:00:00:00:03",
                        "00:00:00:00:00:00:00:03:2",
                        "00:00:00:00:00:00:00:04:1",
                        "00:00:00:00:00:00:00:04",
                        "00:00:00:00:00:00:00:04:2",
                        "00:00:00:00:00:00:00:07:2",
                        "00:00:00:00:00:00:00:07",
                        "00:00:00:00:00:00:00:07:1"
                    ]
                },
                {
                    "cost": 17,
                    "hops": [
                        "00:00:00:00:00:00:00:01:1",
                        "00:00:00:00:00:00:00:01",
                        "00:00:00:00:00:00:00:01:2",
                        "00:00:00:00:00:00:00:02:1",
                        "00:00:00:00:00:00:00:02",
                        "00:00:00:00:00:00:00:02:3",
                        "00:00:00:00:00:00:00:05:1",
                        "00:00:00:00:00:00:00:05",
                        "00:00:00:00:00:00:00:05:2",
                        "00:00:00:00:00:00:00:06:1",
                        "00:00:00:00:00:00:00:06",
                        "00:00:00:00:00:00:00:06:2",
                        "00:00:00:00:00:00:00:04:3",
                        "00:00:00:00:00:00:00:04",
                        "00:00:00:00:00:00:00:04:2",
                        "00:00:00:00:00:00:00:07:2",
                        "00:00:00:00:00:00:00:07",
                        "00:00:00:00:00:00:00:07:1"
                    ]
                }
            ]
        }

        current_path = [
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:02:1")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:02:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:03:1")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:03:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:04:1")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:04:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:07:2")
            ),
        ]

        expected_disjoint_path = [
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:02:1")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
                id_to_interface_mock("00:00:00:00:00:00:00:05:1")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:05:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:06:1")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:06:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:04:3")
            ),
            Link(
                id_to_interface_mock("00:00:00:00:00:00:00:04:2"),
                id_to_interface_mock("00:00:00:00:00:00:00:07:2")
            ),
        ]

        mock_response.json.return_value = {"paths": paths2["paths"]}
        mock_requests_post.return_value = mock_response
        paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
        self.assertEqual(len(paths), 1)
        self.assertEqual(
            [link.id for link in paths[0]],
            [link.id for link in expected_disjoint_path]
        )


1			"""Module to test the Path class."""
2			import sys
3			from unittest import TestCase
4			from unittest.mock import call, patch, Mock, MagicMock
5
6			from kytos.core.common import EntityStatus
7			from kytos.core.link import Link
8			from kytos.core.switch import Switch
9
10			# pylint: disable=wrong-import-position
11
12			sys.path.insert(0, "/var/lib/kytos/napps/..")
13			# pylint: enable=wrong-import-position
14			from napps.kytos.mef_eline.exceptions import InvalidPath # NOQA pycodestyle
15			from napps.kytos.mef_eline.models import Path, DynamicPathManager # NOQA pycodestyle
16			from napps.kytos.mef_eline.tests.helpers import (
17			MockResponse,
18			id_to_interface_mock,
19			get_link_mocked,
20			get_mocked_requests,
21			) # NOQA pycodestyle
22
23
24			class TestPath(TestCase):
25			"""Class to test path methods."""
26
27			def test_is_affected_by_link_1(self):
28			"""Test method is affected by link."""
29			path = Path()
30			self.assertIs(path.is_affected_by_link(), False)
31
32			def test_link_affected_by_interface_1(self):
33			"""Test method to get the link using an interface."""
34			link1 = Mock()
35			link1.endpoint_a = "a"
36			link1.endpoint_b = "b"
37			link2 = Mock()
38			link2.endpoint_a = "c"
39			link2.endpoint_b = "d"
40			path = Path([link1, link2])
41			self.assertIsNone(path.link_affected_by_interface())
42
43			def test_link_affected_by_interface_2(self):
44			"""Test method to get the link using an interface."""
45			link1 = Mock()
46			link1.endpoint_a = "a"
47			link1.endpoint_b = "b"
48			link2 = Mock()
49			link2.endpoint_a = "c"
50			link2.endpoint_b = "d"
51			path = Path([link1, link2])
52			self.assertEqual(path.link_affected_by_interface("a"), link1)
53
54			def test_status_case_1(self):
55			"""Test if empty link is DISABLED."""
56			current_path = Path()
57			self.assertEqual(current_path.status, EntityStatus.DISABLED)
58
59			@patch("requests.get", side_effect=get_mocked_requests)
60			def test_status_case_2(self, requests_mocked):
61			# pylint: disable=unused-argument
62			"""Test if link status is DOWN."""
63			link1 = get_link_mocked()
64			link2 = get_link_mocked()
65			link1.id = "def"
66			link2.id = "abc"
67			links = [link1, link2]
68			current_path = Path(links)
69			self.assertEqual(current_path.status, EntityStatus.DOWN)
70
71			def test_status_case_3(self):
72			"""Test if link status is DISABLED."""
73			links = []
74			current_path = Path(links)
75			self.assertEqual(current_path.status, EntityStatus.DISABLED)
76
77			# This method will be used by the mock to replace requests.get
78			def _mocked_requests_get_status_case_4(self):
79			return MockResponse(
80			{
81			"links": {
82			"abc": {"active": True, "enabled": True},
83			"def": {"active": True, "enabled": True},
84			}
85			},
86			200,
87			)
88
89			@patch("requests.get", side_effect=_mocked_requests_get_status_case_4)
90			def test_status_case_4(self, requests_mocked):
91			# pylint: disable=unused-argument
92			"""Test if link status is UP."""
93			link1 = get_link_mocked()
94			link2 = get_link_mocked()
95			link1.id = "def"
96			link2.id = "abc"
97			links = [link1, link2]
98			current_path = Path(links)
99			self.assertEqual(current_path.status, EntityStatus.UP)
100
101			# This method will be used by the mock to replace requests.get
102			def _mocked_requests_get_status_case_5(self):
103			return MockResponse(
104			{
105			"links": {
106			"abc": {"active": True, "enabled": True},
107			"def": {"active": False, "enabled": False},
108			}
109			},
110			200,
111			)
112
113			@patch("requests.get", side_effect=_mocked_requests_get_status_case_5)
114			def test_status_case_5(self, requests_mocked):
115			# pylint: disable=unused-argument
116			"""Test if link status is UP."""
117			link1 = get_link_mocked()
118			link2 = get_link_mocked()
119			link1.id = "def"
120			link2.id = "abc"
121			links = [link1, link2]
122			current_path = Path(links)
123			self.assertEqual(current_path.status, EntityStatus.DISABLED)
124
125			# This method will be used by the mock to replace requests.get
126			def _mocked_requests_get_status_case_6(self):
127			return MockResponse(
128			{
129			"links": {
130			"abc": {"active": False, "enabled": False},
131			"def": {"active": False, "enabled": True},
132			}
133			},
134			200,
135			)
136
137			@patch("requests.get", side_effect=_mocked_requests_get_status_case_6)
138			def test_status_case_6(self, requests_mocked):
139			# pylint: disable=unused-argument
140			"""Test if link status is UP."""
141			link1 = get_link_mocked()
142			link2 = get_link_mocked()
143			link1.id = "def"
144			link2.id = "abc"
145			links = [link1, link2]
146			current_path = Path(links)
147			self.assertEqual(current_path.status, EntityStatus.DISABLED)
148
149			def test_compare_same_paths(self):
150			"""Test compare paths with same links."""
151			links = [
152			get_link_mocked(
153			endpoint_a_port=9, endpoint_b_port=10, metadata={"s_vlan": 5}
154			),
155			get_link_mocked(
156			endpoint_a_port=11, endpoint_b_port=12, metadata={"s_vlan": 6}
157			),
158			]
159
160			path_1 = Path(links)
161			path_2 = Path(links)
162			self.assertEqual(path_1, path_2)
163
164			def test_compare_different_paths(self):
165			"""Test compare paths with different links."""
166			links_1 = [
167			get_link_mocked(
168			endpoint_a_port=9, endpoint_b_port=10, metadata={"s_vlan": 5}
169			),
170			get_link_mocked(
171			endpoint_a_port=11, endpoint_b_port=12, metadata={"s_vlan": 6}
172			),
173			]
174			links_2 = [
175			get_link_mocked(
176			endpoint_a_port=12, endpoint_b_port=11, metadata={"s_vlan": 5}
177			),
178			get_link_mocked(
179			endpoint_a_port=14, endpoint_b_port=16, metadata={"s_vlan": 11}
180			),
181			]
182
183			path_1 = Path(links_1)
184			path_2 = Path(links_2)
185			self.assertNotEqual(path_1, path_2)
186
187			def test_as_dict(self):
188			"""Test path as dict."""
189			links = [
190			get_link_mocked(link_dict={"id": 3}),
191			get_link_mocked(link_dict={"id": 2}),
192			]
193
194			current_path = Path(links)
195			expected_dict = [{"id": 3}, {"id": 2}]
196			self.assertEqual(expected_dict, current_path.as_dict())
197
198			def test_empty_is_valid(self) -> None:
199			"""Test empty path is valid."""
200			path = Path([])
201			self.assertEqual(path.is_valid(MagicMock(), MagicMock(), False), True)
202
203			def test_is_valid(self):
204			"""Test is_valid method."""
205			switch1 = Switch("00:00:00:00:00:00:00:01")
206			switch2 = Switch("00:00:00:00:00:00:00:02")
207			switch3 = Switch("00:00:00:00:00:00:00:03")
208			switch4 = Switch("00:00:00:00:00:00:00:04")
209			switch5 = Switch("00:00:00:00:00:00:00:05")
210			switch6 = Switch("00:00:00:00:00:00:00:06")
211
212			links1 = [
213			get_link_mocked(switch_a=switch1, switch_b=switch2),
214			get_link_mocked(switch_a=switch2, switch_b=switch3),
215			get_link_mocked(switch_a=switch3, switch_b=switch4),
216			get_link_mocked(switch_a=switch4, switch_b=switch5),
217			get_link_mocked(switch_a=switch5, switch_b=switch6),
218			]
219
220			links2 = [
221			get_link_mocked(switch_a=switch1, switch_b=switch2),
222			get_link_mocked(switch_a=switch3, switch_b=switch2),
223			get_link_mocked(switch_a=switch3, switch_b=switch4),
224			]
225
226			for links, switch_a, switch_z, expected in (
227			(links1, switch1, switch6, True),
228			(links2, switch1, switch4, False),
229			(links1, switch2, switch6, False),
230			):
231			with self.subTest(
232			links=links,
233			switch_a=switch_a,
234			switch_z=switch_z,
235			expected=expected,
236			):
237			path = Path(links)
238			if expected:
239			self.assertEqual(
240			path.is_valid(switch_a, switch_z), expected
241			)
242			else:
243			with self.assertRaises(InvalidPath):
244			path.is_valid(switch_a, switch_z)
245
246
247			class TestDynamicPathManager(TestCase):
248			"""Tests for the DynamicPathManager class"""
249
250			def test_clear_path(self):
251			"""Test _clear_path method"""
252			path = [
253			'00:00:00:00:00:00:00:01:1',
254			'00:00:00:00:00:00:00:02:3',
255			'00:00:00:00:00:00:00:02',
256			'00:00:00:00:00:00:00:02:4',
257			'00:00:00:00:00:00:00:03:2',
258			'00:00:00:00:00:00:00:03',
259			'00:00:00:00:00:00:00:03:1',
260			'00:00:00:00:00:00:00:04:1'
261			]
262			expected_path = [
263			'00:00:00:00:00:00:00:01:1',
264			'00:00:00:00:00:00:00:02:3',
265			'00:00:00:00:00:00:00:02:4',
266			'00:00:00:00:00:00:00:03:2',
267			'00:00:00:00:00:00:00:03:1',
268			'00:00:00:00:00:00:00:04:1'
269			]
270			# pylint: disable=protected-access
271			self.assertEqual(DynamicPathManager._clear_path(path), expected_path)
272
273			def test_create_path_invalid(self):
274			"""Test create_path method"""
275			path = [
276			'00:00:00:00:00:00:00:01:1',
277			'00:00:00:00:00:00:00:02:3',
278			'00:00:00:00:00:00:00:02',
279			'00:00:00:00:00:00:00:02:4',
280			'00:00:00:00:00:00:00:03:2',
281			'00:00:00:00:00:00:00:03',
282			'00:00:00:00:00:00:00:03:1',
283			]
284			self.assertIsNone(DynamicPathManager.create_path(path))
285
286			@patch("requests.post")
287			def test_get_best_path(self, mock_requests_post):
288			"""Test get_best_path method."""
289			controller = MagicMock()
290			controller.get_interface_by_id.side_effect = id_to_interface_mock
291			DynamicPathManager.set_controller(controller)
292
293			paths1 = {
294			"paths": [
295			{
296			"cost": 5,
297			"hops": [
298			"00:00:00:00:00:00:00:01:1",
299			"00:00:00:00:00:00:00:01",
300			"00:00:00:00:00:00:00:01:2",
301			"00:00:00:00:00:00:00:02:2",
302			"00:00:00:00:00:00:00:02",
303			"00:00:00:00:00:00:00:02:1"
304			]
305			},
306			]
307			}
308
309			expected_path = [
310			Link(
311			id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
312			id_to_interface_mock("00:00:00:00:00:00:00:02:2")
313			),
314			]
315
316			# test success case
317			mock_response = MagicMock()
318			mock_response.status_code = 200
319			mock_response.json.return_value = paths1
320			mock_requests_post.return_value = mock_response
321
322			res_paths = list(DynamicPathManager.get_best_path(MagicMock()))
323			self.assertEqual(
324			[link.id for link in res_paths],
325			[link.id for link in expected_path]
326			)
327
328			# test failure when controller dont find the interface on create_path
329			controller.get_interface_by_id.side_effect = [
330			id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
331			None
332			]
333			self.assertIsNone(DynamicPathManager.get_best_path(MagicMock()))
334
335			mock_response.status_code = 400
336			mock_response.json.return_value = {}
337			mock_requests_post.return_value = mock_response
338
339			res_paths = DynamicPathManager.get_best_path(MagicMock())
340			self.assertIsNone(res_paths)
341
342			@patch("requests.post")
343			def test_get_best_paths(self, mock_requests_post):
344			"""Test get_best_paths method."""
345			controller = MagicMock()
346			controller.get_interface_by_id.side_effect = id_to_interface_mock
347			DynamicPathManager.set_controller(controller)
348
349			paths1 = {
350			"paths": [
351			{
352			"cost": 5,
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:1"
360			]
361			},
362			{
363			"cost": 11,
364			"hops": [
365			"00:00:00:00:00:00:00:01:1",
366			"00:00:00:00:00:00:00:01",
367			"00:00:00:00:00:00:00:01:2",
368			"00:00:00:00:00:00:00:02:2",
369			"00:00:00:00:00:00:00:02",
370			"00:00:00:00:00:00:00:02:3",
371			"00:00:00:00:00:00:00:03:3",
372			"00:00:00:00:00:00:00:03",
373			"00:00:00:00:00:00:00:03:4",
374			"00:00:00:00:00:00:00:04:4",
375			"00:00:00:00:00:00:00:04",
376			"00:00:00:00:00:00:00:04:1"
377			]
378			},
379			]
380			}
381
382			expected_paths_0 = [
383			Link(
384			id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
385			id_to_interface_mock("00:00:00:00:00:00:00:02:2")
386			),
387			]
388
389			expected_paths_1 = [
390			Link(
391			id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
392			id_to_interface_mock("00:00:00:00:00:00:00:02:2")
393			),
394			Link(
395			id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
396			id_to_interface_mock("00:00:00:00:00:00:00:03:3")
397			),
398			Link(
399			id_to_interface_mock("00:00:00:00:00:00:00:03:4"),
400			id_to_interface_mock("00:00:00:00:00:00:00:04:4")
401			),
402			]
403
404			mock_response = MagicMock()
405			mock_response.status_code = 200
406			mock_response.json.return_value = paths1
407			mock_requests_post.return_value = mock_response
408			kwargs = {
409			"spf_max_path_cost": 8,
410			"mandatory_metrics": {
411			"ownership": "red"
412			}
413			}
414			circuit = MagicMock()
415			circuit.uni_a.interface.id = "1"
416			circuit.uni_z.interface.id = "2"
417			max_paths = 2
418			res_paths = list(DynamicPathManager.get_best_paths(circuit,
419			max_paths=max_paths, **kwargs))
420			self.assertEqual(
421			[link.id for link in res_paths[0]],
422			[link.id for link in expected_paths_0]
423			)
424			self.assertEqual(
425			[link.id for link in res_paths[1]],
426			[link.id for link in expected_paths_1]
427			)
428			expected_call = call(
429			"http://localhost:8181/api/kytos/pathfinder/v2/",
430			json={
431			**{
432			"source": circuit.uni_a.interface.id,
433			"destination": circuit.uni_z.interface.id,
434			"spf_max_paths": max_paths,
435			},
436			**kwargs
437			},
438			)
439			mock_requests_post.assert_has_calls([expected_call])
440
441			@patch("requests.post")
442			def test_get_disjoint_paths(self, mock_requests_post):
443			"""Test get_disjoint_paths method."""
444
445			controller = MagicMock()
446			controller.get_interface_by_id.side_effect = id_to_interface_mock
447			DynamicPathManager.set_controller(controller)
448
449			evc = MagicMock()
450			evc.secondary_constraints = {
451			"spf_max_path_cost": 20,
452			"mandatory_metrics": {
453			"ownership": "red"
454			}
455			}
456			evc.uni_a.interface.id = "1"
457			evc.uni_z.interface.id = "2"
458
459			# Topo0
460			paths1 = {
461			"paths": [
462			{
463			"cost": 11,
464			"hops": [
465			"00:00:00:00:00:00:00:01:1",
466			"00:00:00:00:00:00:00:01",
467			"00:00:00:00:00:00:00:01:2",
468			"00:00:00:00:00:00:00:02:2",
469			"00:00:00:00:00:00:00:02",
470			"00:00:00:00:00:00:00:02:3",
471			"00:00:00:00:00:00:00:04:2",
472			"00:00:00:00:00:00:00:04",
473			"00:00:00:00:00:00:00:04:3",
474			"00:00:00:00:00:00:00:05:2",
475			"00:00:00:00:00:00:00:05",
476			"00:00:00:00:00:00:00:05:1"
477			]
478			},
479			{
480			"cost": 11,
481			"hops": [
482			"00:00:00:00:00:00:00:01:1",
483			"00:00:00:00:00:00:00:01",
484			"00:00:00:00:00:00:00:01:3",
485			"00:00:00:00:00:00:00:03:2",
486			"00:00:00:00:00:00:00:03",
487			"00:00:00:00:00:00:00:03:3",
488			"00:00:00:00:00:00:00:04:4",
489			"00:00:00:00:00:00:00:04",
490			"00:00:00:00:00:00:00:04:3",
491			"00:00:00:00:00:00:00:05:2",
492			"00:00:00:00:00:00:00:05",
493			"00:00:00:00:00:00:00:05:1"
494			]
495			},
496			{
497			"cost": 14,
498			"hops": [
499			"00:00:00:00:00:00:00:01:1",
500			"00:00:00:00:00:00:00:01",
501			"00:00:00:00:00:00:00:01:2",
502			"00:00:00:00:00:00:00:02:2",
503			"00:00:00:00:00:00:00:02",
504			"00:00:00:00:00:00:00:02:3",
505			"00:00:00:00:00:00:00:04:2",
506			"00:00:00:00:00:00:00:04",
507			"00:00:00:00:00:00:00:04:5",
508			"00:00:00:00:00:00:00:06:2",
509			"00:00:00:00:00:00:00:06",
510			"00:00:00:00:00:00:00:06:3",
511			"00:00:00:00:00:00:00:05:3",
512			"00:00:00:00:00:00:00:05",
513			"00:00:00:00:00:00:00:05:1"
514			]
515			},
516			{
517			"cost": 14,
518			"hops": [
519			"00:00:00:00:00:00:00:01:1",
520			"00:00:00:00:00:00:00:01",
521			"00:00:00:00:00:00:00:01:3",
522			"00:00:00:00:00:00:00:03:2",
523			"00:00:00:00:00:00:00:03",
524			"00:00:00:00:00:00:00:03:3",
525			"00:00:00:00:00:00:00:04:4",
526			"00:00:00:00:00:00:00:04",
527			"00:00:00:00:00:00:00:04:5",
528			"00:00:00:00:00:00:00:06:2",
529			"00:00:00:00:00:00:00:06",
530			"00:00:00:00:00:00:00:06:3",
531			"00:00:00:00:00:00:00:05:3",
532			"00:00:00:00:00:00:00:05",
533			"00:00:00:00:00:00:00:05:1"
534			]
535			},
536			{
537			"cost": 17,
538			"hops": [
539			"00:00:00:00:00:00:00:01:1",
540			"00:00:00:00:00:00:00:01",
541			"00:00:00:00:00:00:00:01:3",
542			"00:00:00:00:00:00:00:03:2",
543			"00:00:00:00:00:00:00:03",
544			"00:00:00:00:00:00:00:03:3",
545			"00:00:00:00:00:00:00:04:4",
546			"00:00:00:00:00:00:00:04",
547			"00:00:00:00:00:00:00:04:5",
548			"00:00:00:00:00:00:00:06:2",
549			"00:00:00:00:00:00:00:06",
550			"00:00:00:00:00:00:00:06:4",
551			"00:00:00:00:00:00:00:07:2",
552			"00:00:00:00:00:00:00:07",
553			"00:00:00:00:00:00:00:07:3",
554			"00:00:00:00:00:00:00:05:4",
555			"00:00:00:00:00:00:00:05",
556			"00:00:00:00:00:00:00:05:1"
557			]
558			},
559			]
560			}
561
562			mock_response = MagicMock()
563			mock_response.status_code = 200
564			mock_response.json.return_value = paths1
565
566			# when we dont have the current_path
567			mock_requests_post.return_value = mock_response
568			disjoint_paths = list(DynamicPathManager.get_disjoint_paths(evc, []))
569			self.assertEqual(disjoint_paths, [])
570
571			current_path = [
572			Link(
573			id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
574			id_to_interface_mock("00:00:00:00:00:00:00:02:2")
575			),
576			Link(
577			id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
578			id_to_interface_mock("00:00:00:00:00:00:00:04:2")
579			),
580			Link(
581			id_to_interface_mock("00:00:00:00:00:00:00:04:3"),
582			id_to_interface_mock("00:00:00:00:00:00:00:05:2")
583			),
584			]
585
586			# only one path available from pathfinder (precesilly the
587			# current_path), so the maximum disjoint path will be empty
588			mock_response.json.return_value = {"paths": paths1["paths"][0:1]}
589			mock_requests_post.return_value = mock_response
590			paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
591			self.assertEqual(len(paths), 0)
592
593			expected_disjoint_path = [
594			Link(
595			id_to_interface_mock("00:00:00:00:00:00:00:01:3"),
596			id_to_interface_mock("00:00:00:00:00:00:00:03:2")
597			),
598			Link(
599			id_to_interface_mock("00:00:00:00:00:00:00:03:3"),
600			id_to_interface_mock("00:00:00:00:00:00:00:04:4")
601			),
602			Link(
603			id_to_interface_mock("00:00:00:00:00:00:00:04:5"),
604			id_to_interface_mock("00:00:00:00:00:00:00:06:2")
605			),
606			Link(
607			id_to_interface_mock("00:00:00:00:00:00:00:06:3"),
608			id_to_interface_mock("00:00:00:00:00:00:00:05:3")
609			),
610			]
611
612			# there are one alternative path
613			mock_response.json.return_value = paths1
614			mock_requests_post.return_value = mock_response
615			paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
616			self.assertEqual(len(paths), 4)
617			# for more information on the paths please refer to EP029
618			self.assertEqual(len(paths[0]), 4) # path S-Z-W-I-D
619			self.assertEqual(len(paths[1]), 5) # path S-Z-W-I-J-D
620			self.assertEqual(len(paths[2]), 3) # path S-Z-W-D
621			self.assertEqual(len(paths[3]), 4) # path S-X-W-I-D
622			self.assertEqual(
623			[link.id for link in paths[0]],
624			[link.id for link in expected_disjoint_path]
625			)
626
627			max_paths = 10
628			expected_call = call(
629			"http://localhost:8181/api/kytos/pathfinder/v2/",
630			json={
631			**{
632			"source": evc.uni_a.interface.id,
633			"destination": evc.uni_z.interface.id,
634			"spf_max_paths": max_paths,
635			},
636			**evc.secondary_constraints
637			},
638			)
639			assert mock_requests_post.call_count >= 1
640			# If secondary_constraints are set they are expected to be parametrized
641			mock_requests_post.assert_has_calls([expected_call])
642
643			# EP029 Topo2
644			paths2 = {
645			"paths": [
646			{
647			"cost": 14,
648			"hops": [
649			"00:00:00:00:00:00:00:01:1",
650			"00:00:00:00:00:00:00:01",
651			"00:00:00:00:00:00:00:01:2",
652			"00:00:00:00:00:00:00:02:1",
653			"00:00:00:00:00:00:00:02",
654			"00:00:00:00:00:00:00:02:2",
655			"00:00:00:00:00:00:00:03:1",
656			"00:00:00:00:00:00:00:03",
657			"00:00:00:00:00:00:00:03:2",
658			"00:00:00:00:00:00:00:04:1",
659			"00:00:00:00:00:00:00:04",
660			"00:00:00:00:00:00:00:04:2",
661			"00:00:00:00:00:00:00:07:2",
662			"00:00:00:00:00:00:00:07",
663			"00:00:00:00:00:00:00:07:1"
664			]
665			},
666			{
667			"cost": 17,
668			"hops": [
669			"00:00:00:00:00:00:00:01:1",
670			"00:00:00:00:00:00:00:01",
671			"00:00:00:00:00:00:00:01:2",
672			"00:00:00:00:00:00:00:02:1",
673			"00:00:00:00:00:00:00:02",
674			"00:00:00:00:00:00:00:02:3",
675			"00:00:00:00:00:00:00:05:1",
676			"00:00:00:00:00:00:00:05",
677			"00:00:00:00:00:00:00:05:2",
678			"00:00:00:00:00:00:00:06:1",
679			"00:00:00:00:00:00:00:06",
680			"00:00:00:00:00:00:00:06:2",
681			"00:00:00:00:00:00:00:04:3",
682			"00:00:00:00:00:00:00:04",
683			"00:00:00:00:00:00:00:04:2",
684			"00:00:00:00:00:00:00:07:2",
685			"00:00:00:00:00:00:00:07",
686			"00:00:00:00:00:00:00:07:1"
687			]
688			}
689			]
690			}
691
692			current_path = [
693			Link(
694			id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
695			id_to_interface_mock("00:00:00:00:00:00:00:02:1")
696			),
697			Link(
698			id_to_interface_mock("00:00:00:00:00:00:00:02:2"),
699			id_to_interface_mock("00:00:00:00:00:00:00:03:1")
700			),
701			Link(
702			id_to_interface_mock("00:00:00:00:00:00:00:03:2"),
703			id_to_interface_mock("00:00:00:00:00:00:00:04:1")
704			),
705			Link(
706			id_to_interface_mock("00:00:00:00:00:00:00:04:2"),
707			id_to_interface_mock("00:00:00:00:00:00:00:07:2")
708			),
709			]
710
711			expected_disjoint_path = [
712			Link(
713			id_to_interface_mock("00:00:00:00:00:00:00:01:2"),
714			id_to_interface_mock("00:00:00:00:00:00:00:02:1")
715			),
716			Link(
717			id_to_interface_mock("00:00:00:00:00:00:00:02:3"),
718			id_to_interface_mock("00:00:00:00:00:00:00:05:1")
719			),
720			Link(
721			id_to_interface_mock("00:00:00:00:00:00:00:05:2"),
722			id_to_interface_mock("00:00:00:00:00:00:00:06:1")
723			),
724			Link(
725			id_to_interface_mock("00:00:00:00:00:00:00:06:2"),
726			id_to_interface_mock("00:00:00:00:00:00:00:04:3")
727			),
728			Link(
729			id_to_interface_mock("00:00:00:00:00:00:00:04:2"),
730			id_to_interface_mock("00:00:00:00:00:00:00:07:2")
731			),
732			]
733
734			mock_response.json.return_value = {"paths": paths2["paths"]}
735			mock_requests_post.return_value = mock_response
736			paths = list(DynamicPathManager.get_disjoint_paths(evc, current_path))
737			self.assertEqual(len(paths), 1)
738			self.assertEqual(
739			[link.id for link in paths[0]],
740			[link.id for link in expected_disjoint_path]
741			)
742

kytos-ng / mef_eline

Pull Request — master (#233)

TestDynamicPathManager.test_get_disjoint_paths() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like