build.tests.unit.test_loop_manager.TestLoopManager.setUp() - Code Metrics - Inspection of "[Feature] Loop detection in the same switch via LL..." - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#36)

by Vinicius

created 2022-03-02 13:26 UTC

TestLoopManager.setUp() A

↳ Parent: build.tests.unit.test_loop_manager

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	1
dl	0
loc	4
rs	10
c	0
b	0
f	0

"""Test LoopManager methods."""
from datetime import timedelta
from unittest import TestCase
from unittest.mock import MagicMock, patch

from kytos.lib.helpers import get_interface_mock, get_switch_mock

from kytos.core.helpers import now
from napps.kytos.of_lldp.loop_manager import LoopManager


# pylint: disable=protected-access
class TestLoopManager(TestCase):
    """Tests for LoopManager."""

    def setUp(self):
        """Execute steps before each tests."""
        controller = MagicMock()
        self.loop_manager = LoopManager(controller)

    def test_is_looped(self):
        """Test _is_looped cases."""

        dpid_a = "00:00:00:00:00:00:00:01"
        dpid_b = "00:00:00:00:00:00:00:02"
        values = [
            (dpid_a, 6, dpid_a, 7, True),
            (dpid_a, 1, dpid_a, 2, True),
            (dpid_a, 7, dpid_a, 7, True),
            (dpid_a, 8, dpid_a, 1, False),  # port_a > port_b
            (dpid_a, 1, dpid_b, 2, False),
            (dpid_a, 2, dpid_b, 1, False),
        ]
        for dpid_a, port_a, dpid_b, port_b, looped in values:
            with self.subTest(
                dpid_a=dpid_a, port_a=port_a, port_b=port_b, looped=looped
            ):
                assert (
                    self.loop_manager._is_looped(
                        dpid_a, port_a, dpid_b, port_b
                    )
                    == looped
                )

    def test_is_loop_ignored(self):
        """Test is_loop_ignored."""

        dpid = "00:00:00:00:00:00:00:01"
        port_a = 1
        port_b = 2
        self.loop_manager.ignored_loops[dpid] = [[port_a, port_b]]

        assert self.loop_manager._is_loop_ignored(
            dpid, port_a=port_a, port_b=port_b
        )
        assert self.loop_manager._is_loop_ignored(
            dpid, port_a=port_b, port_b=port_a
        )

        assert not self.loop_manager._is_loop_ignored(
            dpid, port_a + 20, port_b
        )

        dpid = "00:00:00:00:00:00:00:02"
        assert not self.loop_manager._is_loop_ignored(dpid, port_a, port_b)

    @patch("napps.kytos.of_lldp.loop_manager.log")
    def test_handle_log_action(self, mock_log):
        """Test handle_log_action."""

        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        intf_a = get_interface_mock("s1-eth1", 1, switch)
        intf_b = get_interface_mock("s1-eth2", 2, switch)

        self.loop_manager.handle_log_action(intf_a, intf_b)
        mock_log.warning.call_count = 1
        assert self.loop_manager.loop_counter[dpid][(1, 2)] == 0
        self.loop_manager.handle_log_action(intf_a, intf_b)
        mock_log.warning.call_count = 1
        assert self.loop_manager.loop_counter[dpid][(1, 2)] == 1

    @patch("napps.kytos.of_lldp.loop_manager.requests")
    @patch("napps.kytos.of_lldp.loop_manager.log")
    def test_handle_disable_action(self, mock_log, mock_requests):
        """Test handle_disable_action."""

        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        intf_a = get_interface_mock("s1-eth1", 1, switch)
        intf_b = get_interface_mock("s1-eth2", 2, switch)

        response = MagicMock()
        response.status_code = 200
        mock_requests.post.return_value = response
        self.loop_manager.handle_disable_action(intf_a, intf_b)
        assert mock_requests.post.call_count == 1
        assert mock_log.info.call_count == 1

    @patch("napps.kytos.of_lldp.loop_manager.requests")
    @patch("napps.kytos.of_lldp.loop_manager.log")
    def test_handle_loop_stopped(self, mock_log, mock_requests):
        """Test handle_loop_stopped."""

        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        intf_a = get_interface_mock("s1-eth1", 1, switch)
        intf_b = get_interface_mock("s1-eth2", 2, switch)

        self.loop_manager.loop_state[dpid][(1, 2)] = {"state": "detected"}
        response = MagicMock()
        response.status_code = 200
        mock_requests.post.return_value = response
        self.loop_manager.handle_loop_stopped(intf_a, intf_b)
        assert mock_requests.delete.call_count == 1
        assert "log" in self.loop_manager.actions
        assert mock_log.info.call_count == 1
        assert self.loop_manager.loop_state[dpid][(1, 2)]["state"] == "stopped"

    @patch(
        "napps.kytos.of_lldp.loop_manager.LoopManager.add_interface_metadata"
    )
    def test_handle_loop_detected(self, mock_add_interface_metadata):
        """Test handle_loop_detected."""

        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        intf_a = get_interface_mock("s1-eth1", 1, switch)

        port_pair = (1, 2)
        self.loop_manager.handle_loop_detected(intf_a.id, dpid, port_pair)
        assert (
            self.loop_manager.loop_state[dpid][port_pair]["state"]
            == "detected"
        )
        assert mock_add_interface_metadata.call_count == 1

        # aditional call while is still active it shouldn't add metadata again
        self.loop_manager.handle_loop_detected(intf_a.id, dpid, port_pair)
        assert mock_add_interface_metadata.call_count == 1

        # force a different initial state to ensure it would overwrite
        self.loop_manager.loop_state[dpid][port_pair]["state"] = "stopped"
        self.loop_manager.handle_loop_detected(intf_a.id, dpid, port_pair)
        assert mock_add_interface_metadata.call_count == 2

    @patch("napps.kytos.of_lldp.loop_manager.requests")
    def test_add_interface_metadata(self, mock_requests):
        """Test add interface metadata."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        intf_a = get_interface_mock("s1-eth1", 1, switch)
        metadata = {
            "state": "looped",
            "port_numbers": [1, 2],
            "updated_at": now().strftime("%Y-%m-%dT%H:%M:%S"),
            "detected_at": now().strftime("%Y-%m-%dT%H:%M:%S"),
        }
        self.loop_manager.add_interface_metadata(intf_a.id, metadata)
        assert mock_requests.post.call_count == 1

    @patch("napps.kytos.of_lldp.loop_manager.requests")
    def test_del_interface_metadata(self, mock_requests):
        """Test del interface metadata."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        intf_a = get_interface_mock("s1-eth1", 1, switch)
        self.loop_manager.del_interface_metadata(intf_a.id, "looped")
        assert mock_requests.delete.call_count == 1

    def test_publish_loop_state(self):
        """Test publish_loop_state."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        intf_a = get_interface_mock("s1-eth1", 1, switch)
        intf_b = get_interface_mock("s1-eth2", 2, switch)
        state = "detected"

        self.loop_manager.publish_loop_state(intf_a, intf_b, state)
        assert self.loop_manager.controller.buffers.app.put.call_count == 1

    def test_publish_loop_actions(self):
        """Test publish_loop_actions."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        intf_a = get_interface_mock("s1-eth1", 1, switch)
        intf_b = get_interface_mock("s1-eth2", 2, switch)
        self.loop_manager.publish_loop_actions(intf_a, intf_b)
        assert self.loop_manager.controller.buffers.app.put.call_count == len(
            set(self.loop_manager.actions)
        )

    def test_get_stopped_loops(self):
        """Test get_stopped_loops."""
        dpid = "00:00:00:00:00:00:00:01"
        port_pairs = [(1, 2), (3, 3)]

        delta = now() - timedelta(minutes=1)
        looped_entry = {
            "state": "detected",
            "updated_at": delta.strftime("%Y-%m-%dT%H:%M:%S"),
        }
        for port_pair in port_pairs:
            self.loop_manager.loop_state[dpid][port_pair] = looped_entry
        assert self.loop_manager.get_stopped_loops() == {dpid: port_pairs}

    @patch("napps.kytos.of_lldp.loop_manager.LoopManager.publish_loop_state")
    @patch("napps.kytos.of_lldp.loop_manager.LoopManager.publish_loop_actions")
    def test_process_if_looped(self, mock_publish_actions, mock_publish_state):
        """Test process_if_looped."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        intf_a = get_interface_mock("s1-eth1", 1, switch)
        intf_b = get_interface_mock("s1-eth2", 2, switch)
        self.loop_manager.ignored_loops = {}
        assert self.loop_manager.process_if_looped(intf_a, intf_b)
        assert mock_publish_actions.call_count == 1
        assert mock_publish_state.call_count == 1

    def test_handle_topology_update(self):
        """Test handle_topology pudate."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        mock_topo = MagicMock()
        switch.metadata = {"ignored_loops": [[1, 2]]}
        mock_topo.switches = {dpid: switch}

        assert dpid not in self.loop_manager.ignored_loops
        self.loop_manager.handle_topology_loaded(mock_topo)
        assert self.loop_manager.ignored_loops[dpid] == [[1, 2]]

    def test_handle_switch_metadata_changed_added(self):
        """Test handle_switch_metadata_changed added."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        switch.metadata = {"ignored_loops": [[1, 2]]}

        assert dpid not in self.loop_manager.ignored_loops
        self.loop_manager.handle_switch_metadata_changed(switch)
        assert self.loop_manager.ignored_loops[dpid] == [[1, 2]]

    def test_handle_switch_metadata_changed_removed(self):
        """Test handle_switch_metadata_changed removed."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        switch.id = dpid
        switch.metadata = {"some_key": "some_value"}
        self.loop_manager.ignored_loops[dpid] = [[1, 2]]

        assert dpid in self.loop_manager.ignored_loops
        self.loop_manager.handle_switch_metadata_changed(switch)
        assert dpid not in self.loop_manager.ignored_loops


