build.tests.unit.test_loop_manager.TestLoopManager.test_process_if_looped() - 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 14:58 UTC

TestLoopManager.test_process_if_looped() A

↳ Parent: build.tests.unit.test_loop_manager

Complexity

Conditions

Size

Total Lines	12
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	11
nop	3
dl	0
loc	12
rs	9.85
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


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

kytos-ng / of_lldp

Pull Request — master (#36)

TestLoopManager.test_process_if_looped() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like