#!/usr/bin/env python3 """La gate du veilleur — elle rejoue la panne du 4 au 7 septembre 2026. python3 -m unittest discover -s prometheus/veilleur -v Elle lance le VRAI `veilleur.py` en sous-processus, contre une doublure HTTP qui rend les réponses MESURÉES pendant l'incident (arcodange-org/tools#36), et elle relève le code de sortie et les messages effectivement poussés. ⚠ Ce qu'elle ne rejoue PAS : la couche ext4/Longhorn qui a causé la panne. Elle rejoue la SURFACE que la panne présentait — c'est-à-dire tout ce que le veilleur peut voir depuis l'extérieur. ⚠ Un test qui ne peut pas rougir ne prouve rien. Les deux moitiés se sabotent : retirer la branche du vecteur vide de `observer()` doit faire ROUGIR `test_cas_c_repond_bien_mais_n_enregistre_rien`, et le nominal doit rester VERT. """ import http.server import json import os import socket import subprocess import sys import threading import unittest ICI = os.path.dirname(os.path.abspath(__file__)) VEILLEUR = os.path.join(ICI, "veilleur.py") # La réponse EXACTE que rendait Prometheus pendant les 3 j 5 h : un succès, un # vecteur vide. C'est le cœur du défaut — il ne ressemble pas à une panne. VECTEUR_VIDE = {"status": "success", "data": {"resultType": "vector", "result": []}} def vecteur(valeur): return { "status": "success", "data": {"resultType": "vector", "result": [{"metric": {}, "value": [0, str(valeur)]}]}, } class Doublure(http.server.BaseHTTPRequestHandler): """Prometheus + Alertmanager + Pushgateway + l'API Telegram, en un serveur.""" def log_message(self, *_args): pass def _rendre(self, code, charge): corps = json.dumps(charge).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(corps))) self.end_headers() self.wfile.write(corps) def do_GET(self): etat = self.server.etat if self.path.startswith("/api/v1/query"): etat["appels_query"] += 1 reponses = etat["reponses_query"] i = min(etat["appels_query"] - 1, len(reponses) - 1) return self._rendre(200, reponses[i]) if self.path.startswith("/api/v1/targets"): # 23 cibles annoncées vivantes — le chiffre mesuré pendant la panne. return self._rendre( 200, {"status": "success", "data": {"activeTargets": [{"health": "up"}] * 23}}, ) if self.path.startswith("/api/v2/alerts"): return self._rendre(200, etat["alertes"]) return self._rendre(404, {"erreur": self.path}) def do_POST(self): etat = self.server.etat taille = int(self.headers.get("Content-Length", "0")) charge = self.rfile.read(taille).decode("utf-8") if "/sendMessage" in self.path: etat["messages"].append(charge) return self._rendre(200, {"ok": True, "result": {"message_id": len(etat["messages"])}}) return self._rendre(404, {"erreur": self.path}) def do_PUT(self): taille = int(self.headers.get("Content-Length", "0")) self.server.etat["battements"].append(self.rfile.read(taille).decode("utf-8")) self._rendre(200, {"ok": True}) def port_ferme(): """Un port sur lequel personne n'écoute : c'est (a), Prometheus mort.""" prise = socket.socket() prise.bind(("127.0.0.1", 0)) numero = prise.getsockname()[1] prise.close() return numero class BancDuVeilleur(unittest.TestCase): def setUp(self): self.serveur = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Doublure) self.serveur.etat = { "reponses_query": [vecteur(5.7)], "appels_query": 0, "alertes": [{"labels": {"alertname": "Veilleur"}, "status": {"state": "active"}}], "messages": [], "battements": [], } threading.Thread(target=self.serveur.serve_forever, daemon=True).start() self.base = "http://127.0.0.1:%d" % self.serveur.server_address[1] def tearDown(self): self.serveur.shutdown() self.serveur.server_close() def lancer(self, **surcharges): environnement = dict(os.environ) environnement.update( { "PROMETHEUS_URL": self.base, "ALERTMANAGER_URL": self.base, "PUSHGATEWAY_URL": self.base, "TELEGRAM_API": self.base, "BOT_TOKEN": "jeton-de-banc", "CHAT_ID": "0", "TENTATIVES": "1", "ENTRE_TENTATIVES_S": "0", "TIMEOUT_S": "3", } ) environnement.update({k: str(v) for k, v in surcharges.items()}) try: fini = subprocess.run( [sys.executable, VEILLEUR], env=environnement, capture_output=True, text=True, timeout=60, ) except subprocess.TimeoutExpired: # ⚠ Un sous-processus TUÉ ne rend pas un code : sans cette branche, # l'assertion « il a crié » serait satisfaite par sa MORT. self.fail("le veilleur n'a pas rendu la main en 60 s — verdict impossible") return fini def messages(self): return self.serveur.etat["messages"] # --- (c) le défaut RÉEL : il répond bien, il n'enregistre rien ------------- def test_cas_c_repond_bien_mais_n_enregistre_rien(self): self.serveur.etat["reponses_query"] = [VECTEUR_VIDE] fini = self.lancer() self.assertEqual(fini.returncode, 0, fini.stdout + fini.stderr) self.assertEqual(len(self.messages()), 1, "le veilleur devait crier UNE fois") self.assertIn("enregistre+RIEN", self.messages()[0].replace("%20", "+")) # Le message doit porter la contradiction qui a ouvert l'enquête. self.assertIn("23", self.messages()[0]) # --- (b) il répond, ses données sont périmées ------------------------------ def test_cas_b_donnees_perimees(self): self.serveur.etat["reponses_query"] = [vecteur(277500)] # 3 j 5 h fini = self.lancer() self.assertEqual(fini.returncode, 0, fini.stdout + fini.stderr) self.assertEqual(len(self.messages()), 1) self.assertIn("retarde", self.messages()[0]) self.assertIn("3+j", self.messages()[0].replace("%20", "+")) # --- (a) il est mort ------------------------------------------------------ def test_cas_a_prometheus_injoignable(self): fini = self.lancer(PROMETHEUS_URL="http://127.0.0.1:%d" % port_ferme()) self.assertEqual(fini.returncode, 0, fini.stdout + fini.stderr) self.assertEqual(len(self.messages()), 1) self.assertIn("injoignable", self.messages()[0]) # --- la chaîne d'alerte est muette ---------------------------------------- def test_temoin_disparu_d_alertmanager(self): self.serveur.etat["alertes"] = [] fini = self.lancer() self.assertEqual(fini.returncode, 0, fini.stdout + fini.stderr) self.assertEqual(len(self.messages()), 1) self.assertIn("muette", self.messages()[0]) # --- LE SILENCE NOMINAL --------------------------------------------------- # Un veilleur qui parle quand tout va bien est pire qu'inutile : on apprend # à ne plus l'ouvrir, et il redevient le silence qu'il devait supprimer. def test_se_tait_quand_tout_va_bien(self): fini = self.lancer() self.assertEqual(fini.returncode, 0, fini.stdout + fini.stderr) self.assertEqual(self.messages(), [], "le veilleur a parlé pour rien") self.assertEqual(len(self.serveur.etat["battements"]), 1, "battement non poussé") self.assertIn("veilleur_prometheus_alarmes 0", self.serveur.etat["battements"][0]) # --- il ne réveille personne pour un redéploiement ------------------------- def test_tolere_un_roulement_du_pod(self): self.serveur.etat["reponses_query"] = [VECTEUR_VIDE, vecteur(4.2)] fini = self.lancer(TENTATIVES=3, ENTRE_TENTATIVES_S="0.1") self.assertEqual(fini.returncode, 0, fini.stdout + fini.stderr) self.assertEqual(self.messages(), [], "une panne transitoire a réveillé le fondateur") # --- l'auto-plainte : il ne peut pas crier --------------------------------- def test_se_plaint_quand_telegram_est_injoignable(self): self.serveur.etat["reponses_query"] = [VECTEUR_VIDE] fini = self.lancer(TELEGRAM_API="http://127.0.0.1:%d" % port_ferme()) self.assertEqual(fini.returncode, 2, fini.stdout + fini.stderr) self.assertIn("n'a PAS PU crier", fini.stdout) # Pas de battement : la règle VeilleurMuet finira par tirer. self.assertEqual(self.serveur.etat["battements"], []) # --- il ne se croit pas sur parole ---------------------------------------- def test_refuse_de_tourner_sans_jeton(self): fini = self.lancer(BOT_TOKEN="") self.assertEqual(fini.returncode, 1, fini.stdout + fini.stderr) self.assertIn("BOT_TOKEN", fini.stdout) if __name__ == "__main__": unittest.main()