Test periodic flap aging

tests/flaps_test.py

@@ -1,13 +1,27 @@
+import ipaddress
+import logging
+from asyncio import Future
 from datetime import timedelta
+from unittest.mock import Mock
 
+import pytest
+
+from zino import flaps
+from zino.config.models import PollDevice
 from zino.flaps import (
     FLAP_CEILING,
     FLAP_INIT_VAL,
     FLAP_MIN,
     FLAP_THRESHOLD,
     FlappingState,
     FlappingStates,
+    age_flapping_states,
+    age_single_interface_flapping_state,
+    stabilize_flapping_state,
 )
+from zino.state import ZinoState
+from zino.statemodels import EventState, FlapState, Port, PortStateEvent
+from zino.tasks.linkstatetask import LinkStateTask
 from zino.time import now
 
 
@@ -135,3 +149,173 @@ def test_when_flapping_stats_do_not_exist_get_flap_value_should_return_zero(self
         flapping_states = FlappingStates()
 
         assert flapping_states.get_flap_value(1) == 0
+
+
+class TestAgeSingleInterfaceFlappingState:
+    @pytest.mark.asyncio
+    async def test_it_should_decrease_hist_val(self, state_with_flapstats, polldevs_dict):
+        port: Port = next(iter(state_with_flapstats.devices.devices["localhost"].ports.values()))
+        flapping_state = state_with_flapstats.flapping.interfaces[("localhost", port.ifindex)]
+        initial = flapping_state.hist_val
+
+        await age_single_interface_flapping_state(
+            flapping_state, ("localhost", port.ifindex), state=state_with_flapstats, polldevs=polldevs_dict
+        )
+        assert flapping_state.hist_val < initial
+
+    @pytest.mark.asyncio
+    async def test_when_flap_is_below_threshold_it_should_remove_flapping_state(
+        self, mocked_out_poll_single_interface, state_with_flapstats, polldevs_dict
+    ):
+        port: Port = next(iter(state_with_flapstats.devices.devices["localhost"].ports.values()))
+        flapping_state = state_with_flapstats.flapping.interfaces[("localhost", port.ifindex)]
+        flapping_state.hist_val = FLAP_THRESHOLD
+
+        await age_single_interface_flapping_state(
+            flapping_state, ("localhost", port.ifindex), state=state_with_flapstats, polldevs=polldevs_dict
+        )
+
+        assert not state_with_flapstats.flapping.interfaces
+
+
+class TestStabilizeFlappingState:
+    @pytest.mark.asyncio
+    async def test_when_no_event_exists_it_should_create_an_event(
+        self, mocked_out_poll_single_interface, state_with_flapstats, polldevs_dict
+    ):
+        port: Port = next(iter(state_with_flapstats.devices.devices["localhost"].ports.values()))
+        flapping_state = state_with_flapstats.flapping.interfaces[("localhost", port.ifindex)]
+        flapping_state.hist_val = FLAP_THRESHOLD
+
+        assert len(state_with_flapstats.events.events) == 0
+
+        await stabilize_flapping_state(
+            flapping_state, ("localhost", port.ifindex), state=state_with_flapstats, polldevs=polldevs_dict
+        )
+
+        assert len(state_with_flapstats.events.events) > 0
+
+    @pytest.mark.asyncio
+    async def test_when_a_matching_event_exists_it_should_set_its_flapstate_to_stable(
+        self, mocked_out_poll_single_interface, state_with_flapstats, polldevs_dict
+    ):
+        port: Port = next(iter(state_with_flapstats.devices.devices["localhost"].ports.values()))
+        flapping_state = state_with_flapstats.flapping.interfaces[("localhost", port.ifindex)]
+        flapping_state.hist_val = FLAP_THRESHOLD
+
+        orig_event = state_with_flapstats.events.get_or_create_event("localhost", port.ifindex, PortStateEvent)
+        orig_event.flapstate = FlapState.FLAPPING
+        orig_event.flaps = 42
+        orig_event.port = port.ifdescr
+        orig_event.portstate = port.state
+        orig_event.router = "localhost"
+        orig_event.polladdr = "127.0.0.1"
+        orig_event.priority = 500
+        orig_event.ifindex = port.ifindex
+        orig_event.descr = port.ifalias
+
+        state_with_flapstats.events.commit(orig_event)
+
+        await stabilize_flapping_state(
+            flapping_state, ("localhost", port.ifindex), state=state_with_flapstats, polldevs=polldevs_dict
+        )
+
+        updated_event = state_with_flapstats.events.get_or_create_event("localhost", port.ifindex, PortStateEvent)
+        assert updated_event
+        assert updated_event.flapstate == FlapState.STABLE
+
+    @pytest.mark.asyncio
+    async def test_when_a_matching_closed_event_exists_it_should_set_its_flapstate_to_stable(
+        self, mocked_out_poll_single_interface, state_with_flapstats, polldevs_dict
+    ):
+        port: Port = next(iter(state_with_flapstats.devices.devices["localhost"].ports.values()))
+        flapping_state = state_with_flapstats.flapping.interfaces[("localhost", port.ifindex)]
+        flapping_state.hist_val = FLAP_THRESHOLD
+
+        orig_event = state_with_flapstats.events.get_or_create_event("localhost", port.ifindex, PortStateEvent)
+        orig_event.flapstate = FlapState.FLAPPING
+        orig_event.flaps = 42
+        orig_event.port = port.ifdescr
+        orig_event.portstate = port.state
+        orig_event.router = "localhost"
+        orig_event.polladdr = "127.0.0.1"
+        orig_event.priority = 500
+        orig_event.ifindex = port.ifindex
+        orig_event.descr = port.ifalias
+
+        state_with_flapstats.events.commit(orig_event)
+        orig_event.set_state(EventState.CLOSED)
+        state_with_flapstats.events.commit(orig_event)
+
+        await stabilize_flapping_state(
+            flapping_state, ("localhost", port.ifindex), state=state_with_flapstats, polldevs=polldevs_dict
+        )
+
+        closed_event = state_with_flapstats.events.get_closed_event("localhost", port.ifindex, PortStateEvent)
+        assert closed_event
+        assert closed_event.flapstate == FlapState.STABLE
+
+    @pytest.mark.asyncio
+    async def test_when_port_is_unkown_it_should_still_log_the_change(
+        self,
+        mocked_out_poll_single_interface,
+        state_with_flapstats,
+        polldevs_dict,
+        caplog,
+    ):
+        port: Port = next(iter(state_with_flapstats.devices.devices["localhost"].ports.values()))
+        flapping_state = state_with_flapstats.flapping.interfaces[("localhost", port.ifindex)]
+        flapping_state.hist_val = FLAP_THRESHOLD
+        fake_ifindex = port.ifindex + 666
+
+        with caplog.at_level(logging.INFO):
+            await stabilize_flapping_state(
+                flapping_state, ("localhost", fake_ifindex), state=state_with_flapstats, polldevs=polldevs_dict
+            )
+
+        assert f"{fake_ifindex} stopped flapping" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_age_flapping_states_should_age_all_flapping_states(monkeypatch, event_loop):
+    future = event_loop.create_future()
+    future.set_result(None)
+    mock_ager = Mock(return_value=future)
+    monkeypatch.setattr(flaps, "age_single_interface_flapping_state", mock_ager)
+
+    state = ZinoState()
+    state.flapping.first_flap(("localhost", 1))
+    state.flapping.first_flap(("localhost", 2))
+
+    await age_flapping_states(state, {})
+    assert mock_ager.call_count == 2
+
+
+@pytest.fixture
+def state_with_flapstats(state_with_localhost_with_port) -> ZinoState:
+    initial = FLAP_THRESHOLD * 2
+    port: Port = next(iter(state_with_localhost_with_port.devices.devices["localhost"].ports.values()))
+    last_change = now() - timedelta(minutes=1)
+    flapping_state = FlappingState(
+        hist_val=initial,
+        first_flap=last_change - timedelta(minutes=10),
+        last_flap=last_change,
+        last_age=last_change,
+    )
+    flapstates = FlappingStates(interfaces={("localhost", port.ifindex): flapping_state})
+    state_with_localhost_with_port.flapping = flapstates
+    return state_with_localhost_with_port
+
+
+@pytest.fixture
+def polldevs_dict(polldevs_conf_with_single_router) -> dict[str, PollDevice]:
+    return {"localhost": PollDevice(name="localhost", address=ipaddress.IPv4Address("127.0.0.1"))}
+
+
+@pytest.fixture
+def mocked_out_poll_single_interface(monkeypatch):
+    """Monkey patches LinkStateTask.poll_single_interface to do essentially nothing"""
+    future = Future()
+    future.set_result(None)
+    monkeypatch.setattr(LinkStateTask, "poll_single_interface", future)
+    yield monkeypatch