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

Passed

Push — master ( 813874...7da87a )

unknown

created 2025-06-12 17:19 UTC

TestMain.test_use_latest_topology() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	17
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	11
nop	1
dl	0
loc	17
ccs	10
cts	10
cp	1
crap	1
rs	9.85
c	0
b	0
f	0

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

from kytos.core.events import KytosEvent
from kytos.lib.helpers import get_controller_mock, get_test_client

# pylint: disable=import-error
from napps.kytos.pathfinder.main import Main
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_topology_success_case(self):
        """Test update topology method to success case."""
        topology = get_topology_mock()
        event = KytosEvent(
            name="kytos.topology.updated", content={"topology": topology}
        )
        self.napp.update_topology(event)
        assert self.napp._topology == topology

    def test_update_topology_failure_case(self):
        """Test update topology method to failure case."""
        event = KytosEvent(name="kytos.topology.updated")
        self.napp.update_topology(event)
        assert not self.napp._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()
        event = KytosEvent(
            name="kytos.topology.updated", content={"topology": topology}
        )
        self.napp.update_topology(event)

    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()


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

kytos-ng / pathfinder

Push — master ( 813874...7da87a )

TestMain.test_use_latest_topology() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like