""" Contrôles de santé des services de fond. Règle cardinale : une sonde ne doit JAMAIS interrompre le service qu'elle surveille. Un contrôle de santé qui fait tomber le crawler serait pire que pas de contrôle du tout — d'où le nombre de cas d'échec vérifiés ici. """ import contextlib import pytest from src import health from src.health import ( database_probe, freshness_probe, http_probe, maybe_run, probe, run_checks, write_health, ) class FakeCursor: def __init__(self, results=None, fail=False): self._results = list(results or []) self.queries = [] self.fail = fail def execute(self, sql, params=None): if self.fail: raise RuntimeError("base injoignable") self.queries.append((sql, params)) def fetchone(self): return self._results.pop(0) if self._results else None def __enter__(self): return self def __exit__(self, *a): return False def conn_factory(cursor): @contextlib.contextmanager def _get_conn(): class Conn: def cursor(self_inner): return cursor yield Conn() return _get_conn @pytest.fixture(autouse=True) def _reset_interval(): health._last_run = 0.0 yield health._last_run = 0.0 class TestProbe: def test_transmet_le_resultat(self): assert probe("x", lambda: (True, "ok")) == ("x", True, "ok") def test_convertit_une_exception_en_echec(self): def boom(): raise ValueError("cassé") name, ok, detail = probe("x", boom) assert (name, ok) == ("x", False) assert "ValueError" in detail and "cassé" in detail def test_normalise_un_detail_absent(self): assert probe("x", lambda: (True, None))[2] == "" class TestDatabaseProbe: def test_ok_quand_la_base_repond(self): cur = FakeCursor([(1,)]) assert database_probe(conn_factory(cur))()[0] is True # La sonde doit réellement interroger la base, pas se contenter # d'ouvrir une connexion. assert cur.queries and "SELECT 1" in cur.queries[0][0] def test_echoue_sans_lever_quand_la_base_est_injoignable(self): cur = FakeCursor(fail=True) name, ok, detail = probe("database", database_probe(conn_factory(cur))) assert ok is False assert "injoignable" in detail class TestHttpProbe: def test_ok_sur_200(self): assert http_probe(lambda url, timeout: type("R", (), {"status_code": 200})(), "http://x")()[0] is True def test_echec_sur_500(self): ok, detail = http_probe(lambda url, timeout: type("R", (), {"status_code": 503})(), "http://x")() assert ok is False assert "503" in detail def test_une_erreur_reseau_devient_un_echec_de_sonde(self): def boom(url, timeout): raise ConnectionError("nom d'hôte introuvable") name, ok, detail = probe("dep", http_probe(boom, "http://x")) assert ok is False assert "ConnectionError" in detail class TestFreshnessProbe: def _probe(self, latest, age_h): cur = FakeCursor([(latest,), (age_h,)]) return freshness_probe(conn_factory(cur), "SELECT max(x) FROM t", 12, "test")() def test_ok_quand_les_donnees_sont_fraiches(self): assert self._probe("2026-08-18T10:00:00Z", 3.0)[0] is True def test_echec_quand_les_donnees_sont_trop_anciennes(self): ok, detail = self._probe("2026-08-01T10:00:00Z", 40.0) assert ok is False assert "40.0 h" in detail and "seuil 12 h" in detail def test_borne_exacte_acceptee(self): assert self._probe("x", 12.0)[0] is True # Une base vide au premier démarrage n'est pas une panne. def test_absence_de_donnees_nest_pas_un_echec(self): cur = FakeCursor([(None,)]) ok, detail = freshness_probe(conn_factory(cur), "SELECT max(x) FROM t", 12, "test")() assert ok is True assert "aucune donnée" in detail class TestWriteHealth: def test_agrege_les_sondes_et_enregistre(self): cur = FakeCursor() ok, failed = write_health(conn_factory(cur), "crawler", [("database", True, "ok"), ("dep", True, "ok")]) assert ok is True and failed == [] sql, params = cur.queries[0] assert "INSERT INTO service_health" in sql assert "ON CONFLICT (service) DO UPDATE" in sql # une ligne par service assert params[0] == "crawler" and params[1] is True def test_une_sonde_en_echec_rend_le_service_degrade(self): cur = FakeCursor() ok, failed = write_health(conn_factory(cur), "crawler", [("database", True, "ok"), ("dep", False, "HTTP 503")]) assert ok is False assert failed == ["dep — HTTP 503"] params = cur.queries[0][1] assert params[1] is False assert "dep" in params[3] # la première panne est lisible telle quelle def test_le_detail_par_sonde_est_conserve(self): cur = FakeCursor() write_health(conn_factory(cur), "crawler", [("a", True, ""), ("b", False, "")]) import json assert json.loads(cur.queries[0][1][2]) == {"a": True, "b": False} # Si la base est en panne, la sonde base l'a déjà signalé : échouer ici en # plus ferait tomber le service pour rien. def test_une_base_injoignable_ne_fait_pas_tomber_le_service(self): cur = FakeCursor(fail=True) ok, failed = write_health(conn_factory(cur), "crawler", [("database", False, "ko")]) assert ok is False # pas d'exception propagée class TestRunChecks: def test_journalise_les_sondes_en_echec(self): cur = FakeCursor() messages = [] run_checks(conn_factory(cur), "crawler", [("dep", lambda: (False, "HTTP 503"))], log_event=lambda lvl, msg: messages.append((lvl, msg))) assert messages and messages[0][0] == "warning" assert "DÉGRADÉ" in messages[0][1] and "dep" in messages[0][1] def test_ne_journalise_rien_quand_tout_va_bien(self): cur = FakeCursor() messages = [] run_checks(conn_factory(cur), "crawler", [("dep", lambda: (True, "ok"))], log_event=lambda lvl, msg: messages.append(msg)) assert messages == [] def test_un_log_event_defaillant_ne_fait_pas_tomber_le_controle(self): cur = FakeCursor() def boom(lvl, msg): raise RuntimeError("log cassé") assert run_checks(conn_factory(cur), "crawler", [("dep", lambda: (False, "ko"))], log_event=boom) is False class TestMaybeRun: def test_execute_au_premier_appel(self): cur = FakeCursor() assert maybe_run(conn_factory(cur), "crawler", [("d", lambda: (True, ""))]) is True # Appelée depuis une boucle qui tourne toutes les minutes, la fonction ne # doit pas sonder les dépendances externes à chaque passage. def test_ne_reexecute_pas_avant_lintervalle(self): cur = FakeCursor() probes = [("d", lambda: (True, ""))] maybe_run(conn_factory(cur), "crawler", probes, interval_s=3600) assert maybe_run(conn_factory(cur), "crawler", probes, interval_s=3600) is None def test_force_contourne_lintervalle(self): cur = FakeCursor() probes = [("d", lambda: (True, ""))] maybe_run(conn_factory(cur), "crawler", probes, interval_s=3600) assert maybe_run(conn_factory(cur), "crawler", probes, interval_s=3600, force=True) is True def test_reexecute_une_fois_lintervalle_ecoule(self): cur = FakeCursor() probes = [("d", lambda: (True, ""))] maybe_run(conn_factory(cur), "crawler", probes, interval_s=3600) health._last_run -= 4000 # simule le temps écoulé assert maybe_run(conn_factory(cur), "crawler", probes, interval_s=3600) is True