build.tests.unit.test_main.TestMain.test_shortest_path() - Code Metrics - Inspection of "Merge pull request #107 from kytos-ng/fix/update_e..." - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( bdb3e4...e1fc2e )

by Aldo

created 2025-12-18 18:18 UTC

TestMain.test_shortest_path() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	11
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	7
CRAP Score	2.0438

Importance

Changes

Metric	Value
cc	2
eloc	9
nop	1
dl	0
loc	11
ccs	7
cts	9
cp	0.7778
crap	2.0438
rs	9.95
c	0
b	0
f	0

"""Test Main methods."""
import asyncio
import pytest
from unittest.mock import MagicMock, patch

from kytos.lib.helpers import get_controller_mock, get_test_client

# pylint: disable=import-error
from napps.kytos.pathfinder.main import Main
from napps.kytos.pathfinder.exceptions import LinkNotFound
from tests.helpers import get_topology_mock, get_topology_with_metadata


# pylint: disable=protected-access
class TestMain:
    """Tests for the Main class."""

    def setup_method(self):
        """Execute steps before each tests."""
        self.controller = get_controller_mock()
        self.napp = Main(self.controller)
        self.api_client = get_test_client(self.controller, self.napp)
        self.endpoint = "kytos/pathfinder/v3/"

    def test_update_to_topology_success_case(self):
        """Test update topology method to success case."""
        topology = get_topology_mock()
        self.napp._update_to_topology(topology)
        assert self.napp._topology == topology

    def setting_shortest_path_mocked(self, mock_shortest_paths):
        """Set the primary elements needed to test the retrieving
        process of the shortest path under a mocked approach."""
        self.napp._topology = get_topology_mock()
        path = ["00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:02:1"]
        mock_shortest_paths.return_value = [path]
        return path

    async def test_shortest_path_response(self, monkeypatch):
        """Test shortest path."""
        self.napp.controller.loop = asyncio.get_running_loop()
        cost_mocked_value = 1
        mock_path_cost = MagicMock(return_value=cost_mocked_value)
        monkeypatch.setattr("napps.kytos.pathfinder.graph."
                            "KytosGraph._path_cost", mock_path_cost)
        mock_shortest_paths = MagicMock()
        monkeypatch.setattr("napps.kytos.pathfinder.graph."
                            "KytosGraph.k_shortest_paths", mock_shortest_paths)
        path = self.setting_shortest_path_mocked(mock_shortest_paths)
        data = {
            "source": "00:00:00:00:00:00:00:01:1",
            "destination": "00:00:00:00:00:00:00:02:1",
        }
        response = await self.api_client.post(self.endpoint, json=data)
        assert response.status_code == 200
        expected_response = {
            "paths": [{"hops": path, "cost": cost_mocked_value}]
        }
        assert response.json() == expected_response

    async def test_shortest_path_response_status_code(
        self, monkeypatch
    ):
        """Test shortest path."""
        self.napp.controller.loop = asyncio.get_running_loop()
        cost_mocked_value = 1
        mock_path_cost = MagicMock(return_value=cost_mocked_value)
        monkeypatch.setattr("napps.kytos.pathfinder.graph."
                            "KytosGraph._path_cost", mock_path_cost)
        mock_shortest_paths = MagicMock()
        monkeypatch.setattr("napps.kytos.pathfinder.graph."
                            "KytosGraph.k_shortest_paths", mock_shortest_paths)
        _ = self.setting_shortest_path_mocked(mock_shortest_paths)
        data = {
            "source": "00:00:00:00:00:00:00:01:1",
            "destination": "00:00:00:00:00:00:00:02:1",
        }
        response = await self.api_client.post(self.endpoint, json=data)
        assert response.status_code == 200

    async def setting_shortest_constrained_path_mocked(
        self, mock_constrained_k_shortest_paths
    ):
        """Set the primary elements needed to test the retrieving process
        of the shortest constrained path under a mocked approach."""
        source = "00:00:00:00:00:00:00:01:1"
        destination = "00:00:00:00:00:00:00:02:1"
        path = [source, destination]
        mandatory_metrics = {"ownership": "bob"}
        fle_metrics = {"delay": 30}
        metrics = {**mandatory_metrics, **fle_metrics}
        mock_constrained_k_shortest_paths.return_value = [
            {"hops": path, "metrics": metrics}
        ]
        data = {
            "source": "00:00:00:00:00:00:00:01:1",
            "destination": "00:00:00:00:00:00:00:02:1",
            "mandatory_metrics": {"ownership": "bob"},
            "flexible_metrics": {"delay": 30},
            "minimum_flexible_hits": 1,
        }
        response = await self.api_client.post(self.endpoint, json=data)

        return response, metrics, path

    async def test_shortest_constrained_path_response(
        self, monkeypatch
    ):
        """Test constrained flexible paths."""
        self.napp.controller.loop = asyncio.get_running_loop()
        mock_path_cost = MagicMock(return_value=1)
        monkeypatch.setattr("napps.kytos.pathfinder.graph."
                            "KytosGraph._path_cost", mock_path_cost)
        mock_k = MagicMock()
        monkeypatch.setattr("napps.kytos.pathfinder.graph."
                            "KytosGraph.constrained_k_shortest_paths", mock_k)
        (
            response,
            metrics,
            path,
        ) = await self.setting_shortest_constrained_path_mocked(mock_k)
        expected_response = [
            {"metrics": metrics, "hops": path, "cost": 1}
        ]
        assert response.json()["paths"][0] == expected_response[0]

    async def test_shortest_constrained_path_response_status_code(
        self, monkeypatch
    ):
        """Test constrained flexible paths."""
        self.napp.controller.loop = asyncio.get_running_loop()
        mock_path_cost = MagicMock(return_value=1)
        monkeypatch.setattr("napps.kytos.pathfinder.graph."
                            "KytosGraph._path_cost", mock_path_cost)
        mock_k = MagicMock()
        monkeypatch.setattr("napps.kytos.pathfinder.graph."
                            "KytosGraph.k_shortest_paths", mock_k)
        response, _, _ = await self.setting_shortest_constrained_path_mocked(
            mock_k
        )
        assert response.status_code == 200

    def test_filter_paths_le_cost_response(self):
        """Test filter paths."""
        self.napp._topology = get_topology_mock()
        paths = [
            {
                "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: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:04",
                    "00:00:00:00:00:00:00:04:1",
                ],
                "cost": 6,
            },
            {
                "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:04",
                    "00:00:00:00:00:00:00:04:1",
                ],
                "cost": 3,
            },
        ]
        filtered_paths = self.napp._filter_paths_le_cost(paths, 3)
        assert len(filtered_paths) == 1
        assert filtered_paths[0]["cost"] == 3

    def test_filter_paths_response_on_undesired(self):
        """Test filter paths."""
        self.napp._topology = get_topology_mock()
        edges = [
            (link.endpoint_a.id, link.endpoint_b.id)
            for link in self.napp._topology.links.values()
        ]
        self.napp.graph.graph.edges = edges

        undesired = ["1"]
        non_excluded_edges = self.napp._non_excluded_edges(undesired)
        assert non_excluded_edges == [
            ("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:03:1"),
            ("00:00:00:00:00:00:00:02:2", "00:00:00:00:00:00:00:03:2"),
        ]

        undesired = ["3"]
        non_excluded_edges = self.napp._non_excluded_edges(undesired)
        assert non_excluded_edges == [
            ('00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:02:1'),
            ("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:03:1"),
        ]

        undesired = ["1", "3"]
        non_excluded_edges = self.napp._non_excluded_edges(undesired)
        assert non_excluded_edges == [
            ('00:00:00:00:00:00:00:01:2', '00:00:00:00:00:00:00:03:1'),
        ]

        undesired = ["1", "2", "3"]
        non_excluded_edges = self.napp._non_excluded_edges(undesired)
        assert not non_excluded_edges

        undesired = []
        non_excluded_edges = self.napp._non_excluded_edges(undesired)
        assert non_excluded_edges == [
            ('00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:02:1'),
            ("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:03:1"),
            ("00:00:00:00:00:00:00:02:2", "00:00:00:00:00:00:00:03:2"),
        ]

        undesired = ["1", "2", "3"]
        self.napp._topology = None
        non_excluded_edges = self.napp._non_excluded_edges(undesired)
        assert not non_excluded_edges

    def setting_path(self):
        """Set the primary elements needed to test the topology
        update process under a "real-simulated" scenario."""
        topology = get_topology_with_metadata()
        self.napp._update_to_topology(topology)

    async def test_shortest_path(self):
        """Test shortest path."""
        self.napp.controller.loop = asyncio.get_running_loop()
        self.setting_path()
        source, destination = "User1", "User4"
        data = {"source": source, "destination": destination}
        response = await self.api_client.post(self.endpoint, json=data)

        for path in response.json()["paths"]:
            assert source == path["hops"][0]
            assert destination == path["hops"][-1]

    async def setting_shortest_constrained_path_exception(self, side_effect):
        """Set the primary elements needed to test the shortest
        constrained path behavior under exception actions."""
        self.setting_path()
        with patch(
            "napps.kytos.pathfinder.graph.KytosGraph."
            "constrained_k_shortest_paths",
            side_effect=side_effect,
        ):
            data = {
                "source": "00:00:00:00:00:00:00:01:1",
                "destination": "00:00:00:00:00:00:00:02:1",
                "mandatory_metrics": {"ownership": "bob"},
                "flexible_metrics": {"delay": 30},
                "minimum_flexible_hits": 1,
            }
            response = await self.api_client.post(self.endpoint, json=data)
        return response

    async def test_shortest_constrained_path_400_exception(self):
        """Test shortest path."""
        self.napp.controller.loop = asyncio.get_running_loop()
        res = await self.setting_shortest_constrained_path_exception(TypeError)
        assert res.status_code == 400

    async def test_use_latest_topology(
        self,
    ):
        """Test getting the latest topology"""
        next_topology = MagicMock()
        topology_napp = MagicMock()
        topology_napp.get_latest_topology.return_value = next_topology
        self.napp.controller.napps[("kytos", "topology")] = topology_napp

        self.napp.graph = MagicMock()

        self.napp._topology = None

        self.napp._get_latest_topology()

        assert self.napp._topology == next_topology
        self.napp.graph.update_topology.assert_called_with(next_topology)

    async def test_use_latest_topology_no_update(
        self,
    ):
        """Test getting the latest topology"""
        curr_topology = MagicMock()
        topology_napp = MagicMock()
        topology_napp.get_latest_topology.return_value = curr_topology
        self.napp.controller.napps[("kytos", "topology")] = topology_napp

        self.napp.graph = MagicMock()

        self.napp._topology = curr_topology

        self.napp._get_latest_topology()

        assert self.napp._topology == curr_topology
        self.napp.graph.update_topology.assert_not_called()

    async def test_use_latest_topology_exception(
        self,
    ):
        """Test failing to get the latest topology."""
        self.napp.graph = MagicMock()

        self.napp._topology = None

        self.napp._get_latest_topology()

        assert self.napp._topology is None
        self.napp.graph.update_topology.assert_not_called()

    @patch("napps.kytos.pathfinder.graph.graph.KytosGraph.update_topology")
    def test_get_latest_topology_retry(self, mock_update_topology):
        """Test retry from _get_latest_topology method."""
        self.napp._topology = None
        self.napp.controller.napps[("kytos", "topology")] = MagicMock()
        mock_update_topology.side_effect = LinkNotFound("")

        with pytest.raises(LinkNotFound):
            self.napp._get_latest_topology()

        assert self.napp._topology is None
        assert mock_update_topology.call_count == 3

        mock_update_topology.side_effect = None

        self.napp._get_latest_topology()
        assert self.napp._topology is not None
        assert mock_update_topology.call_count == 4

    @patch("napps.kytos.pathfinder.graph.graph.KytosGraph.update_topology")
    async def test_shortest_path_link_not_found(self, mock_update_topology):
        """Test shortest_path when a link is not found and returns 409."""
        self.napp.controller.loop = asyncio.get_running_loop()
        self.napp.controller.napps[("kytos", "topology")] = MagicMock()
        mock_update_topology.side_effect = LinkNotFound("")
        data = {
            "source": "00:00:00:00:00:00:00:01:1",
            "destination": "00:00:00:00:00:00:00:03:1",
            "spf_max_paths": 10
        }
        response = await self.api_client.post(self.endpoint, json=data)
        assert response.status_code == 409


