build.tests.unit.test_liveness_manager.TestILSM.test_consume_hello() - Code Metrics - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

TestILSM.test_consume_hello() A
last analyzed 2025-07-23 14:51 UTC

↳ Parent: build.tests.unit.test_liveness_manager

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	6
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	2
dl	0
loc	7
ccs	6
cts	6
cp	1
crap	1
rs	10
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_link_status_reason_hook_liveness(self, liveness_manager) -> None:
        """Test link_status_reason_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_reason_hook_liveness(mock_link)
        assert status == frozenset({"liveness"})

        mock_link.metadata = {"liveness_status": "up"}
        status = liveness_manager.link_status_reason_hook_liveness(mock_link)
        assert status == frozenset()

    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_link_status_reason_hook_liveness(self, liveness_manager) -> None:
192		"""Test link_status_reason_hook_liveness."""
193	1	mock_link = MagicMock()
194	1	mock_link.is_active.return_value = True
195	1	mock_link.is_enabled.return_value = True
196	1	mock_link.metadata = {"liveness_status": "down"}
197	1	status = liveness_manager.link_status_reason_hook_liveness(mock_link)
198	1	assert status == frozenset({"liveness"})
199
200	1	mock_link.metadata = {"liveness_status": "up"}
201	1	status = liveness_manager.link_status_reason_hook_liveness(mock_link)
202	1	assert status == frozenset()
203
204	1	def test_try_to_publish_lsm_event(
205		self, liveness_manager, intf_one, intf_two
206		) -> None:
207		"""Test try_to_publish_lsm_event."""
208	1	event_suffix = None
209	1	liveness_manager.try_to_publish_lsm_event(
210		event_suffix, intf_one, intf_two
211		)
212	1	assert liveness_manager.controller.buffers.app.put.call_count == 0
213	1	event_suffix = "up"
214	1	liveness_manager.try_to_publish_lsm_event(
215		event_suffix, intf_one, intf_two
216		)
217	1	assert liveness_manager.controller.buffers.app.put.call_count == 1
218
219	1	async def test_atry_to_publish_lsm_event(
220		self, liveness_manager, intf_one, intf_two
221		) -> None:
222		"""Test atry_to_publish_lsm_event."""
223	1	liveness_manager.controller.buffers.app.aput = AsyncMock()
224	1	event_suffix = None
225	1	await liveness_manager.atry_to_publish_lsm_event(
226		event_suffix, intf_one, intf_two
227		)
228	1	assert liveness_manager.controller.buffers.app.aput.call_count == 0
229	1	event_suffix = "up"
230	1	await liveness_manager.atry_to_publish_lsm_event(
231		event_suffix, intf_one, intf_two
232		)
233	1	assert liveness_manager.controller.buffers.app.aput.call_count == 1
234
235	1	async def test_get_interface_status(
236		self, liveness_manager, intf_one, intf_two
237		) -> None:
238		"""Test get_interface_status."""
239	1	assert liveness_manager.get_interface_status(intf_one.id) == (
240		None,
241		None,
242		)
243	1	liveness_manager.enable(intf_one, intf_two)
244	1	assert liveness_manager.get_interface_status(intf_one.id) == (
245		"init",
246		None,
247		)
248	1	received_at = datetime.utcnow()
249	1	await liveness_manager.consume_hello(intf_one, intf_two, received_at)
250	1	assert liveness_manager.get_interface_status(intf_one.id) == (
251		"up",
252		received_at,
253		)
254
255	1	async def test_consume_hello(
256		self, liveness_manager, intf_one, intf_two
257		) -> None:
258		"""Test consume_hello."""
259	1	assert not liveness_manager.liveness
260	1	received_at = datetime.utcnow()
261	1	liveness_manager.atry_to_publish_lsm_event = AsyncMock()
262
263	1	await liveness_manager.consume_hello(intf_one, intf_two, received_at)
264	1	assert intf_one.id in liveness_manager.liveness
265	1	assert intf_two.id not in liveness_manager.liveness
266	1	entry = liveness_manager.liveness[intf_one.id]
267	1	assert entry["interface_a"] == intf_one
268	1	assert entry["interface_b"] == intf_two
269	1	assert entry["lsm"].ilsm_a.state == "up"
270	1	assert entry["lsm"].ilsm_b.state == "init"
271	1	assert entry["lsm"].state == "init"
272	1	assert liveness_manager.atry_to_publish_lsm_event.call_count == 1
273
274	1	received_at = datetime.utcnow()
275	1	await liveness_manager.consume_hello(intf_two, intf_one, received_at)
276	1	assert entry["lsm"].ilsm_a.state == "up"
277	1	assert entry["lsm"].ilsm_b.state == "up"
278	1	assert entry["lsm"].state == "up"
279	1	assert liveness_manager.atry_to_publish_lsm_event.call_count == 2
280
281	1	async def test_consume_hello_reinit(
282		self, liveness_manager, intf_one, intf_two, intf_three
283		) -> None:
284		"""Test consume_hello reinitialization, this test a corner
285		case where one end of the link has a new interface."""
286	1	assert not liveness_manager.liveness
287	1	received_at = datetime.utcnow()
288	1	liveness_manager.atry_to_publish_lsm_event = AsyncMock()
289
290	1	await liveness_manager.consume_hello(intf_one, intf_two, received_at)
291	1	await liveness_manager.consume_hello(intf_two, intf_one, received_at)
292	1	assert intf_one.id in liveness_manager.liveness
293	1	entry = liveness_manager.liveness[intf_one.id]
294	1	assert entry["interface_a"] == intf_one
295	1	assert entry["interface_b"] == intf_two
296	1	assert entry["lsm"].ilsm_a.state == "up"
297	1	assert entry["lsm"].ilsm_b.state == "up"
298	1	assert entry["lsm"].state == "up"
299	1	assert liveness_manager.atry_to_publish_lsm_event.call_count == 2
300
301	1	await liveness_manager.consume_hello(intf_one, intf_three, received_at)
302	1	entry = liveness_manager.liveness[intf_one.id]
303	1	assert entry["lsm"].ilsm_a.state == "up"
304	1	assert entry["lsm"].ilsm_b.state == "init"
305	1	assert entry["lsm"].state == "init"
306	1	assert liveness_manager.atry_to_publish_lsm_event.call_count == 3
307
308	1	def test_should_call_reaper(self, liveness_manager, intf_one) -> None:
309		"""Test should call reaper."""
310	1	intf_one.switch.is_connected = lambda: True
311	1	intf_one.lldp = True
312	1	liveness_manager.enable(intf_one)
313	1	assert liveness_manager.should_call_reaper(intf_one)
314
315	1	intf_one.lldp = False
316	1	assert not liveness_manager.should_call_reaper(intf_one)
317	1	intf_one.lldp = True
318	1	assert liveness_manager.should_call_reaper(intf_one)
319
320	1	liveness_manager.disable(intf_one)
321	1	assert not liveness_manager.should_call_reaper(intf_one)
322	1	liveness_manager.enable(intf_one)
323	1	assert liveness_manager.should_call_reaper(intf_one)
324
325	1	intf_one.switch.is_connected = lambda: False
326	1	assert not liveness_manager.should_call_reaper(intf_one)
327	1	intf_one.switch.is_connected = lambda: True
328	1	assert liveness_manager.should_call_reaper(intf_one)
329
330	1	def test_reaper(self, liveness_manager, lsm, intf_one, intf_two) -> None:
331		"""Test reaper."""
332	1	intf_one.status, intf_two.status = EntityStatus.UP, EntityStatus.UP
333	1	liveness_manager.liveness = {
334		intf_one.id: {
335		"interface_a": intf_one,
336		"interface_b": intf_two,
337		"lsm": lsm,
338		}
339		}
340	1	liveness_manager.should_call_reaper = MagicMock(return_value=True)
341	1	liveness_manager.try_to_publish_lsm_event = MagicMock()
342	1	lsm.ilsm_a.reaper_check = MagicMock()
343	1	lsm.ilsm_b.reaper_check = MagicMock()
344
345	1	dead_interval = 3
346	1	liveness_manager.reaper(dead_interval)
347
348	1	lsm.ilsm_a.reaper_check.assert_called_with(dead_interval)
349	1	lsm.ilsm_b.reaper_check.assert_called_with(dead_interval)
350	1	assert liveness_manager.try_to_publish_lsm_event.call_count == 1
351
352	1	async def test_consume_hello_if_enabled(self, liveness_manager) -> None:
353		"""Test test_consume_hello_if_enabled."""
354	1	liveness_manager.is_enabled = MagicMock(return_value=True)
355	1	liveness_manager.consume_hello = AsyncMock()
356	1	await liveness_manager.consume_hello_if_enabled(
357		MagicMock(), MagicMock()
358		)
359	1	assert liveness_manager.consume_hello.call_count == 1
360
361	1	liveness_manager.is_enabled = MagicMock(return_value=False)
362	1	await liveness_manager.consume_hello_if_enabled(
363		MagicMock(), MagicMock()
364		)
365		assert liveness_manager.consume_hello.call_count == 1
366

kytos-ng / of_lldp

TestILSM.test_consume_hello() A last analyzed 2025-07-23 14:51 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

TestILSM.test_consume_hello() A
last analyzed 2025-07-23 14:51 UTC