build.tests.unit.test_graph.TestGraph.test_constrained_k_shortest_paths() - Code Metrics - Inspection of "Merge pull request #11 from kytos-ng/fix/issue_10" - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 12cd8c...fd29a5 )

by Vinicius

created 2022-01-27 21:30 UTC

TestGraph.test_constrained_k_shortest_paths() A

↳ Parent: build.tests.unit.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

    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())[:keys_num]:
            topology.switches[key].status = EntityStatus.DISABLED
        for key in topology.switches.keys():
            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.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			def test_update_nodes_not_up(self):
75			"""Test update_nodes entity not up."""
76			topology = get_topology_mock()
77			keys_num = 2
78			for key in list(topology.switches.keys())[:keys_num]:
79			topology.switches[key].status = EntityStatus.DISABLED
80			for key in topology.switches.keys():
81			topology.switches[key].interfaces = {}
82
83			self.kytos_graph.update_nodes(topology.switches)
84			assert self.mock_graph.add_node.call_count == len(topology.switches) - keys_num
85
86			def test_update_nodes(self):
87			"""Test update nodes."""
88			topology = get_topology_mock()
89			self.kytos_graph.update_nodes(topology.switches)
90
91			edge_count = sum(
92			[len(sw.interfaces.values()) for sw in topology.switches.values()]
93			)
94			node_count = len(topology.switches) + edge_count
95			assert self.mock_graph.add_edge.call_count == edge_count
96			assert self.mock_graph.add_node.call_count == node_count
97
98			def test_update_nodes_2(self):
99			"""Test update nodes."""
100
101			effect = MagicMock(side_effect=AttributeError)
102
103			topology = get_topology_mock()
104			with self.assertRaises(Exception):
105			with patch.object(self.mock_graph, "add_node", effect):
106			self.kytos_graph.update_nodes(topology.switches)
107
108			self.assertRaises(AttributeError)
109
110			def test_remove_switch_hops(self):
111			"""Test remove switch hops."""
112			circuit = {
113			"hops": [
114			"00:00:00:00:00:00:00:01:1",
115			"00:00:00:00:00:00:00:01",
116			"00:00:00:00:00:00:00:01:2",
117			]
118			}
119
120			self.kytos_graph._remove_switch_hops(circuit)
121
122			expected_circuit = {
123			"hops": ["00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:01:2"]
124			}
125			self.assertEqual(circuit, expected_circuit)
126
127			@patch("networkx.shortest_simple_paths", return_value=["any"])
128			def test_shortest_paths(self, mock_shortest_simple_paths):
129			"""Test shortest paths."""
130			source, dest = "00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:02:2"
131			k_shortest_paths = self.kytos_graph.k_shortest_paths(source, dest)
132
133			mock_shortest_simple_paths.assert_called_with(
134			self.kytos_graph.graph, source, dest, weight=None
135			)
136			self.assertEqual(k_shortest_paths, ["any"])
137
138			@patch("napps.kytos.pathfinder.graph.combinations", autospec=True)
139			def test_constrained_k_shortest_paths(self, mock_combinations):
140			"""Test shortest constrained paths."""
141			source, dest = "00:00:00:00:00:00:00:01:1", "00:00:00:00:00:00:00:02:2"
142			minimum_hits = 1
143			mandatory_metrics = {"bandwidth": 100}
144			flexible_metrics = {"utilization": 2}
145			mock_combinations.return_value = [(("utilization", 2),)]
146			constrained_k_shortest_paths = [["path1"], ["path2"]]
147
148			self.kytos_graph.graph.edge_subgraph = MagicMock(return_value=None)
149			self.kytos_graph.k_shortest_paths = MagicMock(
150			return_value=constrained_k_shortest_paths
151			)
152			self.kytos_graph._filter_links = MagicMock(side_effect=get_filter_links_fake)
153			k_shortest_paths = self.kytos_graph.constrained_k_shortest_paths(
154			source,
155			dest,
156			minimum_hits=minimum_hits,
157			mandatory_metrics=mandatory_metrics,
158			flexible_metrics=flexible_metrics,
159			)
160
161			self.kytos_graph.k_shortest_paths.assert_has_calls(
162			[
163			call(
164			source,
165			dest,
166			weight=None,
167			k=1,
168			graph=None,
169			)
170			]
171			)
172			self.kytos_graph._filter_links.assert_called()
173			for constrained_path in k_shortest_paths:
174			assert constrained_path["hops"] in constrained_k_shortest_paths
175			assert constrained_path["metrics"] == {
176			"bandwidth": 100,
177			"utilization": 2,
178			}
179
180			def test_get_link_metadata(self):
181			"""Test metadata retrieval."""
182			topology = get_topology_with_metadata_mock()
183			self.kytos_graph.update_nodes(topology.switches)
184			self.kytos_graph.update_links(topology.links)
185			endpoint_a = "S1:1"
186			endpoint_b = "S2:1"
187			metadata = {"reliability": 5, "bandwidth": 100, "delay": 105}
188			self.kytos_graph.get_link_metadata = MagicMock(return_value=metadata)
189
190			result = self.kytos_graph.get_link_metadata(endpoint_a, endpoint_b)
191
192			assert result == metadata
193
194			def test_update_link_metadata(self):
195			"""Test update link metadata."""
196			graph = MagicMock()
197			link = MagicMock()
198			endpoint_a_id = 1
199			endpoint_b_id = 2
200			link.endpoint_a.id = endpoint_a_id
201			link.endpoint_b.id = endpoint_b_id
202			link.metadata.items.return_value = [("reliability", 50)]
203			self.kytos_graph.graph = graph
204			self.kytos_graph.update_link_metadata(link)
205			call_res = str(self.kytos_graph.graph._mock_mock_calls)
206			assert "__getitem__(1)" in call_res
207			assert "__getitem__(2)" in call_res
208			assert "__setitem__('reliability', 50)" in call_res
209
210			def test_update_link_unsupported_metadata(self):
211			"""Test update link metadata with an unsupported key."""
212			graph = MagicMock()
213			link = MagicMock()
214			endpoint_a_id = 1
215			endpoint_b_id = 2
216			link.endpoint_a.id = endpoint_a_id
217			link.endpoint_b.id = endpoint_b_id
218			link.metadata.items.return_value = [("random_metric", 50)]
219			self.kytos_graph.graph = graph
220			self.kytos_graph.update_link_metadata(link)
221			assert self.kytos_graph.graph._mock_mock_calls == []
222

kytos-ng / pathfinder

Push — master ( 12cd8c...fd29a5 )

TestGraph.test_constrained_k_shortest_paths() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like