1		"""Test Main methods."""
2	1	import asyncio
3	1	import pytest
4	1	from unittest.mock import MagicMock, patch
5
6	1	from kytos.lib.helpers import get_controller_mock, get_test_client
7
8		# pylint: disable=import-error
9	1	from napps.kytos.pathfinder.main import Main
10	1	from napps.kytos.pathfinder.exceptions import LinkNotFound
11	1	from tests.helpers import get_topology_mock, get_topology_with_metadata
12
13
14		# pylint: disable=protected-access
15	1	class TestMain:
16		"""Tests for the Main class."""
17
18	1	def setup_method(self):
19		"""Execute steps before each tests."""
20	1	self.controller = get_controller_mock()
21	1	self.napp = Main(self.controller)
22	1	self.api_client = get_test_client(self.controller, self.napp)
23	1	self.endpoint = "kytos/pathfinder/v3/"
24
25	1	def test_update_to_topology_success_case(self):
26		"""Test update topology method to success case."""
27	1	topology = get_topology_mock()
28	1	self.napp._update_to_topology(topology)
29	1	assert self.napp._topology == topology
30
31	1	def setting_shortest_path_mocked(self, mock_shortest_paths):
32		"""Set the primary elements needed to test the retrieving
33		process of the shortest path under a mocked approach."""
34	1	self.napp._topology = get_topology_mock()
35	1	path = ["00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:02:1"]
36	1	mock_shortest_paths.return_value = [path]
37	1	return path
38
39	1	async def test_shortest_path_response(self, monkeypatch):
40		"""Test shortest path."""
41	1	self.napp.controller.loop = asyncio.get_running_loop()
42	1	cost_mocked_value = 1
43	1	mock_path_cost = MagicMock(return_value=cost_mocked_value)
44	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
45		"KytosGraph._path_cost", mock_path_cost)
46	1	mock_shortest_paths = MagicMock()
47	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
48		"KytosGraph.k_shortest_paths", mock_shortest_paths)
49	1	path = self.setting_shortest_path_mocked(mock_shortest_paths)
50	1	data = {
51		"source": "00:00:00:00:00:00:00:01:1",
52		"destination": "00:00:00:00:00:00:00:02:1",
53		}
54	1	response = await self.api_client.post(self.endpoint, json=data)
55	1	assert response.status_code == 200
56	1	expected_response = {
57		"paths": [{"hops": path, "cost": cost_mocked_value}]
58		}
59	1	assert response.json() == expected_response
60
61	1	async def test_shortest_path_response_status_code(
62		self, monkeypatch
63		):
64		"""Test shortest path."""
65	1	self.napp.controller.loop = asyncio.get_running_loop()
66	1	cost_mocked_value = 1
67	1	mock_path_cost = MagicMock(return_value=cost_mocked_value)
68	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
69		"KytosGraph._path_cost", mock_path_cost)
70	1	mock_shortest_paths = MagicMock()
71	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
72		"KytosGraph.k_shortest_paths", mock_shortest_paths)
73	1	_ = self.setting_shortest_path_mocked(mock_shortest_paths)
74	1	data = {
75		"source": "00:00:00:00:00:00:00:01:1",
76		"destination": "00:00:00:00:00:00:00:02:1",
77		}
78	1	response = await self.api_client.post(self.endpoint, json=data)
79	1	assert response.status_code == 200
80
81	1	async def setting_shortest_constrained_path_mocked(
82		self, mock_constrained_k_shortest_paths
83		):
84		"""Set the primary elements needed to test the retrieving process
85		of the shortest constrained path under a mocked approach."""
86	1	source = "00:00:00:00:00:00:00:01:1"
87	1	destination = "00:00:00:00:00:00:00:02:1"
88	1	path = [source, destination]
89	1	mandatory_metrics = {"ownership": "bob"}
90	1	fle_metrics = {"delay": 30}
91	1	metrics = {mandatory_metrics, fle_metrics}
92	1	mock_constrained_k_shortest_paths.return_value = [
93		{"hops": path, "metrics": metrics}
94		]
95	1	data = {
96		"source": "00:00:00:00:00:00:00:01:1",
97		"destination": "00:00:00:00:00:00:00:02:1",
98		"mandatory_metrics": {"ownership": "bob"},
99		"flexible_metrics": {"delay": 30},
100		"minimum_flexible_hits": 1,
101		}
102	1	response = await self.api_client.post(self.endpoint, json=data)
103
104	1	return response, metrics, path
105
106	1	async def test_shortest_constrained_path_response(
107		self, monkeypatch
108		):
109		"""Test constrained flexible paths."""
110	1	self.napp.controller.loop = asyncio.get_running_loop()
111	1	mock_path_cost = MagicMock(return_value=1)
112	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
113		"KytosGraph._path_cost", mock_path_cost)
114	1	mock_k = MagicMock()
115	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
116		"KytosGraph.constrained_k_shortest_paths", mock_k)
117	1	(
118		response,
119		metrics,
120		path,
121		) = await self.setting_shortest_constrained_path_mocked(mock_k)
122	1	expected_response = [
123		{"metrics": metrics, "hops": path, "cost": 1}
124		]
125	1	assert response.json()["paths"][0] == expected_response[0]
126
127	1	async def test_shortest_constrained_path_response_status_code(
128		self, monkeypatch
129		):
130		"""Test constrained flexible paths."""
131	1	self.napp.controller.loop = asyncio.get_running_loop()
132	1	mock_path_cost = MagicMock(return_value=1)
133	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
134		"KytosGraph._path_cost", mock_path_cost)
135	1	mock_k = MagicMock()
136	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
137		"KytosGraph.k_shortest_paths", mock_k)
138	1	response, _, _ = await self.setting_shortest_constrained_path_mocked(
139		mock_k
140		)
141	1	assert response.status_code == 200
142
143	1	def test_filter_paths_le_cost_response(self):
144		"""Test filter paths."""
145	1	self.napp._topology = get_topology_mock()
146	1	paths = [
147		{
148		"hops": [
149		"00:00:00:00:00:00:00:01:1",
150		"00:00:00:00:00:00:00:01",
151		"00:00:00:00:00:00:00:02:1",
152		"00:00:00:00:00:00:00:02",
153		"00:00:00:00:00:00:00:02:2",
154		"00:00:00:00:00:00:00:04",
155		"00:00:00:00:00:00:00:04:1",
156		],
157		"cost": 6,
158		},
159		{
160		"hops": [
161		"00:00:00:00:00:00:00:01:1",
162		"00:00:00:00:00:00:00:01",
163		"00:00:00:00:00:00:00:04",
164		"00:00:00:00:00:00:00:04:1",
165		],
166		"cost": 3,
167		},
168		]
169	1	filtered_paths = self.napp._filter_paths_le_cost(paths, 3)
170	1	assert len(filtered_paths) == 1
171	1	assert filtered_paths[0]["cost"] == 3
172
173	1	def test_filter_paths_response_on_undesired(self):
174		"""Test filter paths."""
175	1	self.napp._topology = get_topology_mock()
176	1	edges = [
177		(link.endpoint_a.id, link.endpoint_b.id)
178		for link in self.napp._topology.links.values()
179		]
180	1	self.napp.graph.graph.edges = edges
181
182	1	undesired = ["1"]
183	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
184	1	assert non_excluded_edges == [
185		("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:03:1"),
186		("00:00:00:00:00:00:00:02:2", "00:00:00:00:00:00:00:03:2"),
187		]
188
189	1	undesired = ["3"]
190	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
191	1	assert non_excluded_edges == [
192		('00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:02:1'),
193		("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:03:1"),
194		]
195
196	1	undesired = ["1", "3"]
197	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
198	1	assert non_excluded_edges == [
199		('00:00:00:00:00:00:00:01:2', '00:00:00:00:00:00:00:03:1'),
200		]
201
202	1	undesired = ["1", "2", "3"]
203	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
204	1	assert not non_excluded_edges
205
206	1	undesired = []
207	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
208	1	assert non_excluded_edges == [
209		('00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:02:1'),
210		("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:03:1"),
211		("00:00:00:00:00:00:00:02:2", "00:00:00:00:00:00:00:03:2"),
212		]
213
214	1	undesired = ["1", "2", "3"]
215	1	self.napp._topology = None
216	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
217	1	assert not non_excluded_edges
218
219	1	def setting_path(self):
220		"""Set the primary elements needed to test the topology
221		update process under a "real-simulated" scenario."""
222	1	topology = get_topology_with_metadata()
223	1	self.napp._update_to_topology(topology)
224
225	1	async def test_shortest_path(self):
226		"""Test shortest path."""
227	1	self.napp.controller.loop = asyncio.get_running_loop()
228	1	self.setting_path()
229	1	source, destination = "User1", "User4"
230	1	data = {"source": source, "destination": destination}
231	1	response = await self.api_client.post(self.endpoint, json=data)
232
233	1	for path in response.json()["paths"]:
234		assert source == path["hops"][0]
235		assert destination == path["hops"][-1]
236
237	1	async def setting_shortest_constrained_path_exception(self, side_effect):
238		"""Set the primary elements needed to test the shortest
239		constrained path behavior under exception actions."""
240	1	self.setting_path()
241	1	with patch(
242		"napps.kytos.pathfinder.graph.KytosGraph."
243		"constrained_k_shortest_paths",
244		side_effect=side_effect,
245		):
246	1	data = {
247		"source": "00:00:00:00:00:00:00:01:1",
248		"destination": "00:00:00:00:00:00:00:02:1",
249		"mandatory_metrics": {"ownership": "bob"},
250		"flexible_metrics": {"delay": 30},
251		"minimum_flexible_hits": 1,
252		}
253	1	response = await self.api_client.post(self.endpoint, json=data)
254		return response
255
256	1	async def test_shortest_constrained_path_400_exception(self):
257		"""Test shortest path."""
258	1	self.napp.controller.loop = asyncio.get_running_loop()
259	1	res = await self.setting_shortest_constrained_path_exception(TypeError)
260		assert res.status_code == 400
261
262	1	async def test_use_latest_topology(
263		self,
264		):
265		"""Test getting the latest topology"""
266	1	next_topology = MagicMock()
267	1	topology_napp = MagicMock()
268	1	topology_napp.get_latest_topology.return_value = next_topology
269	1	self.napp.controller.napps[("kytos", "topology")] = topology_napp
270
271	1	self.napp.graph = MagicMock()
272
273	1	self.napp._topology = None
274
275	1	self.napp._get_latest_topology()
276
277	1	assert self.napp._topology == next_topology
278	1	self.napp.graph.update_topology.assert_called_with(next_topology)
279
280	1	async def test_use_latest_topology_no_update(
281		self,
282		):
283		"""Test getting the latest topology"""
284	1	curr_topology = MagicMock()
285	1	topology_napp = MagicMock()
286	1	topology_napp.get_latest_topology.return_value = curr_topology
287	1	self.napp.controller.napps[("kytos", "topology")] = topology_napp
288
289	1	self.napp.graph = MagicMock()
290
291	1	self.napp._topology = curr_topology
292
293	1	self.napp._get_latest_topology()
294
295	1	assert self.napp._topology == curr_topology
296	1	self.napp.graph.update_topology.assert_not_called()
297
298	1	async def test_use_latest_topology_exception(
299		self,
300		):
301		"""Test failing to get the latest topology."""
302	1	self.napp.graph = MagicMock()
303
304	1	self.napp._topology = None
305
306	1	self.napp._get_latest_topology()
307
308	1	assert self.napp._topology is None
309	1	self.napp.graph.update_topology.assert_not_called()
310
311	1	@patch("napps.kytos.pathfinder.graph.graph.KytosGraph.update_topology")
312	1	def test_get_latest_topology_retry(self, mock_update_topology):
313		"""Test retry from _get_latest_topology method."""
314	1	self.napp._topology = None
315	1	self.napp.controller.napps[("kytos", "topology")] = MagicMock()
316	1	mock_update_topology.side_effect = LinkNotFound("")
317
318	1	with pytest.raises(LinkNotFound):
319	1	self.napp._get_latest_topology()
320
321	1	assert self.napp._topology is None
322	1	assert mock_update_topology.call_count == 3
323
324	1	mock_update_topology.side_effect = None
325
326	1	self.napp._get_latest_topology()
327	1	assert self.napp._topology is not None
328	1	assert mock_update_topology.call_count == 4
329
330	1	@patch("napps.kytos.pathfinder.graph.graph.KytosGraph.update_topology")
331	1	async def test_shortest_path_link_not_found(self, mock_update_topology):
332		"""Test shortest_path when a link is not found and returns 409."""
333	1	self.napp.controller.loop = asyncio.get_running_loop()
334	1	self.napp.controller.napps[("kytos", "topology")] = MagicMock()
335	1	mock_update_topology.side_effect = LinkNotFound("")
336	1	data = {
337		"source": "00:00:00:00:00:00:00:01:1",
338		"destination": "00:00:00:00:00:00:00:03:1",
339		"spf_max_paths": 10
340		}
341	1	response = await self.api_client.post(self.endpoint, json=data)
342		assert response.status_code == 409
343

kytos-ng / pathfinder

Push — master ( bdb3e4...e1fc2e )

TestMain.test_shortest_path() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like