build.tests.unit.test_main.TestMain.setup_method() - Code Metrics - Inspection of "Merge pull request #45 from kytos-ng/chore/starlet..." - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( a120bd...d3dde8 )

by Vinicius

created 2023-05-03 17:56 UTC

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

↳ Parent: build.tests.unit.test_main

Complexity

Conditions

Size

Total Lines	6
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	5
nop	1
dl	0
loc	6
ccs	5
cts	5
cp	1
crap	1
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, 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",
            "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, 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",
            "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, 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_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, 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/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, 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		"desired_links": ["1"]
86		}
87	1	response = await self.api_client.post(self.endpoint, json=data)
88	1	assert response.status_code == 200
89	1	expected_response = {
90		"paths": [{"hops": path, "cost": cost_mocked_value}]
91		}
92	1	assert response.json() == expected_response
93
94	1	async def test_shortest_path_response_status_code(
95		self, monkeypatch, event_loop
96		):
97		"""Test shortest path."""
98	1	self.napp.controller.loop = event_loop
99	1	cost_mocked_value = 1
100	1	mock_path_cost = MagicMock(return_value=cost_mocked_value)
101	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
102		"KytosGraph._path_cost", mock_path_cost)
103	1	mock_shortest_paths = MagicMock()
104	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
105		"KytosGraph.k_shortest_paths", mock_shortest_paths)
106	1	_ = self.setting_shortest_path_mocked(mock_shortest_paths)
107	1	data = {
108		"source": "00:00:00:00:00:00:00:01:1",
109		"destination": "00:00:00:00:00:00:00:02:1",
110		"desired_links": ["1"]
111		}
112	1	response = await self.api_client.post(self.endpoint, json=data)
113	1	assert response.status_code == 200
114
115	1	async def setting_shortest_constrained_path_mocked(
116		self, mock_constrained_k_shortest_paths
117		):
118		"""Set the primary elements needed to test the retrieving process
119		of the shortest constrained path under a mocked approach."""
120	1	source = "00:00:00:00:00:00:00:01:1"
121	1	destination = "00:00:00:00:00:00:00:02:1"
122	1	path = [source, destination]
123	1	mandatory_metrics = {"ownership": "bob"}
124	1	fle_metrics = {"delay": 30}
125	1	metrics = {mandatory_metrics, fle_metrics}
126	1	mock_constrained_k_shortest_paths.return_value = [
127		{"hops": path, "metrics": metrics}
128		]
129	1	data = {
130		"source": "00:00:00:00:00:00:00:01:1",
131		"destination": "00:00:00:00:00:00:00:02:1",
132		"mandatory_metrics": {"ownership": "bob"},
133		"flexible_metrics": {"delay": 30},
134		"minimum_flexible_hits": 1,
135		}
136	1	response = await self.api_client.post(self.endpoint, json=data)
137
138	1	return response, metrics, path
139
140	1	async def test_shortest_constrained_path_response(
141		self, monkeypatch, event_loop
142		):
143		"""Test constrained flexible paths."""
144	1	self.napp.controller.loop = event_loop
145	1	mock_path_cost = MagicMock(return_value=1)
146	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
147		"KytosGraph._path_cost", mock_path_cost)
148	1	mock_k = MagicMock()
149	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
150		"KytosGraph.constrained_k_shortest_paths", mock_k)
151	1	(
152		response,
153		metrics,
154		path,
155		) = await self.setting_shortest_constrained_path_mocked(mock_k)
156	1	expected_response = [
157		{"metrics": metrics, "hops": path, "cost": 1}
158		]
159	1	assert response.json()["paths"][0] == expected_response[0]
160
161	1	async def test_shortest_constrained_path_response_status_code(
162		self, monkeypatch, event_loop
163		):
164		"""Test constrained flexible paths."""
165	1	self.napp.controller.loop = event_loop
166	1	mock_path_cost = MagicMock(return_value=1)
167	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
168		"KytosGraph._path_cost", mock_path_cost)
169	1	mock_k = MagicMock()
170	1	monkeypatch.setattr("napps.kytos.pathfinder.graph."
171		"KytosGraph.k_shortest_paths", mock_k)
172	1	response, _, _ = await self.setting_shortest_constrained_path_mocked(
173		mock_k
174		)
175	1	assert response.status_code == 200
176
177	1	def test_filter_paths_response_on_desired(self):
178		"""Test filter paths."""
179	1	self.napp._topology = get_topology_mock()
180	1	paths = [
181		{
182		"hops": [
183		"00:00:00:00:00:00:00:01:1",
184		"00:00:00:00:00:00:00:02:1",
185		"00:00:00:00:00:00:00:02:2",
186		"00:00:00:00:00:00:00:03:2",
187		]
188		},
189		{
190		"hops": [
191		"00:00:00:00:00:00:00:01:1",
192		"00:00:00:00:00:00:00:01",
193		"00:00:00:00:00:00:00:04",
194		"00:00:00:00:00:00:00:04:1",
195		],
196		"cost": 3,
197		},
198		]
199	1	desired = ["1", "3"]
200
201	1	for link in desired:
202	1	assert self.napp._topology.links[link]
203	1	filtered_paths = self.napp._filter_paths_desired_links(paths, desired)
204	1	assert filtered_paths == [paths[0]]
205
206	1	filtered_paths = self.napp._filter_paths_desired_links(paths, ["1", "2"])
207	1	assert not filtered_paths
208
209	1	filtered_paths = self.napp._filter_paths_desired_links(paths, ["inexistent_id"])
210	1	assert not filtered_paths
211
212	1	def test_filter_paths_le_cost_response(self):
213		"""Test filter paths."""
214	1	self.napp._topology = get_topology_mock()
215	1	paths = [
216		{
217		"hops": [
218		"00:00:00:00:00:00:00:01:1",
219		"00:00:00:00:00:00:00:01",
220		"00:00:00:00:00:00:00:02:1",
221		"00:00:00:00:00:00:00:02",
222		"00:00:00:00:00:00:00:02:2",
223		"00:00:00:00:00:00:00:04",
224		"00:00:00:00:00:00:00:04:1",
225		],
226		"cost": 6,
227		},
228		{
229		"hops": [
230		"00:00:00:00:00:00:00:01:1",
231		"00:00:00:00:00:00:00:01",
232		"00:00:00:00:00:00:00:04",
233		"00:00:00:00:00:00:00:04:1",
234		],
235		"cost": 3,
236		},
237		]
238	1	filtered_paths = self.napp._filter_paths_le_cost(paths, 3)
239	1	assert len(filtered_paths) == 1
240	1	assert filtered_paths[0]["cost"] == 3
241
242	1	def test_filter_paths_response_on_undesired(self):
243		"""Test filter paths."""
244	1	self.napp._topology = get_topology_mock()
245	1	paths = [
246		{
247		"hops": [
248		"00:00:00:00:00:00:00:01:1",
249		"00:00:00:00:00:00:00:02:1",
250		"00:00:00:00:00:00:00:02:2",
251		"00:00:00:00:00:00:00:03:2",
252		]
253		}
254		]
255
256	1	undesired = ["1"]
257	1	filtered_paths = self.napp._filter_paths_undesired_links(paths, undesired)
258	1	assert not filtered_paths
259
260	1	undesired = ["3"]
261	1	filtered_paths = self.napp._filter_paths_undesired_links(paths, undesired)
262	1	assert not filtered_paths
263
264	1	undesired = ["1", "3"]
265	1	filtered_paths = self.napp._filter_paths_undesired_links(paths, undesired)
266	1	assert not filtered_paths
267
268	1	filtered_paths = self.napp._filter_paths_undesired_links(paths, ["none"])
269	1	assert filtered_paths == paths
270
271	1	def setting_path(self):
272		"""Set the primary elements needed to test the topology
273		update process under a "real-simulated" scenario."""
274	1	topology = get_topology_with_metadata()
275	1	event = KytosEvent(
276		name="kytos.topology.updated", content={"topology": topology}
277		)
278	1	self.napp.update_topology(event)
279
280	1	async def test_update_links_changed(self):
281		"""Test update_links_metadata_changed."""
282	1	self.napp.graph.update_link_metadata = MagicMock()
283	1	self.napp.controller.buffers.app.put = MagicMock()
284	1	event = KytosEvent(
285		name="kytos.topology.links.metadata.added",
286		content={"link": MagicMock(), "metadata": {}}
287		)
288	1	self.napp.update_links_metadata_changed(event)
289	1	assert self.napp.graph.update_link_metadata.call_count == 1
290	1	assert self.napp.controller.buffers.app.put.call_count == 0
291
292	1	async def test_update_links_changed_out_of_order(self):
293		"""Test update_links_metadata_changed out of order."""
294	1	self.napp.graph.update_link_metadata = MagicMock()
295	1	self.napp.controller.buffers.app.put = MagicMock()
296	1	link = MagicMock(id="1")
297	1	assert link.id not in self.napp._links_updated_at
298	1	event = KytosEvent(
299		name="kytos.topology.links.metadata.added",
300		content={"link": link, "metadata": {}}
301		)
302	1	self.napp.update_links_metadata_changed(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	second_event = KytosEvent(
308		name="kytos.topology.links.metadata.added",
309		content={"link": link, "metadata": {}}
310		)
311	1	second_event.timestamp = event.timestamp - timedelta(seconds=10)
312	1	self.napp.update_links_metadata_changed(second_event)
313	1	assert self.napp.graph.update_link_metadata.call_count == 1
314	1	assert self.napp.controller.buffers.app.put.call_count == 0
315	1	assert self.napp._links_updated_at[link.id] == event.timestamp
316
317	1	async def test_update_links_changed_key_error(self):
318		"""Test update_links_metadata_changed key_error."""
319	1	self.napp.graph.update_link_metadata = MagicMock()
320	1	self.napp.controller.buffers.app.put = MagicMock()
321	1	event = KytosEvent(
322		name="kytos.topology.links.metadata.added",
323		content={"link": MagicMock()}
324		)
325	1	self.napp.update_links_metadata_changed(event)
326	1	assert self.napp.graph.update_link_metadata.call_count == 1
327	1	assert self.napp.controller.buffers.app.put.call_count == 1
328
329	1	async def test_shortest_path(self, event_loop):
330		"""Test shortest path."""
331	1	self.napp.controller.loop = event_loop
332	1	self.setting_path()
333	1	source, destination = "User1", "User4"
334	1	data = {"source": source, "destination": destination}
335	1	response = await self.api_client.post(self.endpoint, json=data)
336
337	1	for path in response.json()["paths"]:
338		assert source == path["hops"][0]
339		assert destination == path["hops"][-1]
340
341	1	async def setting_shortest_constrained_path_exception(self, side_effect):
342		"""Set the primary elements needed to test the shortest
343		constrained path behavior under exception actions."""
344	1	self.setting_path()
345	1	with patch(
346		"napps.kytos.pathfinder.graph.KytosGraph."
347		"constrained_k_shortest_paths",
348		side_effect=side_effect,
349		):
350	1	data = {
351		"source": "00:00:00:00:00:00:00:01:1",
352		"destination": "00:00:00:00:00:00:00:02:1",
353		"mandatory_metrics": {"ownership": "bob"},
354		"flexible_metrics": {"delay": 30},
355		"minimum_flexible_hits": 1,
356		}
357	1	response = await self.api_client.post(self.endpoint, json=data)
358	1	return response
359
360	1	async def test_shortest_constrained_path_400_exception(self, event_loop):
361		"""Test shortest path."""
362	1	self.napp.controller.loop = event_loop
363	1	res = await self.setting_shortest_constrained_path_exception(TypeError)
364		assert res.status_code == 400
365

kytos-ng / pathfinder

Push — master ( a120bd...d3dde8 )

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

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like