Set-OPS-Public/scripts/tests/test_genome_colis.py

#!/usr/bin/env python3
"""`genome_colis.py` : le colis porte la branche ET les releases jusqu'à la forge du site.

POURQUOI (2026-10-07). Le colis ne portait que `main`. Les étiquettes de release
n'arrivaient donc jamais sur la forge du site, qui fait autorité (D-81) : `v2026.08.21` et
`v2026.10.07` y manquaient, et il a fallu les pousser à la main.

Sur de vrais dépôts git, dans un répertoire jetable (aucun réseau) :
  - un poste, une forge nue, un runner cloné depuis la forge ;
  - le poste fait un commit et étiquette une release ; `emballer`, puis `recevoir` sur le
    runner, avec la forge pour `pushurl` ;
  - la forge porte la nouvelle tête ET l'étiquette ; un dépôt sans étiquette passe aussi ;
  - une étiquette déplacée sur le poste est REFUSÉE : une release ne bouge pas.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path

RACINE = Path(__file__).resolve().parents[2]
COLIS = RACINE / "scripts" / "genome_colis.py"
ENV = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", "GIT_COMMITTER_NAME": "t",
       "GIT_COMMITTER_EMAIL": "t@t", "GIT_CONFIG_GLOBAL": os.devnull, "GIT_CONFIG_NOSYSTEM": "1"}

ECHECS: list[str] = []


def verifier(cond: bool, msg: str) -> None:
    print(("OK     " if cond else "ECHEC  ") + msg)
    if not cond:
        ECHECS.append(msg)


def git(*args: str, cwd: Path | None = None) -> str:
    r = subprocess.run(["git", *args], cwd=cwd, capture_output=True, text=True, env=ENV)
    if r.returncode != 0:
        raise RuntimeError(f"git {' '.join(args)} : {r.stderr.strip()}")
    return r.stdout.strip()


def colis(*args: str) -> subprocess.CompletedProcess:
    return subprocess.run([sys.executable, str(COLIS), *args], capture_output=True, text=True, env=ENV)


def monter(d: Path) -> tuple[Path, Path, Path]:
    forge, poste, runner = d / "forge.git", d / "poste", d / "runner"
    git("init", "-q", "--bare", "-b", "main", str(forge))
    git("init", "-q", "-b", "main", str(poste))
    (poste / "a").write_text("1")
    git("add", "a", cwd=poste)
    git("commit", "-q", "-m", "un", cwd=poste)
    git("push", "-q", str(forge), "main", cwd=poste)
    git("clone", "-q", str(forge), str(runner))
    return forge, poste, runner


def porter(poste: Path, runner: Path, forge: Path, d: Path) -> subprocess.CompletedProcess:
    paquet = d / "x.bundle"
    e = colis("emballer", str(poste), str(paquet))
    if e.returncode != 0:
        return e
    return colis("recevoir", str(runner), str(paquet), str(forge))


def main() -> int:
    with tempfile.TemporaryDirectory() as d:
        d = Path(d)
        forge, poste, runner = monter(d)

        # Un depot sans etiquette : rien ne change pour lui.
        (poste / "a").write_text("2")
        git("commit", "-q", "-am", "deux", cwd=poste)
        r = porter(poste, runner, forge, d)
        verifier(r.returncode == 0 and git("rev-parse", "main", cwd=forge) == git("rev-parse", "HEAD", cwd=poste),
                 f"sans etiquette : la forge porte la tete du poste ({r.stderr.strip()[-160:]})")

        # Une release annotee.
        (poste / "a").write_text("3")
        git("commit", "-q", "-am", "trois", cwd=poste)
        git("tag", "-a", "v1", "-m", "release v1", cwd=poste)
        r = porter(poste, runner, forge, d)
        recu = json.loads(r.stdout) if r.returncode == 0 else {}
        verifier(r.returncode == 0 and "v1" in git("tag", "-l", cwd=forge).split(),
                 f"la release voyage jusqu'a la forge ({r.stderr.strip()[-160:]})")
        verifier(git("rev-parse", "v1", cwd=forge) == git("rev-parse", "v1", cwd=poste),
                 "la forge porte le MEME objet d'etiquette que le poste (annotee, pas recreee)")
        verifier(recu.get("etiquettes") == ["v1"], f"le runner dit les etiquettes qu'il porte ({recu.get('etiquettes')})")
        e = json.loads(colis("emballer", str(poste), str(d / "y.bundle")).stdout)
        verifier(e.get("etiquettes") == ["v1"], "l'emballage dit les etiquettes du poste")

        # Une release deplacee sur le poste : refusee, la forge garde l'ancienne.
        ancienne = git("rev-parse", "v1", cwd=forge)
        (poste / "a").write_text("4")
        git("commit", "-q", "-am", "quatre", cwd=poste)
        git("tag", "-f", "-a", "v1", "-m", "v1 deplacee", cwd=poste)
        r = porter(poste, runner, forge, d)
        verifier(r.returncode != 0 and git("rev-parse", "v1", cwd=forge) == ancienne,
                 f"une etiquette deplacee est refusee, la forge garde la release ({r.stderr.strip()[-120:]})")

    if ECHECS:
        print(f"\n{len(ECHECS)} echec(s).")
        return 1
    print("\nLe colis porte la branche et les releases ; une release ne bouge pas.")
    return 0


if __name__ == "__main__":
    sys.exit(main())