build.tests.unit.test_main.TestMain.test_update_links_changed() - Code Metrics - Inspection of "fix: replaced `kytos/topology.links.metadata.(add..." - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#55)

by Vinicius

created 2023-07-06 20:22 UTC

TestMain.test_update_links_changed() 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	1

Importance

Changes

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

"""Test Main methods."""

from unittest.mock import MagicMock, patch
from datetime import timedelta

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_events_out_of_order(self):
        """Test update topology events out of order.

        If a subsequent older event is sent, then the topology
        shouldn't get updated.
        """
        topology = get_topology_mock()
        assert self.napp._topology_updated_at is None
        first_event = KytosEvent(
            name="kytos.topology.updated", content={"topology": topology}
        )
        self.napp.update_topology(first_event)
        assert self.napp._topology_updated_at == first_event.timestamp
        assert self.napp._topology == topology

        second_topology = get_topology_mock()
        second_event = KytosEvent(
            name="kytos.topology.updated", content={"topology": second_topology}
        )
        second_event.timestamp = first_event.timestamp - timedelta(seconds=10)
        self.napp.update_topology(second_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, event_loop):
        """Test shortest path."""
        self.napp.controller.loop = event_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, event_loop
    ):
        """Test shortest path."""
        self.napp.controller.loop = event_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, event_loop
    ):
        """Test constrained flexible paths."""
        self.napp.controller.loop = event_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, event_loop
    ):
        """Test constrained flexible paths."""
        self.napp.controller.loop = event_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, event_loop):
        """Test shortest path."""
        self.napp.controller.loop = event_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, event_loop):
        """Test shortest path."""
        self.napp.controller.loop = event_loop
        res = await self.setting_shortest_constrained_path_exception(TypeError)
        assert res.status_code == 400


