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

Passed

Pull Request — master (#45)

by Vinicius

created 2023-04-28 03:02 UTC

build.tests.unit.test_main.TestMain.setUp() A

↳ Parent: build.tests.unit.test_main.TestMain.setup_method()

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	3
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	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/v2/"
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):
72		"""Test shortest path."""
73	1	cost_mocked_value = 1
74	1	mock_path_cost = MagicMock(return_value=cost_mocked_value)
75	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
76		"KytosGraph._path_cost", mock_path_cost)
77	1	mock_shortest_paths = MagicMock()
78	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
79		"KytosGraph.k_shortest_paths", mock_shortest_paths)
80	1	path = self.setting_shortest_path_mocked(mock_shortest_paths)
81	1	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	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(self, monkeypatch):
94		"""Test shortest path."""
95	1	cost_mocked_value = 1
96	1	mock_path_cost = MagicMock(return_value=cost_mocked_value)
97	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
98		"KytosGraph._path_cost", mock_path_cost)
99	1	mock_shortest_paths = MagicMock()
100	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
101		"KytosGraph.k_shortest_paths", mock_shortest_paths)
102	1	_ = self.setting_shortest_path_mocked(mock_shortest_paths)
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		"desired_links": ["1"]
107		}
108	1	response = await self.api_client.post(self.endpoint, json=data)
109	1	assert response.status_code == 200
110
111	1	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	1	source = "00:00:00:00:00:00:00:01:1"
117	1	destination = "00:00:00:00:00:00:00:02:1"
118	1	path = [source, destination]
119	1	mandatory_metrics = {"ownership": "bob"}
120	1	fle_metrics = {"delay": 30}
121	1	metrics = {mandatory_metrics, fle_metrics}
122	1	mock_constrained_k_shortest_paths.return_value = [
123		{"hops": path, "metrics": metrics}
124		]
125	1	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	1	response = await self.api_client.post(self.endpoint, json=data)
133
134	1	return response, metrics, path
135
136	1	async def test_shortest_constrained_path_response(self, monkeypatch):
137		"""Test constrained flexible paths."""
138	1	mock_path_cost = MagicMock(return_value=1)
139	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
140		"KytosGraph._path_cost", mock_path_cost)
141	1	mock_k = MagicMock()
142	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
143		"KytosGraph.constrained_k_shortest_paths", mock_k)
144	1	(
145		response,
146		metrics,
147		path,
148		) = await self.setting_shortest_constrained_path_mocked(mock_k)
149	1	expected_response = [
150		{"metrics": metrics, "hops": path, "cost": 1}
151		]
152	1	assert response.json()["paths"][0] == expected_response[0]
153
154	1	async def test_shortest_constrained_path_response_status_code(self, monkeypatch):
155		"""Test constrained flexible paths."""
156	1	mock_path_cost = MagicMock(return_value=1)
157	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
158		"KytosGraph._path_cost", mock_path_cost)
159	1	mock_k = MagicMock()
160	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
161		"KytosGraph.k_shortest_paths", mock_k)
162	1	response, _, _ = await self.setting_shortest_constrained_path_mocked(
163		mock_k
164		)
165	1	assert response.status_code == 200
166
167	1	def test_filter_paths_response_on_desired(self):
168		"""Test filter paths."""
169	1	self.napp._topology = get_topology_mock()
170	1	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	1	desired = ["1", "3"]
190
191	1	for link in desired:
192	1	assert self.napp._topology.links[link]
193	1	filtered_paths = self.napp._filter_paths_desired_links(paths, desired)
194	1	assert filtered_paths == [paths[0]]
195
196	1	filtered_paths = self.napp._filter_paths_desired_links(paths, ["1", "2"])
197	1	assert not filtered_paths
198
199	1	filtered_paths = self.napp._filter_paths_desired_links(paths, ["inexistent_id"])
200	1	assert not filtered_paths
201
202	1	def test_filter_paths_le_cost_response(self):
203		"""Test filter paths."""
204	1	self.napp._topology = get_topology_mock()
205	1	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	1	filtered_paths = self.napp._filter_paths_le_cost(paths, 3)
229	1	assert len(filtered_paths) == 1
230	1	assert filtered_paths[0]["cost"] == 3
231
232	1	def test_filter_paths_response_on_undesired(self):
233		"""Test filter paths."""
234	1	self.napp._topology = get_topology_mock()
235	1	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	1	undesired = ["1"]
247	1	filtered_paths = self.napp._filter_paths_undesired_links(paths, undesired)
248	1	assert not filtered_paths
249
250	1	undesired = ["3"]
251	1	filtered_paths = self.napp._filter_paths_undesired_links(paths, undesired)
252	1	assert not filtered_paths
253
254	1	undesired = ["1", "3"]
255	1	filtered_paths = self.napp._filter_paths_undesired_links(paths, undesired)
256	1	assert not filtered_paths
257
258	1	filtered_paths = self.napp._filter_paths_undesired_links(paths, ["none"])
259	1	assert filtered_paths == paths
260
261	1	def setting_path(self):
262		"""Set the primary elements needed to test the topology
263		update process under a "real-simulated" scenario."""
264	1	topology = get_topology_with_metadata()
265	1	event = KytosEvent(
266		name="kytos.topology.updated", content={"topology": topology}
267		)
268	1	self.napp.update_topology(event)
269
270	1	async def test_update_links_changed(self):
271		"""Test update_links_metadata_changed."""
272	1	self.napp.graph.update_link_metadata = MagicMock()
273	1	self.napp.controller.buffers.app.put = MagicMock()
274	1	event = KytosEvent(
275		name="kytos.topology.links.metadata.added",
276		content={"link": MagicMock(), "metadata": {}}
277		)
278	1	self.napp.update_links_metadata_changed(event)
279	1	assert self.napp.graph.update_link_metadata.call_count == 1
280	1	assert self.napp.controller.buffers.app.put.call_count == 0
281
282	1	async def test_update_links_changed_out_of_order(self):
283		"""Test update_links_metadata_changed out of order."""
284	1	self.napp.graph.update_link_metadata = MagicMock()
285	1	self.napp.controller.buffers.app.put = MagicMock()
286	1	link = MagicMock(id="1")
287	1	assert link.id not in self.napp._links_updated_at
288	1	event = KytosEvent(
289		name="kytos.topology.links.metadata.added",
290		content={"link": link, "metadata": {}}
291		)
292	1	self.napp.update_links_metadata_changed(event)
293	1	assert self.napp.graph.update_link_metadata.call_count == 1
294	1	assert self.napp.controller.buffers.app.put.call_count == 0
295	1	assert self.napp._links_updated_at[link.id] == event.timestamp
296
297	1	second_event = KytosEvent(
298		name="kytos.topology.links.metadata.added",
299		content={"link": link, "metadata": {}}
300		)
301	1	second_event.timestamp = event.timestamp - timedelta(seconds=10)
302	1	self.napp.update_links_metadata_changed(second_event)
303	1	assert self.napp.graph.update_link_metadata.call_count == 1
304	1	assert self.napp.controller.buffers.app.put.call_count == 0
305	1	assert self.napp._links_updated_at[link.id] == event.timestamp
306
307	1	async def test_update_links_changed_key_error(self):
308		"""Test update_links_metadata_changed key_error."""
309	1	self.napp.graph.update_link_metadata = MagicMock()
310	1	self.napp.controller.buffers.app.put = MagicMock()
311	1	event = KytosEvent(
312		name="kytos.topology.links.metadata.added",
313		content={"link": MagicMock()}
314		)
315	1	self.napp.update_links_metadata_changed(event)
316	1	assert self.napp.graph.update_link_metadata.call_count == 1
317	1	assert self.napp.controller.buffers.app.put.call_count == 1
318
319	1	async def test_shortest_path(self):
320		"""Test shortest path."""
321	1	self.setting_path()
322	1	source, destination = "User1", "User4"
323	1	data = {"source": source, "destination": destination}
324	1	response = await self.api_client.post(self.endpoint, json=data)
325
326	1	for path in response.json()["paths"]:
327		assert source == path["hops"][0]
328		assert destination == path["hops"][-1]
329
330	1	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	1	self.setting_path()
334	1	with patch(
335		"napps.kytos.pathfinder.graph.KytosGraph."
336		"constrained_k_shortest_paths",
337		side_effect=side_effect,
338		):
339	1	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	1	response = await self.api_client.post(self.endpoint, json=data)
347	1	return response
348
349	1	async def test_shortest_constrained_path_400_exception(self):
350		"""Test shortest path."""
351	1	res = await self.setting_shortest_constrained_path_exception(TypeError)
352		assert res.status_code == 400
353

kytos-ng / pathfinder

Pull Request — master (#45)

build.tests.unit.test_main.TestMain.setUp() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like