"""TC-01…TC-04: the pure decision function (alert/throttle/realert/recovery). Mirrors the disk_watchdog.decide_action tests — the generalised ``decide`` is a strict superset (boolean ``signal_active`` instead of ``used_pct >= threshold``). """ from watchdog.decision import ( ACTION_ALERT, ACTION_NONE, ACTION_REALERT, ACTION_RECOVERY, AlertState, decide, ) COOLDOWN = 1800.0 def test_tc01_not_alerting_active_alerts(): # TC-01: not-alerting & signal active -> ALERT (one per crossing). prev = AlertState(alerting=False) assert decide(True, prev, now=100.0, cooldown_s=COOLDOWN) == ACTION_ALERT def test_tc01_not_alerting_inactive_is_none(): prev = AlertState(alerting=False) assert decide(False, prev, now=100.0, cooldown_s=COOLDOWN) == ACTION_NONE def test_tc02_alerting_active_in_cooldown_is_none(): # TC-02: alerting & still active & cooldown NOT elapsed -> NONE (anti-spam). prev = AlertState(alerting=True, last_alert_at=1000.0) assert decide(True, prev, now=1000.0 + 10.0, cooldown_s=COOLDOWN) == ACTION_NONE def test_tc03_alerting_active_cooldown_elapsed_realerts(): # TC-03: alerting & still active & cooldown elapsed -> REALERT. prev = AlertState(alerting=True, last_alert_at=1000.0) assert decide(True, prev, now=1000.0 + COOLDOWN, cooldown_s=COOLDOWN) == ACTION_REALERT def test_tc03_alerting_active_no_last_alert_realerts(): # Defensive: alerting but last_alert_at missing -> treat cooldown as elapsed. prev = AlertState(alerting=True, last_alert_at=None) assert decide(True, prev, now=5.0, cooldown_s=COOLDOWN) == ACTION_REALERT def test_tc04_alerting_recovers_when_inactive(): # TC-04: alerting & signal back to normal -> RECOVERY. prev = AlertState(alerting=True, last_alert_at=1000.0) assert decide(False, prev, now=1200.0, cooldown_s=COOLDOWN) == ACTION_RECOVERY def test_cooldown_boundary_is_inclusive(): # Exactly at cooldown boundary -> REALERT (>= semantics, like disk_watchdog). prev = AlertState(alerting=True, last_alert_at=0.0) assert decide(True, prev, now=COOLDOWN, cooldown_s=COOLDOWN) == ACTION_REALERT