155 lines
4.9 KiB
Python
155 lines
4.9 KiB
Python
"""
|
|
Tests for brain_watchdog.main.
|
|
|
|
Module-level env vars are required at import time; set them before the first
|
|
import of the module so tests can run without a real control-plane.
|
|
"""
|
|
import importlib.util
|
|
import os
|
|
import time
|
|
from unittest.mock import patch
|
|
|
|
os.environ.setdefault("CONTROL_PLANE_URL", "http://test-cp:8080")
|
|
os.environ.setdefault("TG_TOKEN", "test_token")
|
|
os.environ.setdefault("TG_CHAT_ID", "12345")
|
|
|
|
import brain_watchdog.main as bwm
|
|
|
|
|
|
def test_package_importable():
|
|
spec = importlib.util.find_spec("brain_watchdog")
|
|
assert spec is not None
|
|
|
|
|
|
def test_check_ok_fresh():
|
|
now = time.time()
|
|
with patch.object(bwm, "http_get", return_value=(200, {"last_update": now - 10})):
|
|
ok, reason = bwm.check()
|
|
assert ok
|
|
assert "ok" in reason
|
|
|
|
|
|
def test_check_fail_stale():
|
|
now = time.time()
|
|
stale_ts = now - (bwm.STALE_THRESHOLD + 120)
|
|
with patch.object(bwm, "http_get", return_value=(200, {"last_update": stale_ts})):
|
|
ok, reason = bwm.check()
|
|
assert not ok
|
|
assert "stale" in reason
|
|
|
|
|
|
def test_check_fail_unreachable():
|
|
with patch.object(bwm, "http_get", return_value=(None, None)):
|
|
ok, reason = bwm.check()
|
|
assert not ok
|
|
assert "unreachable" in reason
|
|
|
|
|
|
def test_check_fail_http_error():
|
|
with patch.object(bwm, "http_get", return_value=(503, None)):
|
|
ok, reason = bwm.check()
|
|
assert not ok
|
|
assert "503" in reason
|
|
|
|
|
|
def test_check_fail_missing_last_update():
|
|
with patch.object(bwm, "http_get", return_value=(200, {"other": "data"})):
|
|
ok, reason = bwm.check()
|
|
assert not ok
|
|
assert "last_update" in reason
|
|
|
|
|
|
def test_check_fail_unparseable_timestamp():
|
|
with patch.object(bwm, "http_get", return_value=(200, {"last_update": "not-a-number"})):
|
|
ok, reason = bwm.check()
|
|
assert not ok
|
|
assert "parseable" in reason
|
|
|
|
|
|
# ---------- Prometheus alert polling ----------
|
|
|
|
SAMPLE_PROM_RESPONSE = {
|
|
"status": "success",
|
|
"data": {
|
|
"alerts": [
|
|
{
|
|
"labels": {"alertname": "NodeDown", "node": "piha"},
|
|
"annotations": {"summary": "Node is down", "description": ""},
|
|
"state": "firing",
|
|
"activeAt": "2026-06-30T10:00:00Z",
|
|
"value": "0e+00",
|
|
},
|
|
{
|
|
"labels": {"alertname": "HighLoad", "node": "solaria"},
|
|
"annotations": {"summary": "High CPU", "description": "CPU above 90%"},
|
|
"state": "pending",
|
|
"activeAt": "2026-06-30T10:01:00Z",
|
|
"value": "0e+00",
|
|
},
|
|
]
|
|
},
|
|
}
|
|
|
|
|
|
def test_check_prometheus_alerts_disabled(monkeypatch):
|
|
monkeypatch.setattr(bwm, "PROMETHEUS_URL", "")
|
|
assert bwm.check_prometheus_alerts() == []
|
|
|
|
|
|
def test_check_prometheus_alerts_parses_firing(monkeypatch):
|
|
monkeypatch.setattr(bwm, "PROMETHEUS_URL", "http://prom:9090")
|
|
with patch.object(bwm, "http_get", return_value=(200, SAMPLE_PROM_RESPONSE)):
|
|
result = bwm.check_prometheus_alerts()
|
|
assert len(result) == 1
|
|
assert result[0]["alertname"] == "NodeDown"
|
|
assert result[0]["node"] == "piha"
|
|
assert result[0]["key"] == "NodeDown:piha"
|
|
assert result[0]["summary"] == "Node is down"
|
|
|
|
|
|
def test_check_prometheus_alerts_unreachable(monkeypatch):
|
|
monkeypatch.setattr(bwm, "PROMETHEUS_URL", "http://prom:9090")
|
|
with patch.object(bwm, "http_get", return_value=(None, None)):
|
|
assert bwm.check_prometheus_alerts() == []
|
|
|
|
|
|
def test_prometheus_debounce_no_duplicate(monkeypatch):
|
|
monkeypatch.setattr(bwm, "PROMETHEUS_URL", "http://prom:9090")
|
|
firing = [
|
|
{
|
|
"key": "NodeDown:piha",
|
|
"alertname": "NodeDown",
|
|
"node": "piha",
|
|
"summary": "down",
|
|
"description": "",
|
|
}
|
|
]
|
|
with patch.object(bwm, "check_prometheus_alerts", return_value=firing):
|
|
with patch.object(bwm, "send_telegram", return_value=True) as mock_tg:
|
|
state: dict = {}
|
|
bwm.handle_prometheus_alerts(state)
|
|
assert mock_tg.call_count == 1
|
|
assert "NodeDown:piha" in state["prom_alerted"]
|
|
|
|
# Same alert still firing — debounce must suppress second send
|
|
bwm.handle_prometheus_alerts(state)
|
|
assert mock_tg.call_count == 1
|
|
|
|
|
|
def test_prometheus_debounce_recovery(monkeypatch):
|
|
monkeypatch.setattr(bwm, "PROMETHEUS_URL", "http://prom:9090")
|
|
state = {
|
|
"prom_alerted": {
|
|
"NodeDown:piha": {"alertname": "NodeDown", "node": "piha"}
|
|
}
|
|
}
|
|
with patch.object(bwm, "check_prometheus_alerts", return_value=[]):
|
|
with patch.object(bwm, "send_telegram", return_value=True) as mock_tg:
|
|
bwm.handle_prometheus_alerts(state)
|
|
|
|
assert mock_tg.call_count == 1
|
|
call_text = mock_tg.call_args[0][0]
|
|
assert "✅" in call_text
|
|
assert "NodeDown" in call_text
|
|
assert "NodeDown:piha" not in state["prom_alerted"]
|