Тесты: дедупликация алертов, возврат в норму, сторож, сводки (13 тестов без сети)

tests/conftest.py

@@ -0,0 +1,62 @@
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+os.environ.update({
+    "BOT_TOKEN": "123456:TEST-token-for-unit-tests-only",
+    "ALLOWED_IDS": "1,2",
+    "AM_ALERTS_URL": "https://example.test/ambot/api/v2/alerts",
+    "PROM_QUERY_URL": "https://example.test/prombot/api/v1/query",
+    "AM_TOKEN": "am-token",
+    "PROM_TOKEN": "prom-token",
+    "SERVERS_ORDER": "spb,ams,dorm",
+    "ALERTBOT_DB": str(Path(tempfile.mkdtemp()) / "test.db"),
+    "ALERT_WATCHDOG_FAILURES": "3",
+})
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import pytest  # noqa: E402
+
+import handlers  # noqa: E402
+
+
+@pytest.fixture(autouse=True)
+def fresh_db():
+    with handlers.db._conn() as conn:
+        conn.execute("DELETE FROM users")
+        conn.execute("DELETE FROM alert_seen")
+    handlers.db.add_user(1, "owner")
+    handlers.db.add_user(2, "friend")
+    handlers.db.set_alerts(2, False)
+
+
+@pytest.fixture
+def sent(monkeypatch):
+    """Messages the bot would send: (user_id, text)."""
+    out = []
+
+    async def fake_send(uid, text):
+        out.append((uid, text))
+
+    monkeypatch.setattr(handlers.bot, "send_message", fake_send)
+    return out
+
+
+@pytest.fixture
+def prom(monkeypatch):
+    """Fake Prometheus: maps a substring of the query to a result vector."""
+    answers = {}
+
+    async def fake_promq(session, query):
+        for key, value in answers.items():
+            if key in query:
+                return value
+        return []
+
+    monkeypatch.setattr(handlers, "_promq", fake_promq)
+    return answers
+
+
+def vec(**by_server):
+    return [{"metric": {"server": k}, "value": [0, str(v)]} for k, v in by_server.items()]

tests/test_alerts.py

@@ -0,0 +1,61 @@
+import asyncio
+
+import handlers
+
+
+def alert(fp, name="NodeDown", state="active", server="spb", summary="Узел не отвечает"):
+    return {"fingerprint": fp, "status": {"state": state},
+            "labels": {"alertname": name, "severity": "critical", "server": server},
+            "annotations": {"summary": summary}}
+
+
+def run(coro):
+    return asyncio.run(coro)
+
+
+def test_new_alert_is_sent_once_to_subscribers(sent):
+    run(handlers.process_alerts([alert("a1")]))
+    run(handlers.process_alerts([alert("a1")]))
+    assert [uid for uid, _ in sent] == [1]
+    assert "Проблема · NodeDown" in sent[0][1]
+
+
+def test_resolved_alert_is_announced(sent):
+    run(handlers.process_alerts([alert("a1")]))
+    run(handlers.process_alerts([]))
+    assert "В норме · NodeDown" in sent[-1][1]
+    assert handlers.db.list_firing_fingerprints() == []
+
+
+def test_suppressed_alerts_are_ignored(sent):
+    run(handlers.process_alerts([alert("a1", state="suppressed"), {"status": {"state": "active"}}]))
+    assert sent == []
+
+
+def test_alert_text_is_html_escaped():
+    text = handlers._fmt_firing(alert("a1", name="<b>x</b>", summary="a < b & c"))
+    assert "&lt;b&gt;x&lt;/b&gt;" in text
+    assert "a &lt; b &amp; c" in text
+
+
+def test_watchdog_reports_once_and_recovers(sent):
+    wd = handlers.Watchdog(threshold=3)
+    for _ in range(5):
+        run(wd.failed("timeout"))
+    assert len(sent) == 1 and "Мониторинг недоступен" in sent[0][1]
+    run(wd.ok())
+    assert "Мониторинг снова доступен" in sent[-1][1]
+    run(wd.ok())
+    assert len(sent) == 2
+
+
+def test_watchdog_ignores_short_glitches(sent):
+    wd = handlers.Watchdog(threshold=3)
+    run(wd.failed("timeout"))
+    run(wd.failed("timeout"))
+    run(wd.ok())
+    assert sent == []
+
+
+def test_token_is_sent_in_header_not_url():
+    assert handlers._auth("secret") == {"Authorization": "Bearer secret"}

tests/test_reports.py

@@ -0,0 +1,69 @@
+import asyncio
+
+import handlers
+from tests.conftest import vec
+
+
+def run(coro):
+    return asyncio.run(coro)
+
+
+def test_uptime_format():
+    assert handlers.fmt_uptime(3 * 86400 + 5 * 3600) == "3д 5ч"
+    assert handlers.fmt_uptime(7200) == "2ч"
+    assert handlers.fmt_uptime(-5) == "0ч"
+
+
+def test_servers_report_marks_problems(prom):
+    import time
+
+    now = time.time()
+    prom.update({
+        "up{": vec(spb=1, ams=0),
+        "node_cpu_seconds_total": vec(spb=12, ams=3),
+        "MemAvailable": vec(spb=95, ams=40),
+        "node_filesystem_avail_bytes": vec(spb=50, ams=91),
+        "node_load1": vec(spb=0.5, ams=0.1),
+        "node_boot_time_seconds": vec(spb=now - 2 * 86400, ams=now - 3600),
+        "SwapTotal": vec(spb=0, ams=0),
+        "hwmon": vec(spb=40),
+    })
+    text = run(handlers.render_servers())
+    assert "<b>SPB</b> · ⚠ память" in text
+    assert "<b>AMS</b> · НЕ ОТВЕЧАЕТ" in text
+    assert "<b>DORM</b> — нет данных" in text
+    assert "аптайм 2д 0ч" in text
+
+
+def test_services_report(prom):
+    prom["up"] = [
+        {"metric": {"job": "node_spb", "instance": "spb:9100"}, "value": [0, "1"]},
+        {"metric": {"job": "blackbox_http", "instance": "https://<bad>.example"}, "value": [0, "0"]},
+    ]
+    text = run(handlers.render_services())
+    assert "1/2 в норме" in text
+    assert "&lt;bad&gt;" in text
+
+
+def test_services_all_ok(prom):
+    prom["up"] = [{"metric": {"job": "node_spb"}, "value": [0, "1"]}]
+    assert "Все 1 в норме" in run(handlers.render_services())
+
+
+def test_certs_report_sorted_with_warning(prom):
+    prom["probe_ssl_earliest_cert_expiry"] = [
+        {"metric": {"instance": "https://deev.space/"}, "value": [0, "60.2"]},
+        {"metric": {"instance": "https://tablo.deev.su"}, "value": [0, "5.1"]},
+    ]
+    text = run(handlers.render_certs())
+    assert "скоро истекают: tablo.deev.su — 5 дн" in text
+    rows = text.split("<pre>")[1].split("</pre>")[0].splitlines()
+    assert rows[0].endswith("tablo.deev.su") and rows[1].endswith("deev.space")
+
+
+def test_prometheus_error_is_reported(monkeypatch):
+    async def boom(session, query):
+        raise RuntimeError("connection refused")
+
+    monkeypatch.setattr(handlers, "_promq", boom)
+    assert "Не удалось получить метрики" in run(handlers.render_servers())