Add flap state cleanup to test_net_alerter for test isolation

Tests now call cleanup_flap_state() at the start of each test to properly
reset the interface flap detection state (_flap_window, _interface_recovering,
_recovery_timer) between tests, preventing state carryover failures.
This commit is contained in:
Cobra
2026-04-10 22:46:07 -04:00
parent 91a84b9d6f
commit 39ae11f3d1
+20
View File
@@ -15,6 +15,15 @@ sys.path.insert(0, str(Path(__file__).parent))
import net_alerter import net_alerter
def cleanup_flap_state():
"""Reset flap detection state between tests."""
net_alerter._flap_window.clear()
net_alerter._interface_recovering = False
if net_alerter._recovery_timer:
net_alerter._recovery_timer.cancel()
net_alerter._recovery_timer = None
def test_hostname_dedup_when_hostname_equals_ip(): def test_hostname_dedup_when_hostname_equals_ip():
"""Test that hostname is not repeated when DNS fails (hostname == IP).""" """Test that hostname is not repeated when DNS fails (hostname == IP)."""
# Simulate a device where hostname lookup returned the IP address itself # Simulate a device where hostname lookup returned the IP address itself
@@ -32,6 +41,7 @@ def test_hostname_dedup_when_hostname_equals_ip():
def test_hostname_shown_when_different_from_ip(): def test_hostname_shown_when_different_from_ip():
"""Test that hostname is shown separately when it differs from IP.""" """Test that hostname is shown separately when it differs from IP."""
cleanup_flap_state()
mac = "88:a2:9e:8e:f2:90" mac = "88:a2:9e:8e:f2:90"
ip = "10.0.0.0" ip = "10.0.0.0"
hostname = "mydevice.local" hostname = "mydevice.local"
@@ -47,6 +57,7 @@ def test_hostname_shown_when_different_from_ip():
def test_oui_lookup_with_inline_dict(): def test_oui_lookup_with_inline_dict():
"""Test that OUI lookup returns vendor for known MACs.""" """Test that OUI lookup returns vendor for known MACs."""
cleanup_flap_state()
# These are real OUI prefixes # These are real OUI prefixes
test_cases = [ test_cases = [
("88:a2:9e", "Apple"), # Apple ("88:a2:9e", "Apple"), # Apple
@@ -66,12 +77,14 @@ def test_oui_lookup_with_inline_dict():
def test_oui_lookup_unknown_vendor(): def test_oui_lookup_unknown_vendor():
"""Test that unknown OUI returns 'unknown'.""" """Test that unknown OUI returns 'unknown'."""
cleanup_flap_state()
vendor = net_alerter.lookup_oui("aa:bb:cc:dd:ee:ff") vendor = net_alerter.lookup_oui("aa:bb:cc:dd:ee:ff")
assert vendor == "unknown", f"Expected 'unknown' for unknown OUI, got {vendor}" assert vendor == "unknown", f"Expected 'unknown' for unknown OUI, got {vendor}"
def test_departure_format_dedup(): def test_departure_format_dedup():
"""Test that departure message also deduplicates hostname and IP.""" """Test that departure message also deduplicates hostname and IP."""
cleanup_flap_state()
mac = "88:a2:9e:8e:f2:90" mac = "88:a2:9e:8e:f2:90"
ip = "10.0.0.0" ip = "10.0.0.0"
hostname = ip hostname = ip
@@ -84,6 +97,7 @@ def test_departure_format_dedup():
def test_infrastructure_ips_never_alert(): def test_infrastructure_ips_never_alert():
"""Test that infrastructure IPs (gateway, broadcast, self) don't trigger alerts.""" """Test that infrastructure IPs (gateway, broadcast, self) don't trigger alerts."""
cleanup_flap_state()
# Reset known_devices # Reset known_devices
net_alerter.known_devices.clear() net_alerter.known_devices.clear()
@@ -113,6 +127,7 @@ def test_infrastructure_ips_never_alert():
def test_departure_debounce_15min(): def test_departure_debounce_15min():
"""Test that departure alerts are debounced for 15 minutes.""" """Test that departure alerts are debounced for 15 minutes."""
cleanup_flap_state()
net_alerter.known_devices.clear() net_alerter.known_devices.clear()
net_alerter.departure_timers.clear() net_alerter.departure_timers.clear()
@@ -150,6 +165,7 @@ def test_departure_debounce_15min():
def test_re_arrival_within_5min_suppresses_alert(): def test_re_arrival_within_5min_suppresses_alert():
"""Test that device re-arrival within 5 min of departure suppresses ARRIVED alert.""" """Test that device re-arrival within 5 min of departure suppresses ARRIVED alert."""
cleanup_flap_state()
net_alerter.known_devices.clear() net_alerter.known_devices.clear()
net_alerter.departure_timers.clear() net_alerter.departure_timers.clear()
net_alerter.last_departed_time.clear() net_alerter.last_departed_time.clear()
@@ -192,6 +208,7 @@ def test_re_arrival_within_5min_suppresses_alert():
def test_dhcp_renewal_dedup_30min(): def test_dhcp_renewal_dedup_30min():
"""Test that DHCP renewal from known device (< 30 min old) doesn't send ARRIVED alert.""" """Test that DHCP renewal from known device (< 30 min old) doesn't send ARRIVED alert."""
cleanup_flap_state()
net_alerter.known_devices.clear() net_alerter.known_devices.clear()
alert_calls = [] alert_calls = []
@@ -223,10 +240,12 @@ def test_dhcp_renewal_dedup_30min():
def test_re_arrival_cancels_departure_timer(): def test_re_arrival_cancels_departure_timer():
""" """
cleanup_flap_state()
Regression test: device departs → timer starts → device re-arrives BEFORE timer fires. Regression test: device departs → timer starts → device re-arrives BEFORE timer fires.
Timer should be cancelled and NO departure alert should be sent. Timer should be cancelled and NO departure alert should be sent.
This tests the exact bug scenario: on_arrival() must cancel any pending timer. This tests the exact bug scenario: on_arrival() must cancel any pending timer.
""" """
cleanup_flap_state()
net_alerter.known_devices.clear() net_alerter.known_devices.clear()
net_alerter.departure_timers.clear() net_alerter.departure_timers.clear()
net_alerter.last_departed_time.clear() net_alerter.last_departed_time.clear()
@@ -295,6 +314,7 @@ def test_rapid_flap_no_duplicate_arrived_alerts():
- on_arrival() checks for departing flag and silently cancels timer - on_arrival() checks for departing flag and silently cancels timer
- No duplicate ARRIVED alerts - No duplicate ARRIVED alerts
""" """
cleanup_flap_state()
net_alerter.known_devices.clear() net_alerter.known_devices.clear()
net_alerter.departure_timers.clear() net_alerter.departure_timers.clear()
net_alerter.last_departed_time.clear() net_alerter.last_departed_time.clear()