build.tests.unit.graph.test_graph.TestGraph.test_constrained_k_shortest_paths() - Code Metrics - Inspection of "Feat/negative ownership" - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#65)

unknown

created 2023-12-11 21:26 UTC

TestGraph.test_constrained_k_shortest_paths() A

↳ Parent: build.tests.unit.graph.test_graph

Complexity

Conditions

Size

Total Lines	40
Code Lines	32

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	32
nop	2
dl	0
loc	40
rs	9.112
c	0
b	0
f	0

"""Test Graph methods."""
from unittest import TestCase
from unittest.mock import MagicMock, call, patch

from kytos.core.common import EntityStatus
from napps.kytos.pathfinder.graph import KytosGraph

# pylint: disable=import-error
from tests.helpers import (
    get_filter_links_fake,
    get_topology_mock,
    get_topology_with_metadata_mock,
)

# pylint: disable=arguments-differ, protected-access, no-member


class TestGraph(TestCase):
    """Tests for the Main class."""

    @patch("networkx.Graph")
    def setUp(self, mock_graph):
        """Execute steps before each tests."""
        self.mock_graph = mock_graph.return_value
        self.kytos_graph = KytosGraph()

    def test_clear(self):
        """Test clear."""
        self.kytos_graph.clear()

        self.mock_graph.clear.assert_called()

    def setting_update_topology(self, *args):
        """Set the primary elements needed to
        test the topology update process."""
        (mock_update_nodes, mock_update_links) = args
        topology = get_topology_mock()
        self.kytos_graph.update_topology(topology)

        self.mock_graph.clear.assert_called()

        return mock_update_nodes, mock_update_links, topology

    @patch("napps.kytos.pathfinder.graph.KytosGraph.update_links")
    @patch("napps.kytos.pathfinder.graph.KytosGraph.update_nodes")
    def test_update_topology_switches(self, *args):
        """Test update topology."""
        mock_update_nodes, _, topology = self.setting_update_topology(*args)
        mock_update_nodes.assert_called_with(topology.switches)

    @patch("napps.kytos.pathfinder.graph.KytosGraph.update_links")
    @patch("napps.kytos.pathfinder.graph.KytosGraph.update_nodes")
    def test_update_topology_links(self, *args):
        """Test update topology."""
        _, mock_update_links, topology = self.setting_update_topology(*args)
        mock_update_links.assert_called_with(topology.links)

    def test_update_links(self):
        """Test update_links."""
        topology = get_topology_mock()
        self.kytos_graph.update_links(topology.links)
        assert self.mock_graph.add_edge.call_count == len(topology.links)

    def test_update_links_not_up(self):
        """Test update_links entity not up."""
        topology = get_topology_mock()
        keys_num = 2
        for key in list(topology.links.keys())[:keys_num]:
            topology.links[key].status = EntityStatus.DOWN

        self.kytos_graph.update_links(topology.links)
        assert self.mock_graph.add_edge.call_count == len(topology.links) - keys_num

    # pylint: disable=consider-using-dict-items
    def test_update_nodes_not_up(self):
        """Test update_nodes entity not up."""
        topology = get_topology_mock()
        keys_num = 2
        for key in list(topology.switches)[:keys_num]:
            topology.switches[key].status = EntityStatus.DISABLED
        for key in topology.switches:
            topology.switches[key].interfaces = {}

        self.kytos_graph.update_nodes(topology.switches)
        assert self.mock_graph.add_node.call_count == len(topology.switches) - keys_num

    def test_update_nodes(self):
        """Test update nodes."""
        topology = get_topology_mock()
        self.kytos_graph.update_nodes(topology.switches)

        edge_count = sum(
            (len(sw.interfaces.values()) for sw in topology.switches.values())

        )
        node_count = len(topology.switches) + edge_count
        assert self.mock_graph.add_edge.call_count == edge_count
        assert self.mock_graph.add_node.call_count == node_count

    def test_update_nodes_2(self):
        """Test update nodes."""

        effect = MagicMock(side_effect=AttributeError)

        topology = get_topology_mock()
        with self.assertRaises(Exception):
            with patch.object(self.mock_graph, "add_node", effect):
                self.kytos_graph.update_nodes(topology.switches)

        self.assertRaises(AttributeError)

    def test_remove_switch_hops(self):
        """Test remove switch hops."""
        circuit = {
            "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:01:2",
            ]
        }

        self.kytos_graph._remove_switch_hops(circuit)

        expected_circuit = {
            "hops": ["00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:01:2"]
        }
        self.assertEqual(circuit, expected_circuit)

    @patch("networkx.shortest_simple_paths", return_value=["any"])
    def test_shortest_paths(self, mock_shortest_simple_paths):
        """Test shortest paths."""
        source, dest = "00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:02:2"
        k_shortest_paths = self.kytos_graph.k_shortest_paths(source, dest)

        mock_shortest_simple_paths.assert_called_with(
            self.kytos_graph.graph, source, dest, weight=None
        )
        self.assertEqual(k_shortest_paths, ["any"])

    @patch("napps.kytos.pathfinder.graph.graph.combinations", autospec=True)
    def test_constrained_k_shortest_paths(self, mock_combinations):
        """Test shortest constrained paths."""
        source, dest = "00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:02:2"
        minimum_hits = 1
        mandatory_metrics = {"bandwidth": 100}
        flexible_metrics = {"utilization": 2}
        mock_combinations.return_value = [(("utilization", 2),)]
        constrained_k_shortest_paths = [["path1"], ["path2"]]

        self.kytos_graph.graph.edge_subgraph = MagicMock(return_value=None)
        self.kytos_graph.k_shortest_paths = MagicMock(
            return_value=constrained_k_shortest_paths
        )
        self.kytos_graph._filter_links = MagicMock(side_effect=get_filter_links_fake)
        k_shortest_paths = self.kytos_graph.constrained_k_shortest_paths(
            source,
            dest,
            minimum_hits=minimum_hits,
            mandatory_metrics=mandatory_metrics,
            flexible_metrics=flexible_metrics,
        )

        self.kytos_graph.k_shortest_paths.assert_has_calls(
            [
                call(
                    source,
                    dest,
                    weight=None,
                    k=1,
                    graph=None,
                )
            ]
        )
        self.kytos_graph._filter_links.assert_called()
        for constrained_path in k_shortest_paths:
            assert constrained_path["hops"] in constrained_k_shortest_paths
            assert constrained_path["metrics"] == {
                "bandwidth": 100,
                "utilization": 2,
            }

    def test_get_link_metadata(self):
        """Test metadata retrieval."""
        topology = get_topology_with_metadata_mock()
        self.kytos_graph.update_nodes(topology.switches)
        self.kytos_graph.update_links(topology.links)
        endpoint_a = "S1:1"
        endpoint_b = "S2:1"
        metadata = {"reliability": 5, "bandwidth": 100, "delay": 105}
        self.kytos_graph.get_link_metadata = MagicMock(return_value=metadata)

        result = self.kytos_graph.get_link_metadata(endpoint_a, endpoint_b)

        assert result == metadata

    def test_update_link_metadata(self):
        """Test update link metadata."""
        graph = MagicMock()
        link = MagicMock()
        endpoint_a_id = 1
        endpoint_b_id = 2
        link.endpoint_a.id = endpoint_a_id
        link.endpoint_b.id = endpoint_b_id
        link.metadata.items.return_value = [("reliability", 50)]
        self.kytos_graph.graph = graph
        self.kytos_graph.update_link_metadata(link)
        call_res = str(self.kytos_graph.graph._mock_mock_calls)
        assert "__getitem__(1)" in call_res
        assert "__getitem__(2)" in call_res
        assert "__setitem__('reliability', 50)" in call_res

    def test_update_link_unsupported_metadata(self):
        """Test update link metadata with an unsupported key."""
        graph = MagicMock()
        link = MagicMock()
        endpoint_a_id = 1
        endpoint_b_id = 2
        link.endpoint_a.id = endpoint_a_id
        link.endpoint_b.id = endpoint_b_id
        link.metadata.items.return_value = [("random_metric", 50)]
        self.kytos_graph.graph = graph
        self.kytos_graph.update_link_metadata(link)
        assert self.kytos_graph.graph._mock_mock_calls == []


