build.tests.unit.test_loop_manager.TestLoopManager.test_process_if_looped() - Code Metrics - Inspection of "Merge pull request #56 from kytos-ng/feature/liven..." - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 127cfb...14f54d )

by Vinicius

created 2022-07-27 13:31 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 AsyncMock, 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


async def test_publish_loop_state():
    """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"
    loop_manager = LoopManager(AsyncMock())
    await loop_manager.apublish_loop_state(intf_a, intf_b, state)
    assert loop_manager.controller.buffers.app.aput.call_count == 1


async def test_publish_loop_actions():
    """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)
    loop_manager = LoopManager(AsyncMock())
    await loop_manager.publish_loop_actions(intf_a, intf_b)
    assert loop_manager.controller.buffers.app.aput.call_count == len(
        set(loop_manager.actions)
    )


async def test_process_if_looped():
    """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)
    loop_manager = LoopManager(AsyncMock())
    loop_manager.ignored_loops = {}
    loop_manager.publish_loop_actions = AsyncMock()
    loop_manager.apublish_loop_state = AsyncMock()
    assert await loop_manager.process_if_looped(intf_a, intf_b)
    assert loop_manager.publish_loop_actions.call_count == 1
    assert loop_manager.apublish_loop_state.call_count == 1


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"]
        mock_requests.post.return_value = MagicMock(status_code=200)
        mock_requests.delete.return_value = MagicMock(status_code=200)
        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"

    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

    @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_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}

    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_incrementally(self):
        """Test handle_switch_metadata_changed incrementally."""
        dpid = "00:00:00:00:00:00:00:01"
        switch = get_switch_mock(dpid, 0x04)
        switch.id = dpid
        switch.metadata = {"ignored_loops": [[1, 2]]}
        self.loop_manager.ignored_loops = {}

        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]]

        switch.metadata = {"ignored_loops": [[1, 2], [3, 4]]}
        self.loop_manager.handle_switch_metadata_changed(switch)
        assert self.loop_manager.ignored_loops[dpid] == [[1, 2], [3, 4]]

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

kytos-ng / of_lldp

Push — master ( 127cfb...14f54d )

TestLoopManager.test_process_if_looped() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like