build.tests.unit.test_liveness_manager.TestLivenessManager.test_get_interface_status() - 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

TestLivenessManager.test_get_interface_status() A

↳ Parent: build.tests.unit.test_liveness_manager

Complexity

Conditions

Size

Total Lines	18
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	14
nop	4
dl	0
loc	18
rs	9.7
c	0
b	0
f	0

"""Test LivenessManager."""
# pylint: disable=no-self-use,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

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

kytos-ng / of_lldp

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

TestLivenessManager.test_get_interface_status() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like