Set-OPS-Public/scripts/verifier_genome_a_jour.py

#!/usr/bin/env python3
"""Les depots locaux sont-ils a jour sur leur forge, AVANT de deployer ?

UN RUNNER TIRE CE QU'ON POUSSE — IL NE LE RECOIT PAS (2026-09-13).

Trois fois dans une meme soiree, un runner perime a menace de rebatir un etat depasse :

  - le runner du SITE, treize commits en retard, s'appretait a materialiser un locataire
    avec son ancien plan — serveur de sauvegarde inutile, cache d'artefacts en trop,
    runner sans pouvoir de configurer, et aucune machine avec son plancher de memoire ;
  - le runner du LOCATAIRE, clone pendant l'insemination donc avant trois correctifs,
    s'appretait a reposer un plan d'administration pointant le WAN d'un AUTRE site — et
    a refermer derriere lui la porte que l'exploitant venait de rouvrir.

AUCUN DES DEUX N'AURAIT ECHOUE. Un deploiement depuis un genome perime REUSSIT : il
applique fidelement un etat qui n'a plus cours. C'est le mode de defaillance le plus
couteux de cette famille, parce qu'il se presente en vert.

CE QUE CETTE GARDE FAIT, ET CE QU'ELLE NE FAIT PAS :

  en RETARD   -> REFUS. Deployer poserait un etat qu'on sait depasse.
  en AVANCE   -> note, et on continue. Un mainteneur qui travaille localement est normal.
  DIVERGE     -> note, et on continue. C'est a l'humain de trancher, pas a une garde.
  injoignable -> note, et on continue. Une forge en panne ne doit pas bloquer une
                 exploitation ; on le DIT, c'est tout. Le silence serait la faute.

`FORCE=1` passe outre — pour le cas legitime ou l'on deploie sciemment un etat local.
"""
from __future__ import annotations

import os
import subprocess
import sys
from pathlib import Path

RACINE = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(RACINE / "scripts"))

DELAI_FETCH = 20  # s — une forge lente ne doit pas immobiliser un deploiement


def _git(depot: Path, *args: str, delai: int | None = None) -> tuple[int, str]:
    try:
        p = subprocess.run(["git", "-C", str(depot), *args],
                           capture_output=True, text=True, timeout=delai)
        return p.returncode, (p.stdout or p.stderr).strip()
    except subprocess.TimeoutExpired:
        return 124, "delai depasse"
    except Exception as e:                                    # git absent, droits…
        return 1, f"{e.__class__.__name__}: {e}"


def depots_a_verifier() -> list[tuple[str, Path]]:
    """Le moteur, et l'ecosysteme monte. Les deux que le deploiement LIT."""
    out: list[tuple[str, Path]] = [("moteur", RACINE)]
    try:
        from inventory_rules import instance_courante
        inst = instance_courante()
        if inst and Path(inst).resolve() != RACINE:
            out.append(("instance", Path(inst).resolve()))
    except Exception:
        pass
    return out


def etat(depot: Path) -> tuple[str, str]:
    """(verdict, detail) — verdict dans {a-jour, retard, avance, diverge, inconnu}."""
    if not (depot / ".git").exists():
        return "inconnu", "pas un depot git"
    rc, _ = _git(depot, "fetch", "--quiet", delai=DELAI_FETCH)
    if rc != 0:
        return "inconnu", "forge injoignable"
    rc, amont = _git(depot, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}")
    if rc != 0:
        return "inconnu", "aucune branche de suivi"
    rc, compte = _git(depot, "rev-list", "--left-right", "--count", f"HEAD...{amont}")
    if rc != 0:
        return "inconnu", compte
    devant, derriere = (compte.split() + ["0", "0"])[:2]
    devant, derriere = int(devant), int(derriere)
    if derriere and devant:
        return "diverge", f"{devant} devant, {derriere} derriere ({amont})"
    if derriere:
        return "retard", f"{derriere} commit(s) derriere {amont}"
    if devant:
        return "avance", f"{devant} commit(s) devant {amont}"
    return "a-jour", amont


def verifier() -> int:
    retards, notes = [], []
    for role, depot in depots_a_verifier():
        verdict, detail = etat(depot)
        ligne = f"  {role:<9} {depot.name:<20} {detail}"
        if verdict == "retard":
            retards.append(ligne)
        elif verdict != "a-jour":
            notes.append(f"{ligne}   [{verdict}]")

    for n in notes:
        print("note :" + n, file=sys.stderr)

    if not retards:
        print("OK : genome a jour.")
        return 0

    print("Refus: le genome local est EN RETARD sur sa forge.", file=sys.stderr)
    for r in retards:
        print(r, file=sys.stderr)
    print("", file=sys.stderr)
    print("Un deploiement depuis un genome perime REUSSIT — il applique fidelement",
          file=sys.stderr)
    print("un etat qui n'a plus cours, et se presente en vert.", file=sys.stderr)
    print("", file=sys.stderr)
    print("  git -C <depot> pull --ff-only     puis relancer", file=sys.stderr)
    print("  FORCE=1 <commande>                pour deployer l'etat local sciemment",
          file=sys.stderr)
    return 2


if __name__ == "__main__":
    if os.environ.get("FORCE") in ("1", "true", "yes"):
        print("note : FORCE=1 — controle du genome passe outre.", file=sys.stderr)
        raise SystemExit(0)
    raise SystemExit(verifier())