tools/prometheus/veilleur/test_veilleur.py

#!/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()