build.tests.unit.test_liveness_manager.TestILSM.test_min_hellos() - Code Metrics - Inspection of "Merge pull request #107 from kytos-ng/fix/liveness..." - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ad8a25...eaed3c )

by Vinicius

created 2024-10-16 15:12 UTC

TestILSM.test_min_hellos() A

↳ Parent: build.tests.unit.test_liveness_manager

Complexity

Conditions

Size

Total Lines	21
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	17
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
eloc	17
nop	1
dl	0
loc	21
ccs	17
cts	17
cp	1
crap	2
rs	9.55
c	0
b	0
f	0

"""Test LivenessManager."""
# pylint: disable=invalid-name,protected-access
from datetime import datetime, timedelta
from unittest.mock import AsyncMock, MagicMock

import pytest

from kytos.core.common import EntityStatus
from napps.kytos.of_lldp.managers.liveness import ILSM, LSM


class TestILSM:

    """TestILSM."""

    @pytest.mark.parametrize(
        "from_state,to",
        [
            ("init", "up"),
            ("init", "down"),
            ("up", "down"),
            ("up", "init"),
            ("down", "up"),
            ("down", "init"),
        ],
    )
    def test_ilsm_transitions(self, from_state, to) -> None:
        """Test ILSM transitions."""
        ilsm = ILSM(state=from_state)
        assert ilsm.state == from_state
        assert ilsm.transition_to(to) == to
        assert ilsm.state == to

    def test_ilsm_invalid_transition(self) -> None:
        """Test ILSM invalid transition."""
        ilsm = ILSM(state="down")
        assert ilsm.state == "down"
        assert not ilsm.transition_to("down")
        assert not ilsm.transition_to("invalid_state")
        assert ilsm.state == "down"

    def test_repr(self, ilsm) -> None:
        """Test repr."""
        assert str(ilsm) == "ILSM(init, None)"

    def test_consume_hello(self, ilsm) -> None:
        """Test consume_hello."""
        assert ilsm.state == "init"
        received_at = datetime.utcnow()
        assert ilsm.consume_hello(received_at) == "up"
        assert ilsm.state == "up"
        assert ilsm.last_hello_at == received_at

    def test_min_hellos(self) -> None:
        """Test consume_hello min hellos."""
        min_hellos = 3
        assert min_hellos >= 2
        ilsm = ILSM(state="init", min_hellos=min_hellos)
        assert ilsm.state == "init"
        assert ilsm.hello_counter == 0

        # it shouldn't transition on the first min_hellos - 1
        for i in range(min_hellos - 1):
            received_at = datetime.utcnow()
            assert not ilsm.consume_hello(received_at)
            assert ilsm.state == "init"
            assert ilsm.last_hello_at == received_at
            assert ilsm.hello_counter == i + 1

        received_at = datetime.utcnow()
        assert ilsm.consume_hello(received_at) == "up"
        assert ilsm.state == "up"
        assert ilsm.last_hello_at == received_at
        assert ilsm.hello_counter == 0

    @pytest.mark.parametrize(
        "delta_secs, expected_state", [(0, "up"), (9, "up"), (10, "down")]
    )
    def test_reaper_check(self, ilsm, delta_secs, expected_state) -> None:
        """Test reaper_check."""
        assert ilsm.state == "init"
        dead_interval = 9
        delta = timedelta(seconds=delta_secs)
        received_at = datetime.utcnow() - delta
        ilsm.consume_hello(received_at)
        ilsm.reaper_check(dead_interval)
        assert ilsm.state == expected_state


