build.tests.unit.test_graph.TestGraph.test_update_link_metadata() - Code Metrics - Inspection of "[Feature] Subscribed to `kytos/topology.links.meta..." - kytos-ng/pathfinder - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#5)

by Vinicius

created 2021-12-20 18:35 UTC

TestGraph.test_update_link_metadata() A

↳ Parent: build.tests.unit.test_graph

Complexity

Conditions

Size

Total Lines	15
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	14
nop	1
dl	0
loc	15
rs	9.7
c	0
b	0
f	0

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

from napps.kytos.pathfinder.graph import KytosGraph
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_nodes(self):
        """Test update nodes."""
        topology = get_topology_mock()
        self.kytos_graph.update_nodes(topology.switches)
        switch = topology.switches["00:00:00:00:00:00:00:01"]

        calls = [call(switch.id)]
        calls += [
            call(interface.id) for interface in switch.interfaces.values()
        ]
        self.mock_graph.add_node.assert_has_calls(calls)

        calls = [
            call(switch.id, interface.id)
            for interface in switch.interfaces.values()
        ]

        self.mock_graph.add_edge.assert_has_calls(calls)

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

kytos-ng / pathfinder

Pull Request — master (#5)

TestGraph.test_update_link_metadata() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like