1			"""Test LoopManager methods."""
2			from datetime import timedelta
3			from unittest import TestCase
4			from unittest.mock import MagicMock, patch
5
6			from kytos.lib.helpers import get_interface_mock, get_switch_mock
7
8			from kytos.core.helpers import now
9			from napps.kytos.of_lldp.loop_manager import LoopManager
10
11
12			# pylint: disable=protected-access
13			class TestLoopManager(TestCase):
14			"""Tests for LoopManager."""
15
16			def setUp(self):
17			"""Execute steps before each tests."""
18			controller = MagicMock()
19			self.loop_manager = LoopManager(controller)
20
21			def test_is_looped(self):
22			"""Test _is_looped cases."""
23
24			dpid_a = "00:00:00:00:00:00:00:01"
25			dpid_b = "00:00:00:00:00:00:00:02"
26			values = [
27			(dpid_a, 6, dpid_a, 7, True),
28			(dpid_a, 1, dpid_a, 2, True),
29			(dpid_a, 7, dpid_a, 7, True),
30			(dpid_a, 8, dpid_a, 1, False), # port_a > port_b
31			(dpid_a, 1, dpid_b, 2, False),
32			(dpid_a, 2, dpid_b, 1, False),
33			]
34			for dpid_a, port_a, dpid_b, port_b, looped in values:
35			with self.subTest(
36			dpid_a=dpid_a, port_a=port_a, port_b=port_b, looped=looped
37			):
38			assert (
39			self.loop_manager._is_looped(
40			dpid_a, port_a, dpid_b, port_b
41			)
42			== looped
43			)
44
45			def test_is_loop_ignored(self):
46			"""Test is_loop_ignored."""
47
48			dpid = "00:00:00:00:00:00:00:01"
49			port_a = 1
50			port_b = 2
51			self.loop_manager.ignored_loops[dpid] = [[port_a, port_b]]
52
53			assert self.loop_manager._is_loop_ignored(
54			dpid, port_a=port_a, port_b=port_b
55			)
56			assert self.loop_manager._is_loop_ignored(
57			dpid, port_a=port_b, port_b=port_a
58			)
59
60			assert not self.loop_manager._is_loop_ignored(
61			dpid, port_a + 20, port_b
62			)
63
64			dpid = "00:00:00:00:00:00:00:02"
65			assert not self.loop_manager._is_loop_ignored(dpid, port_a, port_b)
66
67			@patch("napps.kytos.of_lldp.loop_manager.log")
68			def test_handle_log_action(self, mock_log):
69			"""Test handle_log_action."""
70
71			dpid = "00:00:00:00:00:00:00:01"
72			switch = get_switch_mock(dpid, 0x04)
73			intf_a = get_interface_mock("s1-eth1", 1, switch)
74			intf_b = get_interface_mock("s1-eth2", 2, switch)
75
76			self.loop_manager.handle_log_action(intf_a, intf_b)
77			mock_log.warning.call_count = 1
78			assert self.loop_manager.loop_counter[dpid][(1, 2)] == 0
79			self.loop_manager.handle_log_action(intf_a, intf_b)
80			mock_log.warning.call_count = 1
81			assert self.loop_manager.loop_counter[dpid][(1, 2)] == 1
82
83			@patch("napps.kytos.of_lldp.loop_manager.requests")
84			@patch("napps.kytos.of_lldp.loop_manager.log")
85			def test_handle_disable_action(self, mock_log, mock_requests):
86			"""Test handle_disable_action."""
87
88			dpid = "00:00:00:00:00:00:00:01"
89			switch = get_switch_mock(dpid, 0x04)
90			intf_a = get_interface_mock("s1-eth1", 1, switch)
91			intf_b = get_interface_mock("s1-eth2", 2, switch)
92
93			response = MagicMock()
94			response.status_code = 200
95			mock_requests.post.return_value = response
96			self.loop_manager.handle_disable_action(intf_a, intf_b)
97			assert mock_requests.post.call_count == 1
98			assert mock_log.info.call_count == 1
99
100			@patch("napps.kytos.of_lldp.loop_manager.requests")
101			@patch("napps.kytos.of_lldp.loop_manager.log")
102			def test_handle_loop_stopped(self, mock_log, mock_requests):
103			"""Test handle_loop_stopped."""
104
105			dpid = "00:00:00:00:00:00:00:01"
106			switch = get_switch_mock(dpid, 0x04)
107			intf_a = get_interface_mock("s1-eth1", 1, switch)
108			intf_b = get_interface_mock("s1-eth2", 2, switch)
109
110			self.loop_manager.loop_state[dpid][(1, 2)] = {"state": "detected"}
111			response = MagicMock()
112			response.status_code = 200
113			mock_requests.post.return_value = response
114			self.loop_manager.handle_loop_stopped(intf_a, intf_b)
115			assert mock_requests.delete.call_count == 1
116			assert "log" in self.loop_manager.actions
117			assert mock_log.info.call_count == 1
118			assert self.loop_manager.loop_state[dpid][(1, 2)]["state"] == "stopped"
119
120			@patch(
121			"napps.kytos.of_lldp.loop_manager.LoopManager.add_interface_metadata"
122			)
123			def test_handle_loop_detected(self, mock_add_interface_metadata):
124			"""Test handle_loop_detected."""
125
126			dpid = "00:00:00:00:00:00:00:01"
127			switch = get_switch_mock(dpid, 0x04)
128			intf_a = get_interface_mock("s1-eth1", 1, switch)
129
130			port_pair = (1, 2)
131			self.loop_manager.handle_loop_detected(intf_a.id, dpid, port_pair)
132			assert (
133			self.loop_manager.loop_state[dpid][port_pair]["state"]
134			== "detected"
135			)
136			assert mock_add_interface_metadata.call_count == 1
137
138			# aditional call while is still active it shouldn't add metadata again
139			self.loop_manager.handle_loop_detected(intf_a.id, dpid, port_pair)
140			assert mock_add_interface_metadata.call_count == 1
141
142			# force a different initial state to ensure it would overwrite
143			self.loop_manager.loop_state[dpid][port_pair]["state"] = "stopped"
144			self.loop_manager.handle_loop_detected(intf_a.id, dpid, port_pair)
145			assert mock_add_interface_metadata.call_count == 2
146
147			@patch("napps.kytos.of_lldp.loop_manager.requests")
148			def test_add_interface_metadata(self, mock_requests):
149			"""Test add interface metadata."""
150			dpid = "00:00:00:00:00:00:00:01"
151			switch = get_switch_mock(dpid, 0x04)
152			intf_a = get_interface_mock("s1-eth1", 1, switch)
153			metadata = {
154			"state": "looped",
155			"port_numbers": [1, 2],
156			"updated_at": now().strftime("%Y-%m-%dT%H:%M:%S"),
157			"detected_at": now().strftime("%Y-%m-%dT%H:%M:%S"),
158			}
159			self.loop_manager.add_interface_metadata(intf_a.id, metadata)
160			assert mock_requests.post.call_count == 1
161
162			@patch("napps.kytos.of_lldp.loop_manager.requests")
163			def test_del_interface_metadata(self, mock_requests):
164			"""Test del interface metadata."""
165			dpid = "00:00:00:00:00:00:00:01"
166			switch = get_switch_mock(dpid, 0x04)
167			intf_a = get_interface_mock("s1-eth1", 1, switch)
168			self.loop_manager.del_interface_metadata(intf_a.id, "looped")
169			assert mock_requests.delete.call_count == 1
170
171			def test_publish_loop_state(self):
172			"""Test publish_loop_state."""
173			dpid = "00:00:00:00:00:00:00:01"
174			switch = get_switch_mock(dpid, 0x04)
175			intf_a = get_interface_mock("s1-eth1", 1, switch)
176			intf_b = get_interface_mock("s1-eth2", 2, switch)
177			state = "detected"
178
179			self.loop_manager.publish_loop_state(intf_a, intf_b, state)
180			assert self.loop_manager.controller.buffers.app.put.call_count == 1
181
182			def test_publish_loop_actions(self):
183			"""Test publish_loop_actions."""
184			dpid = "00:00:00:00:00:00:00:01"
185			switch = get_switch_mock(dpid, 0x04)
186			intf_a = get_interface_mock("s1-eth1", 1, switch)
187			intf_b = get_interface_mock("s1-eth2", 2, switch)
188			self.loop_manager.publish_loop_actions(intf_a, intf_b)
189			assert self.loop_manager.controller.buffers.app.put.call_count == len(
190			set(self.loop_manager.actions)
191			)
192
193			def test_get_stopped_loops(self):
194			"""Test get_stopped_loops."""
195			dpid = "00:00:00:00:00:00:00:01"
196			port_pairs = [(1, 2), (3, 3)]
197
198			delta = now() - timedelta(minutes=1)
199			looped_entry = {
200			"state": "detected",
201			"updated_at": delta.strftime("%Y-%m-%dT%H:%M:%S"),
202			}
203			for port_pair in port_pairs:
204			self.loop_manager.loop_state[dpid][port_pair] = looped_entry
205			assert self.loop_manager.get_stopped_loops() == {dpid: port_pairs}
206
207			@patch("napps.kytos.of_lldp.loop_manager.LoopManager.publish_loop_state")
208			@patch("napps.kytos.of_lldp.loop_manager.LoopManager.publish_loop_actions")
209			def test_process_if_looped(self, mock_publish_actions, mock_publish_state):
210			"""Test process_if_looped."""
211			dpid = "00:00:00:00:00:00:00:01"
212			switch = get_switch_mock(dpid, 0x04)
213			intf_a = get_interface_mock("s1-eth1", 1, switch)
214			intf_b = get_interface_mock("s1-eth2", 2, switch)
215			self.loop_manager.ignored_loops = {}
216			assert self.loop_manager.process_if_looped(intf_a, intf_b)
217			assert mock_publish_actions.call_count == 1
218			assert mock_publish_state.call_count == 1
219
220			def test_handle_topology_update(self):
221			"""Test handle_topology pudate."""
222			dpid = "00:00:00:00:00:00:00:01"
223			switch = get_switch_mock(dpid, 0x04)
224			mock_topo = MagicMock()
225			switch.metadata = {"ignored_loops": [[1, 2]]}
226			mock_topo.switches = {dpid: switch}
227
228			assert dpid not in self.loop_manager.ignored_loops
229			self.loop_manager.handle_topology_loaded(mock_topo)
230			assert self.loop_manager.ignored_loops[dpid] == [[1, 2]]
231
232			def test_handle_switch_metadata_changed_added(self):
233			"""Test handle_switch_metadata_changed added."""
234			dpid = "00:00:00:00:00:00:00:01"
235			switch = get_switch_mock(dpid, 0x04)
236			switch.metadata = {"ignored_loops": [[1, 2]]}
237
238			assert dpid not in self.loop_manager.ignored_loops
239			self.loop_manager.handle_switch_metadata_changed(switch)
240			assert self.loop_manager.ignored_loops[dpid] == [[1, 2]]
241
242			def test_handle_switch_metadata_changed_removed(self):
243			"""Test handle_switch_metadata_changed removed."""
244			dpid = "00:00:00:00:00:00:00:01"
245			switch = get_switch_mock(dpid, 0x04)
246			switch.id = dpid
247			switch.metadata = {"some_key": "some_value"}
248			self.loop_manager.ignored_loops[dpid] = [[1, 2]]
249
250			assert dpid in self.loop_manager.ignored_loops
251			self.loop_manager.handle_switch_metadata_changed(switch)
252			assert dpid not in self.loop_manager.ignored_loops
253

kytos-ng / of_lldp

Pull Request — master (#36)

TestLoopManager.setUp() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like