build.tests.unit.test_main.TestMain.setting_shortest_path_mocked() - Code Metrics - Inspection of "chore: `@rest` endpoints are now run by `starlette..." - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#45)

by Vinicius

created 2023-04-21 14:17 UTC

TestMain.setting_shortest_path_mocked() A

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	7
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	5
nop	2
dl	0
loc	7
rs	10
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/v2/"

    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):
        """Test shortest path."""
        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",
            "desired_links": ["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."""
        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",
            "desired_links": ["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."""
        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."""
        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_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)

    async 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

    async def test_update_links_changed_out_of_order(self):
        """Test update_links_metadata_changed out of order."""
        self.napp.graph.update_link_metadata = MagicMock()
        self.napp.controller.buffers.app.put = MagicMock()
        link = MagicMock(id="1")
        assert link.id not in self.napp._links_updated_at
        event = KytosEvent(
            name="kytos.topology.links.metadata.added",
            content={"link": link, "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
        assert self.napp._links_updated_at[link.id] == event.timestamp

        second_event = KytosEvent(
            name="kytos.topology.links.metadata.added",
            content={"link": link, "metadata": {}}
        )
        second_event.timestamp = event.timestamp - timedelta(seconds=10)
        self.napp.update_links_metadata_changed(second_event)
        assert self.napp.graph.update_link_metadata.call_count == 1
        assert self.napp.controller.buffers.app.put.call_count == 0
        assert self.napp._links_updated_at[link.id] == event.timestamp

    async 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

    async def test_shortest_path(self):
        """Test shortest path."""
        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."""
        res = await self.setting_shortest_constrained_path_exception(TypeError)
        assert res.status_code == 400


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

kytos-ng / pathfinder

Pull Request — master (#45)

TestMain.setting_shortest_path_mocked() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like