import sys
import os
import time
from unittest.mock import patch
import pytest
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
SRC_ROOT = os.path.join(PROJECT_ROOT, 'src')
if SRC_ROOT not in sys.path:
sys.path.insert(0, SRC_ROOT)
from virtcca_deploy.monitor.core.alert_suppressor import AlertSuppressor, VALID_SOURCE_TYPES
class TestAlertSuppressorInit:
def test_default_parameters(self):
suppressor = AlertSuppressor()
assert suppressor.repeat_interval == 60
assert suppressor.flap_window == 600
assert suppressor.flap_threshold == 3
assert suppressor.short_lived_threshold == 30
def test_custom_parameters(self):
suppressor = AlertSuppressor(
repeat_interval=120,
flap_window=300,
flap_threshold=5,
short_lived_threshold=60
)
assert suppressor.repeat_interval == 120
assert suppressor.flap_window == 300
assert suppressor.flap_threshold == 5
assert suppressor.short_lived_threshold == 60
def test_state_initialization(self):
suppressor = AlertSuppressor()
state = suppressor.state["test_key"]
assert state["last_sent"] == 0
assert state["current_state"] == "OK"
assert state["last_change"] == 0
assert len(state["history"]) == 0
class TestAlertSuppressorNormalFlow:
def test_first_alert_should_send(self):
suppressor = AlertSuppressor()
result = suppressor.should_send("vm-001", "ALERT")
assert result is True
def test_first_recovery_should_be_suppressed(self):
suppressor = AlertSuppressor()
suppressor.should_send("vm-001", "ALERT")
result = suppressor.should_send("vm-001", "OK")
assert result is False
def test_state_transition_ok_to_ALERT(self):
suppressor = AlertSuppressor()
suppressor.should_send("vm-001", "ALERT")
state = suppressor.state["vm-001"]
assert state["current_state"] == "ALERT"
def test_state_transition_ALERT_to_OK(self):
suppressor = AlertSuppressor()
suppressor.should_send("vm-001", "ALERT")
suppressor.should_send("vm-001", "OK")
state = suppressor.state["vm-001"]
assert state["current_state"] == "OK"
def test_recovery_after_threshold_should_send(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 31
result = suppressor.should_send("vm-001", "OK")
assert result is True
def test_multiple_keys_independent(self):
suppressor = AlertSuppressor()
suppressor.should_send("vm-001", "ALERT")
result = suppressor.should_send("vm-002", "ALERT")
assert result is True
def test_history_records_transitions(self):
suppressor = AlertSuppressor()
suppressor.should_send("vm-001", "ALERT")
suppressor.should_send("vm-001", "OK")
suppressor.should_send("vm-001", "ALERT")
state = suppressor.state["vm-001"]
assert len(state["history"]) == 3
class TestShortLivedSuppression:
def test_short_lived_recovery_suppressed(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
suppressor.should_send("vm-001", "ALERT")
result = suppressor.should_send("vm-001", "OK")
assert result is False
def test_long_lived_recovery_allowed(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
suppressor.should_send("vm-001", "ALERT")
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time', return_value=time.time() + 31):
result = suppressor.should_send("vm-001", "OK")
assert result is True
def test_boundary_exactly_threshold(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
suppressor.should_send("vm-001", "ALERT")
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time', return_value=time.time() + 30):
result = suppressor.should_send("vm-001", "OK")
assert result is True
def test_boundary_one_second_below_threshold(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
suppressor.should_send("vm-001", "ALERT")
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time', return_value=time.time() + 29):
result = suppressor.should_send("vm-001", "OK")
assert result is False
def test_short_lived_does_not_block_state_update(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
suppressor.should_send("vm-001", "ALERT")
suppressor.should_send("vm-001", "OK")
state = suppressor.state["vm-001"]
assert state["current_state"] == "OK"
def test_short_lived_updates_history(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
suppressor.should_send("vm-001", "ALERT")
suppressor.should_send("vm-001", "OK")
state = suppressor.state["vm-001"]
assert len(state["history"]) == 2
assert state["history"][-1][1] == "OK"
def test_short_lived_recovery_does_not_update_last_sent(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 10
result = suppressor.should_send("vm-001", "OK")
assert result is False
state = suppressor.state["vm-001"]
assert state["last_sent"] == base_time
def test_long_lived_recovery_updates_last_sent(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 31
result = suppressor.should_send("vm-001", "OK")
assert result is True
state = suppressor.state["vm-001"]
assert state["last_sent"] == base_time + 31
class TestRepeatSuppression:
def test_repeat_alert_within_interval_suppressed(self):
suppressor = AlertSuppressor(repeat_interval=60)
suppressor.should_send("vm-001", "ALERT")
result = suppressor.should_send("vm-001", "ALERT")
assert result is False
def test_repeat_alert_after_interval_allowed(self):
suppressor = AlertSuppressor(repeat_interval=60)
suppressor.should_send("vm-001", "ALERT")
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time', return_value=time.time() + 61):
result = suppressor.should_send("vm-001", "ALERT")
assert result is True
def test_repeat_boundary_exactly_interval(self):
suppressor = AlertSuppressor(repeat_interval=60)
suppressor.should_send("vm-001", "ALERT")
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time', return_value=time.time() + 60):
result = suppressor.should_send("vm-001", "ALERT")
assert result is False
def test_repeat_boundary_one_second_after(self):
suppressor = AlertSuppressor(repeat_interval=60)
suppressor.should_send("vm-001", "ALERT")
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time', return_value=time.time() + 61):
result = suppressor.should_send("vm-001", "ALERT")
assert result is True
def test_last_sent_updated_on_send(self):
suppressor = AlertSuppressor(repeat_interval=60)
initial_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time', return_value=initial_time):
suppressor.should_send("vm-001", "ALERT")
state = suppressor.state["vm-001"]
assert state["last_sent"] == initial_time
class TestFlapDetection:
def test_flap_detection_suppresses_alert(self):
suppressor = AlertSuppressor(flap_window=600, flap_threshold=3)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 10
suppressor.should_send("vm-001", "OK")
mock_time.return_value = base_time + 20
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 30
result = suppressor.should_send("vm-001", "ALERT")
assert result is False
def test_flap_detection_allows_recovery(self):
suppressor = AlertSuppressor(flap_window=600, flap_threshold=3)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 10
suppressor.should_send("vm-001", "OK")
mock_time.return_value = base_time + 20
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 30
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 40
result = suppressor.should_send("vm-001", "OK")
assert result is True
def test_flap_window_expiration(self):
suppressor = AlertSuppressor(flap_window=60, flap_threshold=3)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 10
suppressor.should_send("vm-001", "OK")
mock_time.return_value = base_time + 20
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 120
result = suppressor.should_send("vm-001", "ALERT")
assert result is True
def test_flap_threshold_boundary(self):
suppressor = AlertSuppressor(flap_window=600, flap_threshold=3)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 10
suppressor.should_send("vm-001", "OK")
mock_time.return_value = base_time + 20
result = suppressor.should_send("vm-001", "ALERT")
assert result is True
def test_flap_threshold_exactly_met(self):
suppressor = AlertSuppressor(flap_window=600, flap_threshold=3)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 10
suppressor.should_send("vm-001", "OK")
mock_time.return_value = base_time + 20
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 30
result = suppressor.should_send("vm-001", "ALERT")
assert result is False
class TestInvalidInput:
def test_invalid_state_returns_false(self):
suppressor = AlertSuppressor()
result = suppressor.should_send("vm-001", "INVALID")
assert result is False
def test_empty_string_state_returns_false(self):
suppressor = AlertSuppressor()
result = suppressor.should_send("vm-001", "")
assert result is False
def test_none_state_returns_false(self):
suppressor = AlertSuppressor()
result = suppressor.should_send("vm-001", None)
assert result is False
def test_lowercase_state_returns_false(self):
suppressor = AlertSuppressor()
result = suppressor.should_send("vm-001", "alert")
assert result is False
def test_valid_source_types_constant(self):
assert VALID_SOURCE_TYPES == ("OK", "ALERT")
class TestEdgeCases:
def test_rapid_OK_ALERT_cycles(self):
suppressor = AlertSuppressor(
repeat_interval=5,
flap_window=60,
flap_threshold=5,
short_lived_threshold=10
)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
for i in range(10):
mock_time.return_value = base_time + i * 2
state = "ALERT" if i % 2 == 0 else "OK"
suppressor.should_send("vm-001", state)
state = suppressor.state["vm-001"]
assert len(state["history"]) == 10
def test_history_max_length(self):
suppressor = AlertSuppressor()
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
for i in range(30):
mock_time.return_value = base_time + i
state = "ALERT" if i % 2 == 0 else "OK"
suppressor.should_send("vm-001", state)
state = suppressor.state["vm-001"]
assert len(state["history"]) == 20
def test_same_state_repeated_no_history_update(self):
suppressor = AlertSuppressor()
suppressor.should_send("vm-001", "ALERT")
initial_history_len = len(suppressor.state["vm-001"]["history"])
suppressor.should_send("vm-001", "ALERT")
assert len(suppressor.state["vm-001"]["history"]) == initial_history_len
def test_state_drift_prevention(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
suppressor.should_send("vm-001", "ALERT")
suppressor.should_send("vm-001", "OK")
state = suppressor.state["vm-001"]
assert state["current_state"] == "OK"
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time', return_value=time.time() + 10):
result = suppressor.should_send("vm-001", "ALERT")
assert result is False
assert state["current_state"] == "ALERT"
def test_consecutive_OK_states(self):
suppressor = AlertSuppressor()
suppressor.should_send("vm-001", "ALERT")
suppressor.should_send("vm-001", "OK")
result = suppressor.should_send("vm-001", "OK")
assert result is True
def test_ALERT_after_long_OK(self):
suppressor = AlertSuppressor(repeat_interval=60)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 100
suppressor.should_send("vm-001", "OK")
mock_time.return_value = base_time + 200
result = suppressor.should_send("vm-001", "ALERT")
assert result is True
class TestComplexScenarios:
def test_realistic_flapping_scenario(self):
suppressor = AlertSuppressor(
repeat_interval=60,
flap_window=300,
flap_threshold=3,
short_lived_threshold=30
)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
assert suppressor.should_send("vm-001", "ALERT") is True
mock_time.return_value = base_time + 10
assert suppressor.should_send("vm-001", "OK") is False
mock_time.return_value = base_time + 50
assert suppressor.should_send("vm-001", "ALERT") is False
mock_time.return_value = base_time + 60
assert suppressor.should_send("vm-001", "OK") is False
mock_time.return_value = base_time + 100
assert suppressor.should_send("vm-001", "ALERT") is False
mock_time.return_value = base_time + 130
assert suppressor.should_send("vm-001", "OK") is True
def test_recovery_notification_always_allowed(self):
suppressor = AlertSuppressor(flap_window=600, flap_threshold=3)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
for i in range(5):
mock_time.return_value = base_time + i * 10
suppressor.should_send("vm-001", "ALERT")
suppressor.should_send("vm-001", "OK")
mock_time.return_value = base_time + 60
result = suppressor.should_send("vm-001", "OK")
assert result is True
def test_multiple_vms_independent_suppression(self):
suppressor = AlertSuppressor(repeat_interval=60)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time', return_value=base_time):
suppressor.should_send("vm-001", "ALERT")
assert suppressor.should_send("vm-001", "ALERT") is False
assert suppressor.should_send("vm-002", "ALERT") is True
assert suppressor.should_send("vm-003", "ALERT") is True
def test_state_machine_accuracy(self):
suppressor = AlertSuppressor()
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
assert suppressor.state["vm-001"]["current_state"] == "ALERT"
mock_time.return_value = base_time + 100
suppressor.should_send("vm-001", "OK")
assert suppressor.state["vm-001"]["current_state"] == "OK"
mock_time.return_value = base_time + 200
suppressor.should_send("vm-001", "ALERT")
assert suppressor.state["vm-001"]["current_state"] == "ALERT"
class TestNotificationPolicyVsStateMachine:
def test_suppressed_notification_still_updates_state(self):
suppressor = AlertSuppressor(short_lived_threshold=30)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 10
result = suppressor.should_send("vm-001", "OK")
assert result is False
assert suppressor.state["vm-001"]["current_state"] == "OK"
def test_flap_suppressed_ALERT_still_updates_state(self):
suppressor = AlertSuppressor(flap_window=600, flap_threshold=3)
base_time = time.time()
with patch('virtcca_deploy.monitor.core.alert_suppressor.time.time') as mock_time:
mock_time.return_value = base_time
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 10
suppressor.should_send("vm-001", "OK")
mock_time.return_value = base_time + 20
suppressor.should_send("vm-001", "ALERT")
mock_time.return_value = base_time + 30
result = suppressor.should_send("vm-001", "ALERT")
assert result is False
assert suppressor.state["vm-001"]["current_state"] == "ALERT"
def test_repeat_suppressed_ALERT_state_unchanged(self):
suppressor = AlertSuppressor(repeat_interval=60)
suppressor.should_send("vm-001", "ALERT")
result = suppressor.should_send("vm-001", "ALERT")
assert result is False
assert suppressor.state["vm-001"]["current_state"] == "ALERT"