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