1			"""Test Graph methods."""
2			from unittest import TestCase
3			from unittest.mock import MagicMock, call, patch
4
5			from kytos.core.common import EntityStatus
6			from napps.kytos.pathfinder.graph import KytosGraph
7
8			# pylint: disable=import-error
9			from tests.helpers import (
10			get_filter_links_fake,
11			get_topology_mock,
12			get_topology_with_metadata_mock,
13			)
14
15			# pylint: disable=arguments-differ, protected-access, no-member
16
17
18			class TestGraph(TestCase):
19			"""Tests for the Main class."""
20
21			@patch("networkx.Graph")
22			def setUp(self, mock_graph):
23			"""Execute steps before each tests."""
24			self.mock_graph = mock_graph.return_value
25			self.kytos_graph = KytosGraph()
26
27			def test_clear(self):
28			"""Test clear."""
29			self.kytos_graph.clear()
30
31			self.mock_graph.clear.assert_called()
32
33			def setting_update_topology(self, *args):
34			"""Set the primary elements needed to
35			test the topology update process."""
36			(mock_update_nodes, mock_update_links) = args
37			topology = get_topology_mock()
38			self.kytos_graph.update_topology(topology)
39
40			self.mock_graph.clear.assert_called()
41
42			return mock_update_nodes, mock_update_links, topology
43
44			@patch("napps.kytos.pathfinder.graph.KytosGraph.update_links")
45			@patch("napps.kytos.pathfinder.graph.KytosGraph.update_nodes")
46			def test_update_topology_switches(self, *args):
47			"""Test update topology."""
48			mock_update_nodes, _, topology = self.setting_update_topology(*args)
49			mock_update_nodes.assert_called_with(topology.switches)
50
51			@patch("napps.kytos.pathfinder.graph.KytosGraph.update_links")
52			@patch("napps.kytos.pathfinder.graph.KytosGraph.update_nodes")
53			def test_update_topology_links(self, *args):
54			"""Test update topology."""
55			_, mock_update_links, topology = self.setting_update_topology(*args)
56			mock_update_links.assert_called_with(topology.links)
57
58			def test_update_links(self):
59			"""Test update_links."""
60			topology = get_topology_mock()
61			self.kytos_graph.update_links(topology.links)
62			assert self.mock_graph.add_edge.call_count == len(topology.links)
63
64			def test_update_links_not_up(self):
65			"""Test update_links entity not up."""
66			topology = get_topology_mock()
67			keys_num = 2
68			for key in list(topology.links.keys())[:keys_num]:
69			topology.links[key].status = EntityStatus.DOWN
70
71			self.kytos_graph.update_links(topology.links)
72			assert self.mock_graph.add_edge.call_count == len(topology.links) - keys_num
73
74			# pylint: disable=consider-using-dict-items
75			def test_update_nodes_not_up(self):
76			"""Test update_nodes entity not up."""
77			topology = get_topology_mock()
78			keys_num = 2
79			for key in list(topology.switches)[:keys_num]:
80			topology.switches[key].status = EntityStatus.DISABLED
81			for key in topology.switches:
82			topology.switches[key].interfaces = {}
83
84			self.kytos_graph.update_nodes(topology.switches)
85			assert self.mock_graph.add_node.call_count == len(topology.switches) - keys_num
86
87			def test_update_nodes(self):
88			"""Test update nodes."""
89			topology = get_topology_mock()
90			self.kytos_graph.update_nodes(topology.switches)
91
92			edge_count = sum(
93			(len(sw.interfaces.values()) for sw in topology.switches.values())
			0 ignored issues – show Comprehensibility Best Practice introduced 2022-09-12 15:59 UTC by Report Bug Copy Issue Report The variable `sw` does not seem to be defined. Loading history...
94			)
95			node_count = len(topology.switches) + edge_count
96			assert self.mock_graph.add_edge.call_count == edge_count
97			assert self.mock_graph.add_node.call_count == node_count
98
99			def test_update_nodes_2(self):
100			"""Test update nodes."""
101
102			effect = MagicMock(side_effect=AttributeError)
103
104			topology = get_topology_mock()
105			with self.assertRaises(Exception):
106			with patch.object(self.mock_graph, "add_node", effect):
107			self.kytos_graph.update_nodes(topology.switches)
108
109			self.assertRaises(AttributeError)
110
111			def test_remove_switch_hops(self):
112			"""Test remove switch hops."""
113			circuit = {
114			"hops": [
115			"00:00:00:00:00:00:00:01:1",
116			"00:00:00:00:00:00:00:01",
117			"00:00:00:00:00:00:00:01:2",
118			]
119			}
120
121			self.kytos_graph._remove_switch_hops(circuit)
122
123			expected_circuit = {
124			"hops": ["00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:01:2"]
125			}
126			self.assertEqual(circuit, expected_circuit)
127
128			@patch("networkx.shortest_simple_paths", return_value=["any"])
129			def test_shortest_paths(self, mock_shortest_simple_paths):
130			"""Test shortest paths."""
131			source, dest = "00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:02:2"
132			k_shortest_paths = self.kytos_graph.k_shortest_paths(source, dest)
133
134			mock_shortest_simple_paths.assert_called_with(
135			self.kytos_graph.graph, source, dest, weight=None
136			)
137			self.assertEqual(k_shortest_paths, ["any"])
138
139			@patch("napps.kytos.pathfinder.graph.graph.combinations", autospec=True)
140			def test_constrained_k_shortest_paths(self, mock_combinations):
141			"""Test shortest constrained paths."""
142			source, dest = "00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:02:2"
143			minimum_hits = 1
144			mandatory_metrics = {"bandwidth": 100}
145			flexible_metrics = {"utilization": 2}
146			mock_combinations.return_value = [(("utilization", 2),)]
147			constrained_k_shortest_paths = [["path1"], ["path2"]]
148
149			self.kytos_graph.graph.edge_subgraph = MagicMock(return_value=None)
150			self.kytos_graph.k_shortest_paths = MagicMock(
151			return_value=constrained_k_shortest_paths
152			)
153			self.kytos_graph._filter_links = MagicMock(side_effect=get_filter_links_fake)
154			k_shortest_paths = self.kytos_graph.constrained_k_shortest_paths(
155			source,
156			dest,
157			minimum_hits=minimum_hits,
158			mandatory_metrics=mandatory_metrics,
159			flexible_metrics=flexible_metrics,
160			)
161
162			self.kytos_graph.k_shortest_paths.assert_has_calls(
163			[
164			call(
165			source,
166			dest,
167			weight=None,
168			k=1,
169			graph=None,
170			)
171			]
172			)
173			self.kytos_graph._filter_links.assert_called()
174			for constrained_path in k_shortest_paths:
175			assert constrained_path["hops"] in constrained_k_shortest_paths
176			assert constrained_path["metrics"] == {
177			"bandwidth": 100,
178			"utilization": 2,
179			}
180
181			def test_get_link_metadata(self):
182			"""Test metadata retrieval."""
183			topology = get_topology_with_metadata_mock()
184			self.kytos_graph.update_nodes(topology.switches)
185			self.kytos_graph.update_links(topology.links)
186			endpoint_a = "S1:1"
187			endpoint_b = "S2:1"
188			metadata = {"reliability": 5, "bandwidth": 100, "delay": 105}
189			self.kytos_graph.get_link_metadata = MagicMock(return_value=metadata)
190
191			result = self.kytos_graph.get_link_metadata(endpoint_a, endpoint_b)
192
193			assert result == metadata
194
195			def test_update_link_metadata(self):
196			"""Test update link metadata."""
197			graph = MagicMock()
198			link = MagicMock()
199			endpoint_a_id = 1
200			endpoint_b_id = 2
201			link.endpoint_a.id = endpoint_a_id
202			link.endpoint_b.id = endpoint_b_id
203			link.metadata.items.return_value = [("reliability", 50)]
204			self.kytos_graph.graph = graph
205			self.kytos_graph.update_link_metadata(link)
206			call_res = str(self.kytos_graph.graph._mock_mock_calls)
207			assert "__getitem__(1)" in call_res
208			assert "__getitem__(2)" in call_res
209			assert "__setitem__('reliability', 50)" in call_res
210
211			def test_update_link_unsupported_metadata(self):
212			"""Test update link metadata with an unsupported key."""
213			graph = MagicMock()
214			link = MagicMock()
215			endpoint_a_id = 1
216			endpoint_b_id = 2
217			link.endpoint_a.id = endpoint_a_id
218			link.endpoint_b.id = endpoint_b_id
219			link.metadata.items.return_value = [("random_metric", 50)]
220			self.kytos_graph.graph = graph
221			self.kytos_graph.update_link_metadata(link)
222			assert self.kytos_graph.graph._mock_mock_calls == []
223

kytos-ng / pathfinder

Pull Request — master (#65)

TestGraph.test_constrained_k_shortest_paths() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like