build.tests.unit.test_main.TestMain.test_update_links_changed() - Code Metrics - Inspection of "Fix/topo events 2" - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#32)

by Vinicius

created 2022-11-02 21:23 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 %

Importance

Changes

Metric	Value
cc	1
eloc	9
nop	1
dl	0
loc	11
rs	9.95
c	0
b	0
f	0

"""Test Main methods."""

from unittest import TestCase
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(TestCase):
    """Tests for the Main class."""

    def setUp(self):
        """Execute steps before each tests."""
        self.napp = Main(get_controller_mock())

    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)

        self.assertEqual(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)

        self.assertIsNone(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]

        api = get_test_client(self.napp.controller, self.napp)

        return api, path

    @patch("napps.kytos.pathfinder.graph.KytosGraph._path_cost")
    @patch("napps.kytos.pathfinder.graph.KytosGraph.k_shortest_paths")
    def test_shortest_path_response(self, mock_shortest_paths, path_cost):
        """Test shortest path."""
        cost_mocked_value = 1
        path_cost.return_value = cost_mocked_value
        api, path = self.setting_shortest_path_mocked(mock_shortest_paths)
        url = "http://127.0.0.1:8181/api/kytos/pathfinder/v2"
        data = {
            "source": "00:00:00:00:00:00:00:01:1",
            "destination": "00:00:00:00:00:00:00:02:1",
            "desired_links": ["1"],
            "undesired_links": None,
        }
        response = api.open(url, method="POST", json=data)

        expected_response = {
            "paths": [{"hops": path, "cost": cost_mocked_value}]
        }
        self.assertEqual(response.json, expected_response)

    @patch("napps.kytos.pathfinder.graph.KytosGraph._path_cost")
    @patch("napps.kytos.pathfinder.graph.KytosGraph.k_shortest_paths")
    def test_shortest_path_response_status_code(
        self, mock_shortest_paths, path_cost
    ):
        """Test shortest path."""
        path_cost.return_value = 1
        api, _ = self.setting_shortest_path_mocked(mock_shortest_paths)
        url = "http://127.0.0.1:8181/api/kytos/pathfinder/v2"
        data = {
            "source": "00:00:00:00:00:00:00:01:1",
            "destination": "00:00:00:00:00:00:00:02:1",
            "desired_links": ["1"],
            "undesired_links": None,
        }
        response = api.open(url, method="POST", json=data)

        self.assertEqual(response.status_code, 200)

    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]
        base_metrics = {"ownership": "bob"}
        fle_metrics = {"delay": 30}
        metrics = {**base_metrics, **fle_metrics}
        mock_constrained_k_shortest_paths.return_value = [
            {"hops": [path], "metrics": metrics}
        ]

        api = get_test_client(self.napp.controller, self.napp)
        url = "http://127.0.0.1:8181/api/kytos/pathfinder/v2/"
        data = {
            "source": "00:00:00:00:00:00:00:01:1",
            "destination": "00:00:00:00:00:00:00:02:1",
            "base_metrics": {"ownership": "bob"},
            "flexible_metrics": {"delay": 30},
            "minimum_flexible_hits": 1,
        }
        response = api.open(url, method="POST", json=data)

        return response, metrics, path

    @patch("napps.kytos.pathfinder.graph.KytosGraph._path_cost")
    @patch(
        "napps.kytos.pathfinder.graph.KytosGraph.constrained_k_shortest_paths",
        autospec=True,
    )
    def test_shortest_constrained_path_response(
        self, mock_constrained_k_shortest_paths, path_cost
    ):
        """Test constrained flexible paths."""
        cost_mocked_value = 1
        path_cost.return_value = cost_mocked_value
        (
            response,
            metrics,
            path,
        ) = self.setting_shortest_constrained_path_mocked(
            mock_constrained_k_shortest_paths
        )
        expected_response = [
            {"metrics": metrics, "hops": [path], "cost": cost_mocked_value}
        ]

        self.assertDictEqual(response.json["paths"][0], expected_response[0])

    @patch("napps.kytos.pathfinder.graph.KytosGraph._path_cost")
    @patch(
        "napps.kytos.pathfinder.graph.KytosGraph.constrained_k_shortest_paths",
        autospec=True,
    )
    def test_shortest_constrained_path_response_status_code(
        self, mock_constrained_k_shortest_paths, path_cost
    ):
        """Test constrained flexible paths."""
        path_cost.return_value = 1
        response, _, _ = self.setting_shortest_constrained_path_mocked(
            mock_constrained_k_shortest_paths
        )

        self.assertEqual(response.status_code, 200)

    def test_filter_paths_response_on_desired(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:02:1",
                    "00:00:00:00:00:00:00:02:2",
                    "00:00:00:00:00:00:00:03:2",
                ]
            },
            {
                "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,
            },
        ]
        desired = ["1", "3"]

        for link in desired:
            assert self.napp._topology.links[link]
        filtered_paths = self.napp._filter_paths_desired_links(paths, desired)
        assert filtered_paths == [paths[0]]

        filtered_paths = self.napp._filter_paths_desired_links(paths, ["1", "2"])
        assert not filtered_paths

        filtered_paths = self.napp._filter_paths_desired_links(paths, ["inexistent_id"])
        assert not filtered_paths

    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()
        paths = [
            {
                "hops": [
                    "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:02:2",
                    "00:00:00:00:00:00:00:03:2",
                ]
            }
        ]

        undesired = ["1"]
        filtered_paths = self.napp._filter_paths_undesired_links(paths, undesired)
        assert not filtered_paths

        undesired = ["3"]
        filtered_paths = self.napp._filter_paths_undesired_links(paths, undesired)
        assert not filtered_paths

        undesired = ["1", "3"]
        filtered_paths = self.napp._filter_paths_undesired_links(paths, undesired)
        assert not filtered_paths

        filtered_paths = self.napp._filter_paths_undesired_links(paths, ["none"])
        assert filtered_paths == paths

    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)

    def test_update_links_changed(self):
        """Test update_links_metadata_changed."""
        self.napp.graph.update_link_metadata = MagicMock()
        self.napp.controller.buffers.app.put = MagicMock()
        event = KytosEvent(
            name="kytos.topology.links.metadata.added",
            content={"link": MagicMock(), "metadata": {}}
        )
        self.napp.update_links_metadata_changed(event)
        assert self.napp.graph.update_link_metadata.call_count == 1
        assert self.napp.controller.buffers.app.put.call_count == 0

    def test_update_links_changed_key_error(self):
        """Test update_links_metadata_changed key_error."""
        self.napp.graph.update_link_metadata = MagicMock()
        self.napp.controller.buffers.app.put = MagicMock()
        event = KytosEvent(
            name="kytos.topology.links.metadata.added",
            content={"link": MagicMock()}
        )
        self.napp.update_links_metadata_changed(event)
        assert self.napp.graph.update_link_metadata.call_count == 1
        assert self.napp.controller.buffers.app.put.call_count == 1

    def test_shortest_path(self):
        """Test shortest path."""
        self.setting_path()

        api = get_test_client(self.napp.controller, self.napp)
        url = "http://127.0.0.1:8181/api/kytos/pathfinder/v2/"

        source, destination = "User1", "User4"
        data = {"source": source, "destination": destination}

        response = api.open(url, method="POST", json=data)

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

    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()
        api = get_test_client(self.napp.controller, self.napp)

        with patch(
            "napps.kytos.pathfinder.graph.KytosGraph."
            "constrained_k_shortest_paths",
            side_effect=side_effect,
        ):
            url = "http://127.0.0.1:8181/api/kytos/pathfinder/v2/"

            data = {
                "source": "00:00:00:00:00:00:00:01:1",
                "destination": "00:00:00:00:00:00:00:02:1",
                "base_metrics": {"ownership": "bob"},
                "flexible_metrics": {"delay": 30},
                "minimum_flexible_hits": 1,
            }

            response = api.open(url, method="POST", json=data)

        return response

    def test_shortest_constrained_path_400_exception(self):
        """Test shortest path."""
        response = self.setting_shortest_constrained_path_exception(TypeError)

        self.assertEqual(response.status_code, 400)


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

kytos-ng / pathfinder

Pull Request — master (#32)

TestMain.test_update_links_changed() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like