1		"""Test Main methods."""
2
3	1	from unittest.mock import MagicMock, patch
4	1	from datetime import timedelta
5
6	1	from kytos.core.events import KytosEvent
7	1	from kytos.lib.helpers import get_controller_mock, get_test_client
8
9		# pylint: disable=import-error
10	1	from napps.kytos.pathfinder.main import Main
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_topology_success_case(self):
26		"""Test update topology method to success case."""
27	1	topology = get_topology_mock()
28	1	event = KytosEvent(
29		name="kytos.topology.updated", content={"topology": topology}
30		)
31	1	self.napp.update_topology(event)
32	1	assert self.napp._topology == topology
33
34	1	def test_update_topology_events_out_of_order(self):
35		"""Test update topology events out of order.
36
37		If a subsequent older event is sent, then the topology
38		shouldn't get updated.
39		"""
40	1	topology = get_topology_mock()
41	1	assert self.napp._topology_updated_at is None
42	1	first_event = KytosEvent(
43		name="kytos.topology.updated", content={"topology": topology}
44		)
45	1	self.napp.update_topology(first_event)
46	1	assert self.napp._topology_updated_at == first_event.timestamp
47	1	assert self.napp._topology == topology
48
49	1	second_topology = get_topology_mock()
50	1	second_event = KytosEvent(
51		name="kytos.topology.updated", content={"topology": second_topology}
52		)
53	1	second_event.timestamp = first_event.timestamp - timedelta(seconds=10)
54	1	self.napp.update_topology(second_event)
55	1	assert self.napp._topology == topology
56
57	1	def test_update_topology_failure_case(self):
58		"""Test update topology method to failure case."""
59	1	event = KytosEvent(name="kytos.topology.updated")
60	1	self.napp.update_topology(event)
61	1	assert not self.napp._topology
62
63	1	def setting_shortest_path_mocked(self, mock_shortest_paths):
64		"""Set the primary elements needed to test the retrieving
65		process of the shortest path under a mocked approach."""
66	1	self.napp._topology = get_topology_mock()
67	1	path = ["00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:02:1"]
68	1	mock_shortest_paths.return_value = [path]
69	1	return path
70
71	1	async def test_shortest_path_response(self, monkeypatch, event_loop):
72		"""Test shortest path."""
73	1	self.napp.controller.loop = event_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	path = 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	1	expected_response = {
89		"paths": [{"hops": path, "cost": cost_mocked_value}]
90		}
91	1	assert response.json() == expected_response
92
93	1	async def test_shortest_path_response_status_code(
94		self, monkeypatch, event_loop
95		):
96		"""Test shortest path."""
97	1	self.napp.controller.loop = event_loop
98	1	cost_mocked_value = 1
99	1	mock_path_cost = MagicMock(return_value=cost_mocked_value)
100	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
101		"KytosGraph._path_cost", mock_path_cost)
102	1	mock_shortest_paths = MagicMock()
103	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
104		"KytosGraph.k_shortest_paths", mock_shortest_paths)
105	1	_ = self.setting_shortest_path_mocked(mock_shortest_paths)
106	1	data = {
107		"source": "00:00:00:00:00:00:00:01:1",
108		"destination": "00:00:00:00:00:00:00:02:1",
109		}
110	1	response = await self.api_client.post(self.endpoint, json=data)
111	1	assert response.status_code == 200
112
113	1	async def setting_shortest_constrained_path_mocked(
114		self, mock_constrained_k_shortest_paths
115		):
116		"""Set the primary elements needed to test the retrieving process
117		of the shortest constrained path under a mocked approach."""
118	1	source = "00:00:00:00:00:00:00:01:1"
119	1	destination = "00:00:00:00:00:00:00:02:1"
120	1	path = [source, destination]
121	1	mandatory_metrics = {"ownership": "bob"}
122	1	fle_metrics = {"delay": 30}
123	1	metrics = {mandatory_metrics, fle_metrics}
124	1	mock_constrained_k_shortest_paths.return_value = [
125		{"hops": path, "metrics": metrics}
126		]
127	1	data = {
128		"source": "00:00:00:00:00:00:00:01:1",
129		"destination": "00:00:00:00:00:00:00:02:1",
130		"mandatory_metrics": {"ownership": "bob"},
131		"flexible_metrics": {"delay": 30},
132		"minimum_flexible_hits": 1,
133		}
134	1	response = await self.api_client.post(self.endpoint, json=data)
135
136	1	return response, metrics, path
137
138	1	async def test_shortest_constrained_path_response(
139		self, monkeypatch, event_loop
140		):
141		"""Test constrained flexible paths."""
142	1	self.napp.controller.loop = event_loop
143	1	mock_path_cost = MagicMock(return_value=1)
144	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
145		"KytosGraph._path_cost", mock_path_cost)
146	1	mock_k = MagicMock()
147	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
148		"KytosGraph.constrained_k_shortest_paths", mock_k)
149	1	(
150		response,
151		metrics,
152		path,
153		) = await self.setting_shortest_constrained_path_mocked(mock_k)
154	1	expected_response = [
155		{"metrics": metrics, "hops": path, "cost": 1}
156		]
157	1	assert response.json()["paths"][0] == expected_response[0]
158
159	1	async def test_shortest_constrained_path_response_status_code(
160		self, monkeypatch, event_loop
161		):
162		"""Test constrained flexible paths."""
163	1	self.napp.controller.loop = event_loop
164	1	mock_path_cost = MagicMock(return_value=1)
165	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
166		"KytosGraph._path_cost", mock_path_cost)
167	1	mock_k = MagicMock()
168	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
169		"KytosGraph.k_shortest_paths", mock_k)
170	1	response, _, _ = await self.setting_shortest_constrained_path_mocked(
171		mock_k
172		)
173	1	assert response.status_code == 200
174
175	1	def test_filter_paths_le_cost_response(self):
176		"""Test filter paths."""
177	1	self.napp._topology = get_topology_mock()
178	1	paths = [
179		{
180		"hops": [
181		"00:00:00:00:00:00:00:01:1",
182		"00:00:00:00:00:00:00:01",
183		"00:00:00:00:00:00:00:02:1",
184		"00:00:00:00:00:00:00:02",
185		"00:00:00:00:00:00:00:02:2",
186		"00:00:00:00:00:00:00:04",
187		"00:00:00:00:00:00:00:04:1",
188		],
189		"cost": 6,
190		},
191		{
192		"hops": [
193		"00:00:00:00:00:00:00:01:1",
194		"00:00:00:00:00:00:00:01",
195		"00:00:00:00:00:00:00:04",
196		"00:00:00:00:00:00:00:04:1",
197		],
198		"cost": 3,
199		},
200		]
201	1	filtered_paths = self.napp._filter_paths_le_cost(paths, 3)
202	1	assert len(filtered_paths) == 1
203	1	assert filtered_paths[0]["cost"] == 3
204
205	1	def test_filter_paths_response_on_undesired(self):
206		"""Test filter paths."""
207	1	self.napp._topology = get_topology_mock()
208	1	edges = [
209		(link.endpoint_a.id, link.endpoint_b.id)
210		for link in self.napp._topology.links.values()
211		]
212	1	self.napp.graph.graph.edges = edges
213
214	1	undesired = ["1"]
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:2", "00:00:00:00:00:00:00:03:1"),
218		("00:00:00:00:00:00:00:02:2", "00:00:00:00:00:00:00:03:2"),
219		]
220
221	1	undesired = ["3"]
222	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
223	1	assert non_excluded_edges == [
224		('00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:02:1'),
225		("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:03:1"),
226		]
227
228	1	undesired = ["1", "3"]
229	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
230	1	assert non_excluded_edges == [
231		('00:00:00:00:00:00:00:01:2', '00:00:00:00:00:00:00:03:1'),
232		]
233
234	1	undesired = ["1", "2", "3"]
235	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
236	1	assert not non_excluded_edges
237
238	1	undesired = []
239	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
240	1	assert non_excluded_edges == [
241		('00:00:00:00:00:00:00:01:1', '00:00:00:00:00:00:00:02:1'),
242		("00:00:00:00:00:00:00:01:2", "00:00:00:00:00:00:00:03:1"),
243		("00:00:00:00:00:00:00:02:2", "00:00:00:00:00:00:00:03:2"),
244		]
245
246	1	undesired = ["1", "2", "3"]
247	1	self.napp._topology = None
248	1	non_excluded_edges = self.napp._non_excluded_edges(undesired)
249	1	assert not non_excluded_edges
250
251	1	def setting_path(self):
252		"""Set the primary elements needed to test the topology
253		update process under a "real-simulated" scenario."""
254	1	topology = get_topology_with_metadata()
255	1	event = KytosEvent(
256		name="kytos.topology.updated", content={"topology": topology}
257		)
258	1	self.napp.update_topology(event)
259
260	1	async def test_shortest_path(self, event_loop):
261		"""Test shortest path."""
262	1	self.napp.controller.loop = event_loop
263	1	self.setting_path()
264	1	source, destination = "User1", "User4"
265	1	data = {"source": source, "destination": destination}
266	1	response = await self.api_client.post(self.endpoint, json=data)
267
268	1	for path in response.json()["paths"]:
269		assert source == path["hops"][0]
270		assert destination == path["hops"][-1]
271
272	1	async def setting_shortest_constrained_path_exception(self, side_effect):
273		"""Set the primary elements needed to test the shortest
274		constrained path behavior under exception actions."""
275	1	self.setting_path()
276	1	with patch(
277		"napps.kytos.pathfinder.graph.KytosGraph."
278		"constrained_k_shortest_paths",
279		side_effect=side_effect,
280		):
281	1	data = {
282		"source": "00:00:00:00:00:00:00:01:1",
283		"destination": "00:00:00:00:00:00:00:02:1",
284		"mandatory_metrics": {"ownership": "bob"},
285		"flexible_metrics": {"delay": 30},
286		"minimum_flexible_hits": 1,
287		}
288	1	response = await self.api_client.post(self.endpoint, json=data)
289	1	return response
290
291	1	async def test_shortest_constrained_path_400_exception(self, event_loop):
292		"""Test shortest path."""
293	1	self.napp.controller.loop = event_loop
294	1	res = await self.setting_shortest_constrained_path_exception(TypeError)
295		assert res.status_code == 400
296

kytos-ng / pathfinder

Pull Request — master (#55)

TestMain.test_update_links_changed() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like