class TestLSM:

    """Test LSM."""

    def test_rpr(self, lsm) -> None:
        """Test repr."""
        assert str(lsm) == f"LSM(init, {str(lsm.ilsm_a)}, {str(lsm.ilsm_b)})"

    @pytest.mark.parametrize(
        "ilsm_a_state,ilsm_b_state,expected",
        [
            ("init", "dontcare", "init"),
            ("dontcare", "init", "init"),
            ("down", "dontcare", "down"),
            ("dontcare", "down", "down"),
            ("up", "init", "init"),
            ("init", "up", "init"),
            ("up", "up", "up"),
        ],
    )
    def test_agg_state(
        self, ilsm_a_state, ilsm_b_state, expected, lsm
    ) -> None:
        """Test aggregated state."""
        lsm.ilsm_a.state, lsm.ilsm_b.state = ilsm_a_state, ilsm_b_state
        assert lsm.agg_state() == expected

    @pytest.mark.parametrize(
        "from_state,to",
        [
            ("init", "up"),
            ("init", "down"),
            ("up", "down"),
            ("up", "init"),
            ("down", "up"),
            ("down", "init"),
        ],
    )
    def test_lsm_transitions(self, from_state, to) -> None:
        """Test LSM transitions."""
        lsm = LSM(ILSM(from_state), ILSM(from_state), from_state)
        assert lsm._transition_to(to) == to
        assert lsm.state == to

    def test_next_state(self, lsm) -> None:
        """Test next_state."""
        lsm.ilsm_a.transition_to("up")
        lsm.ilsm_b.transition_to("up")
        assert lsm.next_state() == "up"
        assert lsm.state == "up"

        assert not lsm.next_state()
        assert lsm.state == "up"

        lsm.ilsm_a.transition_to("down")
        assert lsm.next_state() == "down"
        assert lsm.state == "down"

        assert not lsm.next_state()
        assert lsm.state == "down"


