141 lines
6.2 KiB
Python
141 lines
6.2 KiB
Python
"""Test del giro ORARIO di monitor_health (#30).
|
|
|
|
I sorveglianti a max_age<=6h (cuscino_watch, venue_watch, collect_chain) girano alle 00:30 col
|
|
report giornaliero e basta: un guasto si scopre fino a 24h dopo. Qui si guarda ogni ora, ma si
|
|
notifica solo cio' che e' NUOVO rispetto al marcatore "gia' detto" — altrimenti un monitor fermo
|
|
manderebbe lo stesso Telegram ogni ora per giorni.
|
|
|
|
Le funzioni PURE (filtro per max_age, chiave dell'alert, chi e' nuovo, il marcatore aggiornato)
|
|
stanno in src/live/monitor_health.py e si testano su righe finte, senza toccare disco o rete.
|
|
L'orchestrazione (leggere/scrivere il marcatore, chiamare notify) sta nello script e si testa qui
|
|
sotto con monkeypatch — NIENTE Telegram vero."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
sys.path.insert(0, str(ROOT / "scripts" / "live"))
|
|
|
|
from src.live.monitor_health import ( # noqa: E402
|
|
MonitorSpec,
|
|
alert_nuovi,
|
|
chiave_alert,
|
|
marcatore_aggiornato,
|
|
monitors_orari,
|
|
)
|
|
|
|
import monitor_health as MH # noqa: E402 (scripts/live/monitor_health.py)
|
|
|
|
# righe finte, nella forma che assess()/check_all() restituiscono davvero
|
|
ROW_FERMO = dict(name="venue_watch", gate="tripwire di venue orario", n_bars=3, age_h=4.0,
|
|
coverage=0.9, expected=3, status="FERMO", why="ultima barra 4h fa (limite 3h)")
|
|
ROW_OK = dict(name="venue_watch", gate="tripwire di venue orario", n_bars=10, age_h=0.2,
|
|
coverage=1.0, expected=10, status="OK", why="")
|
|
ROW_NUOVO = dict(name="venue_watch", gate=None, n_bars=1, age_h=0.1, coverage=None,
|
|
expected=None, status="NUOVO", why="1 barra: non c'e' abbastanza serie")
|
|
|
|
|
|
# ===========================================================================
|
|
# funzioni PURE — filtro per max_age (src/live/monitor_health.py)
|
|
# ===========================================================================
|
|
def test_il_filtro_deriva_dalla_tabella_non_da_una_lista_di_nomi():
|
|
specs = (
|
|
MonitorSpec("a", "a", "runs.jsonl", 1.0, max_age_h=3.0),
|
|
MonitorSpec("b", "b", "runs.jsonl", 1.0, max_age_h=6.0), # al bordo: dentro
|
|
MonitorSpec("c", "c", "runs.jsonl", 24.0, max_age_h=30.0),
|
|
)
|
|
assert {m.name for m in monitors_orari(specs)} == {"a", "b"}
|
|
|
|
|
|
def test_i_tre_orari_di_oggi_sono_quelli_noti():
|
|
"""Se la tabella cambia questo test lo dice: e' l'ancora fra il registro e il flag."""
|
|
assert {m.name for m in monitors_orari()} == {"collect_chain", "cuscino_watch", "venue_watch"}
|
|
|
|
|
|
# ===========================================================================
|
|
# funzioni PURE — chiave, nuovi, marcatore
|
|
# ===========================================================================
|
|
def test_chiave_e_nome_piu_stato_non_il_testo():
|
|
"""Il testo (`why`) contiene l'eta', che cambia ogni ora anche se il guasto e' lo stesso."""
|
|
assert chiave_alert(ROW_FERMO) == "venue_watch:FERMO"
|
|
piu_vecchio = {**ROW_FERMO, "age_h": 9.0, "why": "ultima barra 9h fa (limite 3h)"}
|
|
assert chiave_alert(piu_vecchio) == chiave_alert(ROW_FERMO)
|
|
|
|
|
|
def test_alert_nuovo_se_la_chiave_non_e_nel_marcatore():
|
|
assert alert_nuovi([ROW_FERMO], set()) == [ROW_FERMO]
|
|
|
|
|
|
def test_lo_stesso_alert_al_secondo_giro_non_e_piu_nuovo():
|
|
assert alert_nuovi([ROW_FERMO], {"venue_watch:FERMO"}) == []
|
|
|
|
|
|
def test_nuovo_non_e_un_alert_da_notificare():
|
|
assert alert_nuovi([ROW_NUOVO], set()) == []
|
|
|
|
|
|
def test_il_marcatore_aggiunge_la_chiave_di_un_alert():
|
|
assert marcatore_aggiornato([ROW_FERMO], set()) == {"venue_watch:FERMO"}
|
|
|
|
|
|
def test_il_ritorno_a_ok_toglie_la_chiave_dal_marcatore():
|
|
"""Cosi' un nuovo guasto sullo stesso monitor riparte con un nuovo avviso."""
|
|
assert marcatore_aggiornato([ROW_OK], {"venue_watch:FERMO"}) == set()
|
|
|
|
|
|
# ===========================================================================
|
|
# orchestrazione — scripts/live/monitor_health.py main_orari(), NIENTE Telegram
|
|
# ===========================================================================
|
|
def test_notify_fallito_non_scrive_il_marcatore(tmp_path, monkeypatch):
|
|
"""Stessa regola di fee_watch #29: niente 'gia' detto' su un invio fallito, altrimenti il
|
|
prossimo giro non ritenterebbe."""
|
|
detto = tmp_path / "detto.json"
|
|
monkeypatch.setattr(MH, "DETTO", detto)
|
|
monkeypatch.setattr(MH, "check_all", lambda root, monitors=None: [ROW_FERMO])
|
|
chiamate = []
|
|
monkeypatch.setattr(MH, "notify", lambda *a, **k: (chiamate.append((a, k)), False)[1])
|
|
monkeypatch.setattr(sys, "argv", ["monitor_health.py", "--quiet", "--solo-orari"])
|
|
|
|
rc = MH.main_orari()
|
|
|
|
assert rc == 2 # c'e' un alert
|
|
assert len(chiamate) == 1 # e si e' provato a notificarlo
|
|
assert not detto.exists() # ma l'invio e' fallito: nessun marcatore
|
|
|
|
|
|
def test_notify_riuscito_scrive_il_marcatore_e_non_rinotifica(tmp_path, monkeypatch):
|
|
detto = tmp_path / "detto.json"
|
|
monkeypatch.setattr(MH, "DETTO", detto)
|
|
monkeypatch.setattr(MH, "check_all", lambda root, monitors=None: [ROW_FERMO])
|
|
monkeypatch.setattr(sys, "argv", ["monitor_health.py", "--quiet", "--solo-orari"])
|
|
|
|
monkeypatch.setattr(MH, "notify", lambda *a, **k: True)
|
|
rc1 = MH.main_orari()
|
|
assert rc1 == 2
|
|
assert json.loads(detto.read_text())["detti"] == ["venue_watch:FERMO"]
|
|
|
|
chiamate = []
|
|
monkeypatch.setattr(MH, "notify", lambda *a, **k: (chiamate.append(1), True)[1])
|
|
rc2 = MH.main_orari()
|
|
assert rc2 == 2 # l'alert c'e' ancora (anche se gia' detto: exit 2 comunque)
|
|
assert chiamate == [] # ma non si rimanda lo stesso Telegram
|
|
|
|
|
|
def test_tornato_ok_il_giro_dopo_pulisce_il_marcatore(tmp_path, monkeypatch):
|
|
detto = tmp_path / "detto.json"
|
|
detto.write_text(json.dumps({"detti": ["venue_watch:FERMO"], "ts": "2026-09-27T00:00:00Z"}))
|
|
monkeypatch.setattr(MH, "DETTO", detto)
|
|
monkeypatch.setattr(MH, "check_all", lambda root, monitors=None: [ROW_OK])
|
|
monkeypatch.setattr(sys, "argv", ["monitor_health.py", "--quiet", "--solo-orari"])
|
|
chiamate = []
|
|
monkeypatch.setattr(MH, "notify", lambda *a, **k: (chiamate.append(1), True)[1])
|
|
|
|
rc = MH.main_orari()
|
|
|
|
assert rc == 0
|
|
assert chiamate == [] # OK non notifica mai
|
|
assert json.loads(detto.read_text())["detti"] == []
|