homelab-codex-ws/services/brain-watchdog/tests/test_main.py

"""
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"]