class TestLivenessManager:

    """TestLivenessManager."""

    def test_is_enabled(self, liveness_manager, intf_one) -> None:
        """Test is_enabled."""
        assert not liveness_manager.interfaces
        assert not liveness_manager.is_enabled(intf_one)
        liveness_manager.interfaces[intf_one.id] = intf_one
        assert liveness_manager.is_enabled(intf_one)

    def test_enable(self, liveness_manager, intf_one, intf_two) -> None:
        """Test enable."""
        assert not liveness_manager.interfaces
        liveness_manager.enable(intf_one, intf_two)
        assert intf_one.id in liveness_manager.interfaces
        assert intf_two.id in liveness_manager.interfaces

    def test_disable(self, liveness_manager, intf_one, intf_two) -> None:
        """Test disable."""
        assert not liveness_manager.interfaces
        liveness_manager.enable(intf_one, intf_two)
        assert intf_one.id in liveness_manager.interfaces
        liveness_manager.disable(intf_one, intf_two)
        assert intf_one.id not in liveness_manager.interfaces

    def test_link_status_hook_liveness(self, liveness_manager) -> None:
        """Test link_status_hook_liveness."""
        mock_link = MagicMock()
        mock_link.is_active.return_value = True
        mock_link.is_enabled.return_value = True
        mock_link.metadata = {"liveness_status": "down"}
        status = liveness_manager.link_status_hook_liveness(mock_link)
        assert status == EntityStatus.DOWN

        mock_link.metadata = {"liveness_status": "up"}
        status = liveness_manager.link_status_hook_liveness(mock_link)
        assert status is None

    def test_try_to_publish_lsm_event(
        self, liveness_manager, intf_one, intf_two
    ) -> None:
        """Test try_to_publish_lsm_event."""
        event_suffix = None
        liveness_manager.try_to_publish_lsm_event(
            event_suffix, intf_one, intf_two
        )
        assert liveness_manager.controller.buffers.app.put.call_count == 0
        event_suffix = "up"
        liveness_manager.try_to_publish_lsm_event(
            event_suffix, intf_one, intf_two
        )
        assert liveness_manager.controller.buffers.app.put.call_count == 1

    async def test_atry_to_publish_lsm_event(
        self, liveness_manager, intf_one, intf_two
    ) -> None:
        """Test atry_to_publish_lsm_event."""
        liveness_manager.controller.buffers.app.aput = AsyncMock()
        event_suffix = None
        await liveness_manager.atry_to_publish_lsm_event(
            event_suffix, intf_one, intf_two
        )
        assert liveness_manager.controller.buffers.app.aput.call_count == 0
        event_suffix = "up"
        await liveness_manager.atry_to_publish_lsm_event(
            event_suffix, intf_one, intf_two
        )
        assert liveness_manager.controller.buffers.app.aput.call_count == 1

    async def test_get_interface_status(
        self, liveness_manager, intf_one, intf_two
    ) -> None:
        """Test get_interface_status."""
        assert liveness_manager.get_interface_status(intf_one.id) == (
            None,
            None,
        )
        liveness_manager.enable(intf_one, intf_two)
        assert liveness_manager.get_interface_status(intf_one.id) == (
            "init",
            None,
        )
        received_at = datetime.utcnow()
        await liveness_manager.consume_hello(intf_one, intf_two, received_at)
        assert liveness_manager.get_interface_status(intf_one.id) == (
            "up",
            received_at,
        )

    async def test_consume_hello(
        self, liveness_manager, intf_one, intf_two
    ) -> None:
        """Test consume_hello."""
        assert not liveness_manager.liveness
        received_at = datetime.utcnow()
        liveness_manager.atry_to_publish_lsm_event = AsyncMock()

        await liveness_manager.consume_hello(intf_one, intf_two, received_at)
        assert intf_one.id in liveness_manager.liveness
        assert intf_two.id not in liveness_manager.liveness
        entry = liveness_manager.liveness[intf_one.id]
        assert entry["interface_a"] == intf_one
        assert entry["interface_b"] == intf_two
        assert entry["lsm"].ilsm_a.state == "up"
        assert entry["lsm"].ilsm_b.state == "init"
        assert entry["lsm"].state == "init"
        assert liveness_manager.atry_to_publish_lsm_event.call_count == 1

        received_at = datetime.utcnow()
        await liveness_manager.consume_hello(intf_two, intf_one, received_at)
        assert entry["lsm"].ilsm_a.state == "up"
        assert entry["lsm"].ilsm_b.state == "up"
        assert entry["lsm"].state == "up"
        assert liveness_manager.atry_to_publish_lsm_event.call_count == 2

    async def test_consume_hello_reinit(
        self, liveness_manager, intf_one, intf_two, intf_three
    ) -> None:
        """Test consume_hello reinitialization, this test a corner
        case where one end of the link has a new interface."""
        assert not liveness_manager.liveness
        received_at = datetime.utcnow()
        liveness_manager.atry_to_publish_lsm_event = AsyncMock()

        await liveness_manager.consume_hello(intf_one, intf_two, received_at)
        await liveness_manager.consume_hello(intf_two, intf_one, received_at)
        assert intf_one.id in liveness_manager.liveness
        entry = liveness_manager.liveness[intf_one.id]
        assert entry["interface_a"] == intf_one
        assert entry["interface_b"] == intf_two
        assert entry["lsm"].ilsm_a.state == "up"
        assert entry["lsm"].ilsm_b.state == "up"
        assert entry["lsm"].state == "up"
        assert liveness_manager.atry_to_publish_lsm_event.call_count == 2

        await liveness_manager.consume_hello(intf_one, intf_three, received_at)
        entry = liveness_manager.liveness[intf_one.id]
        assert entry["lsm"].ilsm_a.state == "up"
        assert entry["lsm"].ilsm_b.state == "init"
        assert entry["lsm"].state == "init"
        assert liveness_manager.atry_to_publish_lsm_event.call_count == 3

    def test_should_call_reaper(self, liveness_manager, intf_one) -> None:
        """Test should call reaper."""
        intf_one.switch.is_connected = lambda: True
        intf_one.lldp = True
        liveness_manager.enable(intf_one)
        assert liveness_manager.should_call_reaper(intf_one)

        intf_one.lldp = False
        assert not liveness_manager.should_call_reaper(intf_one)
        intf_one.lldp = True
        assert liveness_manager.should_call_reaper(intf_one)

        liveness_manager.disable(intf_one)
        assert not liveness_manager.should_call_reaper(intf_one)
        liveness_manager.enable(intf_one)
        assert liveness_manager.should_call_reaper(intf_one)

        intf_one.switch.is_connected = lambda: False
        assert not liveness_manager.should_call_reaper(intf_one)
        intf_one.switch.is_connected = lambda: True
        assert liveness_manager.should_call_reaper(intf_one)

    def test_reaper(self, liveness_manager, lsm, intf_one, intf_two) -> None:
        """Test reaper."""
        intf_one.status, intf_two.status = EntityStatus.UP, EntityStatus.UP
        liveness_manager.liveness = {
            intf_one.id: {
                "interface_a": intf_one,
                "interface_b": intf_two,
                "lsm": lsm,
            }
        }
        liveness_manager.should_call_reaper = MagicMock(return_value=True)
        liveness_manager.try_to_publish_lsm_event = MagicMock()
        lsm.ilsm_a.reaper_check = MagicMock()
        lsm.ilsm_b.reaper_check = MagicMock()

        dead_interval = 3
        liveness_manager.reaper(dead_interval)

        lsm.ilsm_a.reaper_check.assert_called_with(dead_interval)
        lsm.ilsm_b.reaper_check.assert_called_with(dead_interval)
        assert liveness_manager.try_to_publish_lsm_event.call_count == 1

    async def test_consume_hello_if_enabled(self, liveness_manager) -> None:
        """Test test_consume_hello_if_enabled."""
        liveness_manager.is_enabled = MagicMock(return_value=True)
        liveness_manager.consume_hello = AsyncMock()
        await liveness_manager.consume_hello_if_enabled(
            MagicMock(), MagicMock()
        )
        assert liveness_manager.consume_hello.call_count == 1

        liveness_manager.is_enabled = MagicMock(return_value=False)
        await liveness_manager.consume_hello_if_enabled(
            MagicMock(), MagicMock()
        )
        assert liveness_manager.consume_hello.call_count == 1


1		"""Test LivenessManager."""
2		# pylint: disable=invalid-name,protected-access
3	1	from datetime import datetime, timedelta
4	1	from unittest.mock import AsyncMock, MagicMock
5
6	1	import pytest
7
8	1	from kytos.core.common import EntityStatus
9	1	from napps.kytos.of_lldp.managers.liveness import ILSM, LSM
10
11
12	1	class TestILSM:
13
14		"""TestILSM."""
15
16	1	@pytest.mark.parametrize(
17		"from_state,to",
18		[
19		("init", "up"),
20		("init", "down"),
21		("up", "down"),
22		("up", "init"),
23		("down", "up"),
24		("down", "init"),
25		],
26		)
27	1	def test_ilsm_transitions(self, from_state, to) -> None:
28		"""Test ILSM transitions."""
29	1	ilsm = ILSM(state=from_state)
30	1	assert ilsm.state == from_state
31	1	assert ilsm.transition_to(to) == to
32	1	assert ilsm.state == to
33
34	1	def test_ilsm_invalid_transition(self) -> None:
35		"""Test ILSM invalid transition."""
36	1	ilsm = ILSM(state="down")
37	1	assert ilsm.state == "down"
38	1	assert not ilsm.transition_to("down")
39	1	assert not ilsm.transition_to("invalid_state")
40	1	assert ilsm.state == "down"
41
42	1	def test_repr(self, ilsm) -> None:
43		"""Test repr."""
44	1	assert str(ilsm) == "ILSM(init, None)"
45
46	1	def test_consume_hello(self, ilsm) -> None:
47		"""Test consume_hello."""
48	1	assert ilsm.state == "init"
49	1	received_at = datetime.utcnow()
50	1	assert ilsm.consume_hello(received_at) == "up"
51	1	assert ilsm.state == "up"
52	1	assert ilsm.last_hello_at == received_at
53
54	1	def test_min_hellos(self) -> None:
55		"""Test consume_hello min hellos."""
56	1	min_hellos = 3
57	1	assert min_hellos >= 2
58	1	ilsm = ILSM(state="init", min_hellos=min_hellos)
59	1	assert ilsm.state == "init"
60	1	assert ilsm.hello_counter == 0
61
62		# it shouldn't transition on the first min_hellos - 1
63	1	for i in range(min_hellos - 1):
64	1	received_at = datetime.utcnow()
65	1	assert not ilsm.consume_hello(received_at)
66	1	assert ilsm.state == "init"
67	1	assert ilsm.last_hello_at == received_at
68	1	assert ilsm.hello_counter == i + 1
69
70	1	received_at = datetime.utcnow()
71	1	assert ilsm.consume_hello(received_at) == "up"
72	1	assert ilsm.state == "up"
73	1	assert ilsm.last_hello_at == received_at
74	1	assert ilsm.hello_counter == 0
75
76	1	@pytest.mark.parametrize(
77		"delta_secs, expected_state", [(0, "up"), (9, "up"), (10, "down")]
78		)
79	1	def test_reaper_check(self, ilsm, delta_secs, expected_state) -> None:
80		"""Test reaper_check."""
81	1	assert ilsm.state == "init"
82	1	dead_interval = 9
83	1	delta = timedelta(seconds=delta_secs)
84	1	received_at = datetime.utcnow() - delta
85	1	ilsm.consume_hello(received_at)
86	1	ilsm.reaper_check(dead_interval)
87	1	assert ilsm.state == expected_state
88
89
90	1	class TestLSM:
91
92		"""Test LSM."""
93
94	1	def test_rpr(self, lsm) -> None:
95		"""Test repr."""
96	1	assert str(lsm) == f"LSM(init, {str(lsm.ilsm_a)}, {str(lsm.ilsm_b)})"
97
98	1	@pytest.mark.parametrize(
99		"ilsm_a_state,ilsm_b_state,expected",
100		[
101		("init", "dontcare", "init"),
102		("dontcare", "init", "init"),
103		("down", "dontcare", "down"),
104		("dontcare", "down", "down"),
105		("up", "init", "init"),
106		("init", "up", "init"),
107		("up", "up", "up"),
108		],
109		)
110	1	def test_agg_state(
111		self, ilsm_a_state, ilsm_b_state, expected, lsm
112		) -> None:
113		"""Test aggregated state."""
114	1	lsm.ilsm_a.state, lsm.ilsm_b.state = ilsm_a_state, ilsm_b_state
115	1	assert lsm.agg_state() == expected
116
117	1	@pytest.mark.parametrize(
118		"from_state,to",
119		[
120		("init", "up"),
121		("init", "down"),
122		("up", "down"),
123		("up", "init"),
124		("down", "up"),
125		("down", "init"),
126		],
127		)
128	1	def test_lsm_transitions(self, from_state, to) -> None:
129		"""Test LSM transitions."""
130	1	lsm = LSM(ILSM(from_state), ILSM(from_state), from_state)
131	1	assert lsm._transition_to(to) == to
132	1	assert lsm.state == to
133
134	1	def test_next_state(self, lsm) -> None:
135		"""Test next_state."""
136	1	lsm.ilsm_a.transition_to("up")
137	1	lsm.ilsm_b.transition_to("up")
138	1	assert lsm.next_state() == "up"
139	1	assert lsm.state == "up"
140
141	1	assert not lsm.next_state()
142	1	assert lsm.state == "up"
143
144	1	lsm.ilsm_a.transition_to("down")
145	1	assert lsm.next_state() == "down"
146	1	assert lsm.state == "down"
147
148	1	assert not lsm.next_state()
149	1	assert lsm.state == "down"
150
151
152	1	class TestLivenessManager:
153
154		"""TestLivenessManager."""
155
156	1	def test_is_enabled(self, liveness_manager, intf_one) -> None:
157		"""Test is_enabled."""
158	1	assert not liveness_manager.interfaces
159	1	assert not liveness_manager.is_enabled(intf_one)
160	1	liveness_manager.interfaces[intf_one.id] = intf_one
161	1	assert liveness_manager.is_enabled(intf_one)
162
163	1	def test_enable(self, liveness_manager, intf_one, intf_two) -> None:
164		"""Test enable."""
165	1	assert not liveness_manager.interfaces
166	1	liveness_manager.enable(intf_one, intf_two)
167	1	assert intf_one.id in liveness_manager.interfaces
168	1	assert intf_two.id in liveness_manager.interfaces
169
170	1	def test_disable(self, liveness_manager, intf_one, intf_two) -> None:
171		"""Test disable."""
172	1	assert not liveness_manager.interfaces
173	1	liveness_manager.enable(intf_one, intf_two)
174	1	assert intf_one.id in liveness_manager.interfaces
175	1	liveness_manager.disable(intf_one, intf_two)
176	1	assert intf_one.id not in liveness_manager.interfaces
177
178	1	def test_link_status_hook_liveness(self, liveness_manager) -> None:
179		"""Test link_status_hook_liveness."""
180	1	mock_link = MagicMock()
181	1	mock_link.is_active.return_value = True
182	1	mock_link.is_enabled.return_value = True
183	1	mock_link.metadata = {"liveness_status": "down"}
184	1	status = liveness_manager.link_status_hook_liveness(mock_link)
185	1	assert status == EntityStatus.DOWN
186
187	1	mock_link.metadata = {"liveness_status": "up"}
188	1	status = liveness_manager.link_status_hook_liveness(mock_link)
189	1	assert status is None
190
191	1	def test_try_to_publish_lsm_event(
192		self, liveness_manager, intf_one, intf_two
193		) -> None:
194		"""Test try_to_publish_lsm_event."""
195	1	event_suffix = None
196	1	liveness_manager.try_to_publish_lsm_event(
197		event_suffix, intf_one, intf_two
198		)
199	1	assert liveness_manager.controller.buffers.app.put.call_count == 0
200	1	event_suffix = "up"
201	1	liveness_manager.try_to_publish_lsm_event(
202		event_suffix, intf_one, intf_two
203		)
204	1	assert liveness_manager.controller.buffers.app.put.call_count == 1
205
206	1	async def test_atry_to_publish_lsm_event(
207		self, liveness_manager, intf_one, intf_two
208		) -> None:
209		"""Test atry_to_publish_lsm_event."""
210	1	liveness_manager.controller.buffers.app.aput = AsyncMock()
211	1	event_suffix = None
212	1	await liveness_manager.atry_to_publish_lsm_event(
213		event_suffix, intf_one, intf_two
214		)
215	1	assert liveness_manager.controller.buffers.app.aput.call_count == 0
216	1	event_suffix = "up"
217	1	await liveness_manager.atry_to_publish_lsm_event(
218		event_suffix, intf_one, intf_two
219		)
220	1	assert liveness_manager.controller.buffers.app.aput.call_count == 1
221
222	1	async def test_get_interface_status(
223		self, liveness_manager, intf_one, intf_two
224		) -> None:
225		"""Test get_interface_status."""
226	1	assert liveness_manager.get_interface_status(intf_one.id) == (
227		None,
228		None,
229		)
230	1	liveness_manager.enable(intf_one, intf_two)
231	1	assert liveness_manager.get_interface_status(intf_one.id) == (
232		"init",
233		None,
234		)
235	1	received_at = datetime.utcnow()
236	1	await liveness_manager.consume_hello(intf_one, intf_two, received_at)
237	1	assert liveness_manager.get_interface_status(intf_one.id) == (
238		"up",
239		received_at,
240		)
241
242	1	async def test_consume_hello(
243		self, liveness_manager, intf_one, intf_two
244		) -> None:
245		"""Test consume_hello."""
246	1	assert not liveness_manager.liveness
247	1	received_at = datetime.utcnow()
248	1	liveness_manager.atry_to_publish_lsm_event = AsyncMock()
249
250	1	await liveness_manager.consume_hello(intf_one, intf_two, received_at)
251	1	assert intf_one.id in liveness_manager.liveness
252	1	assert intf_two.id not in liveness_manager.liveness
253	1	entry = liveness_manager.liveness[intf_one.id]
254	1	assert entry["interface_a"] == intf_one
255	1	assert entry["interface_b"] == intf_two
256	1	assert entry["lsm"].ilsm_a.state == "up"
257	1	assert entry["lsm"].ilsm_b.state == "init"
258	1	assert entry["lsm"].state == "init"
259	1	assert liveness_manager.atry_to_publish_lsm_event.call_count == 1
260
261	1	received_at = datetime.utcnow()
262	1	await liveness_manager.consume_hello(intf_two, intf_one, received_at)
263	1	assert entry["lsm"].ilsm_a.state == "up"
264	1	assert entry["lsm"].ilsm_b.state == "up"
265	1	assert entry["lsm"].state == "up"
266	1	assert liveness_manager.atry_to_publish_lsm_event.call_count == 2
267
268	1	async def test_consume_hello_reinit(
269		self, liveness_manager, intf_one, intf_two, intf_three
270		) -> None:
271		"""Test consume_hello reinitialization, this test a corner
272		case where one end of the link has a new interface."""
273	1	assert not liveness_manager.liveness
274	1	received_at = datetime.utcnow()
275	1	liveness_manager.atry_to_publish_lsm_event = AsyncMock()
276
277	1	await liveness_manager.consume_hello(intf_one, intf_two, received_at)
278	1	await liveness_manager.consume_hello(intf_two, intf_one, received_at)
279	1	assert intf_one.id in liveness_manager.liveness
280	1	entry = liveness_manager.liveness[intf_one.id]
281	1	assert entry["interface_a"] == intf_one
282	1	assert entry["interface_b"] == intf_two
283	1	assert entry["lsm"].ilsm_a.state == "up"
284	1	assert entry["lsm"].ilsm_b.state == "up"
285	1	assert entry["lsm"].state == "up"
286	1	assert liveness_manager.atry_to_publish_lsm_event.call_count == 2
287
288	1	await liveness_manager.consume_hello(intf_one, intf_three, received_at)
289	1	entry = liveness_manager.liveness[intf_one.id]
290	1	assert entry["lsm"].ilsm_a.state == "up"
291	1	assert entry["lsm"].ilsm_b.state == "init"
292	1	assert entry["lsm"].state == "init"
293	1	assert liveness_manager.atry_to_publish_lsm_event.call_count == 3
294
295	1	def test_should_call_reaper(self, liveness_manager, intf_one) -> None:
296		"""Test should call reaper."""
297	1	intf_one.switch.is_connected = lambda: True
298	1	intf_one.lldp = True
299	1	liveness_manager.enable(intf_one)
300	1	assert liveness_manager.should_call_reaper(intf_one)
301
302	1	intf_one.lldp = False
303	1	assert not liveness_manager.should_call_reaper(intf_one)
304	1	intf_one.lldp = True
305	1	assert liveness_manager.should_call_reaper(intf_one)
306
307	1	liveness_manager.disable(intf_one)
308	1	assert not liveness_manager.should_call_reaper(intf_one)
309	1	liveness_manager.enable(intf_one)
310	1	assert liveness_manager.should_call_reaper(intf_one)
311
312	1	intf_one.switch.is_connected = lambda: False
313	1	assert not liveness_manager.should_call_reaper(intf_one)
314	1	intf_one.switch.is_connected = lambda: True
315	1	assert liveness_manager.should_call_reaper(intf_one)
316
317	1	def test_reaper(self, liveness_manager, lsm, intf_one, intf_two) -> None:
318		"""Test reaper."""
319	1	intf_one.status, intf_two.status = EntityStatus.UP, EntityStatus.UP
320	1	liveness_manager.liveness = {
321		intf_one.id: {
322		"interface_a": intf_one,
323		"interface_b": intf_two,
324		"lsm": lsm,
325		}
326		}
327	1	liveness_manager.should_call_reaper = MagicMock(return_value=True)
328	1	liveness_manager.try_to_publish_lsm_event = MagicMock()
329	1	lsm.ilsm_a.reaper_check = MagicMock()
330	1	lsm.ilsm_b.reaper_check = MagicMock()
331
332	1	dead_interval = 3
333	1	liveness_manager.reaper(dead_interval)
334
335	1	lsm.ilsm_a.reaper_check.assert_called_with(dead_interval)
336	1	lsm.ilsm_b.reaper_check.assert_called_with(dead_interval)
337	1	assert liveness_manager.try_to_publish_lsm_event.call_count == 1
338
339	1	async def test_consume_hello_if_enabled(self, liveness_manager) -> None:
340		"""Test test_consume_hello_if_enabled."""
341	1	liveness_manager.is_enabled = MagicMock(return_value=True)
342	1	liveness_manager.consume_hello = AsyncMock()
343	1	await liveness_manager.consume_hello_if_enabled(
344		MagicMock(), MagicMock()
345		)
346	1	assert liveness_manager.consume_hello.call_count == 1
347
348	1	liveness_manager.is_enabled = MagicMock(return_value=False)
349	1	await liveness_manager.consume_hello_if_enabled(
350		MagicMock(), MagicMock()
351		)
352		assert liveness_manager.consume_hello.call_count == 1
353

kytos-ng / of_lldp

Push — master ( ad8a25...eaed3c )

TestILSM.test_min_hellos() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like