mathben/erplibre
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

[FIX] qemu install : poser mise, pyenv et starship, verdict par outil

Browse source 
Without pipefail, « curl … | sh » returns the status of sh, 0 on empty input:
a failed download passed for an install, its fallback never ran, and the
failure surfaced later as « command not found ». mise and pyenv are downloaded
into a mktemp file, then run, and a check refuses any fallback behind a pipe
without pipefail. starship's installer runs as root under a root timeout, never
reaching the « sudo -v » that sudo-rs refuses, and its shell hook is guarded.
Each tool then reports the version found or « not installed », without
tripping « set -e ».

--- FR ---

Sans pipefail, « curl … | sh » rend le statut de sh, 0 sur une entrée vide :
un téléchargement raté passait pour une pose, son repli ne tournait jamais, et
l'échec se lisait plus loin sous « command not found ». mise et pyenv sont
téléchargés dans un fichier tiré par mktemp, puis exécutés, et un contrôle
refuse tout repli derrière un tube sans pipefail. L'installateur de starship
tourne en root sous un délai root, sans atteindre le « sudo -v » que sudo-rs
refuse, et son crochet de shell est gardé. Chaque outil dit ensuite la version
trouvée ou « non installé », sans faire tomber « set -e ».

Assisted-by: Claude Opus 5
This commit is contained in:
Mathieu Benoit 2026-09-14 14:41:20 -04:00
parent 6ecf4c86ce
commit 01c5290e30
8 changed files with 1164 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
script/install/lib_python_provider.sh
View file

@ -107,18 +107,36 @@ el_mise_install() {
# Installe la version via pyenv, en posant pyenv lui-même au besoin. # Installe la version via pyenv, en posant pyenv lui-même au besoin.
el_pyenv_install() { el_pyenv_install() {
local version="$1" root exe local version="$1" root exe installer
root="$(el_pyenv_root)" root="$(el_pyenv_root)"
exe="$(el_pyenv_exec_path "${version}")" exe="$(el_pyenv_exec_path "${version}")"
if [[ ! -d "${root}" ]]; then if [[ ! -d "${root}" ]]; then
echo "---- Installation de pyenv dans ${root} ----" >&2 echo "---- Installation de pyenv dans ${root} ----" >&2
# « -f » : sans lui, curl livre le CORPS d'une erreur HTTP à bash, qui # Téléchargé dans un fichier, PUIS exécuté. Quand curl alimente bash par
# l'exécute. Une page d'erreur de miroir, un portail captif ou le 504 # un tube, le statut du tube est celui de bash, qui rend 0 sur une entrée
# d'un cache hors ligne devient alors une suite de commandes, et le # vide : sans pipefail — que rien ne pose dans cette chaîne —, un
# lecteur reçoit « command not found » à la place de la cause. Avec # « || return 1 » placé après le tube ne se déclencherait jamais. Un
# « -f », curl rend non nul et le repli ci-dessous s'énonce. # téléchargement raté mènerait alors à « pyenv: command not found », puis
curl -fsSL https://raw.githubusercontent.com/pyenv/pyenv-installer/master/bin/pyenv-installer \ # à un échec de compilation, deux messages qui accusent la mauvaise
| bash >&2 || return 1 # étape. Lu seul, le statut de curl nomme la vraie cause.
#
# « -f » : sans lui, curl écrit le CORPS d'une erreur HTTP — page d'erreur
# de miroir, portail captif, 504 d'un cache hors ligne —, que bash
# exécuterait ensuite comme un script.
installer="$(mktemp)" || return 1
if ! curl -fsSL -o "${installer}" \
https://raw.githubusercontent.com/pyenv/pyenv-installer/master/bin/pyenv-installer; then
echo "Telechargement de l'installateur pyenv impossible" \
"(reseau ou cache) : pyenv n'est pas pose." >&2
rm -f "${installer}"
return 1
fi
if ! bash "${installer}" >&2; then
echo "L'installateur de pyenv a echoue (voir ci-dessus)." >&2
rm -f "${installer}"
return 1
fi
rm -f "${installer}"
fi fi
export PATH="${root}/bin:$PATH" export PATH="${root}/bin:$PATH"
eval "$(pyenv init - 2> /dev/null)" || true eval "$(pyenv init - 2> /dev/null)" || true

50
script/todo/dev_tools.py
View file

@ -34,14 +34,58 @@ STARSHIP_UPSTREAM = f"curl -fsSL {STARSHIP_URL} | sh"
# Sans terminal — une pose par SSH dans une VM — l'installateur demande une # Sans terminal — une pose par SSH dans une VM — l'installateur demande une
# confirmation que personne ne donnera. « -y » la donne d'avance. # confirmation que personne ne donnera. « -y » la donne d'avance.
STARSHIP_UPSTREAM_YES = f"curl -fsSL {STARSHIP_URL} | sh -s -- -y" #
# Et il tourne en root. Sans droit d'écriture sur /usr/local/bin, son
# répertoire par défaut, l'installateur appelle « sudo -v » AVANT de
# télécharger quoi que ce soit. Or « sudo -v » exige un mot de passe dès
# qu'une seule règle sudoers qui vise le compte n'est pas NOPASSWD — celle
# du groupe d'administration ne l'est pas —, et sudo-rs ignore « verifypw »,
# le réglage qui l'en dispenserait. En root le répertoire est inscriptible :
# aucun « sudo -v » n'est atteint, et le binaire atterrit dans un répertoire
# que porte le PATH de tout compte, SSH non interactif compris.
#
# La borne de temps passe DERRIÈRE sudo. Un « timeout » lancé sans privilège
# ne peut pas tuer un installateur root : son signal au groupe de processus
# échoue en EPERM sur les processus root, et sudo ne relaie pas un signal
# venu de son propre groupe. Un « timeout » root, lui, tue l'installateur et
# ses enfants. Il reste en deçà de la borne extérieure de la pose, qui ne
# couvre plus que la moitié non privilégiée du tube : curl.
#
# « -k » : SIGTERM se laisse ignorer, et « timeout » seul attendrait alors
# sans fin. Passé ce délai, il envoie SIGKILL, qui ne s'ignore pas. Borne et
# délai additionnés restent sous la borne extérieure.
STARSHIP_ROOT_TIMEOUT = 280
STARSHIP_ROOT_KILL_AFTER = 10
STARSHIP_UPSTREAM_VM = (
f"curl -fsSL {STARSHIP_URL}"
f" | sudo timeout -k {STARSHIP_ROOT_KILL_AFTER}"
f" {STARSHIP_ROOT_TIMEOUT} sh -s -- -y"
)
# Ce que chaque shell écrit pour lancer starship. La ligne va en FIN de # Ce que chaque shell écrit pour lancer starship. La ligne va en FIN de
# fichier : starship compose le prompt et doit passer après tout ce qui y # fichier : starship compose le prompt et doit passer après tout ce qui y
# touche. # touche.
#
# bash et zsh la gardent par « command -v » : sans binaire — une pose qui a
# échoué, un binaire retiré depuis —, « eval "$(starship init …)" » écrirait
# « command not found » à l'ouverture de chaque shell. La garde est DANS la
# ligne, et non autour de son écriture : elle vaut aussi pour le binaire qui
# disparaît après coup. « starship init » y reste en clair, parce que c'est
# le motif qui dédoublonne, dans le menu de l'hôte comme dans la VM.
#
# Un « if » et non un « && » : la ligne est la DERNIÈRE du fichier, et le
# fichier rend son statut. « a && b » rend 1 quand starship manque, si bien
# qu'un script sous « set -e » qui lit ce fichier s'arrêterait là ; un « if »
# sans branche prise rend 0.
STARSHIP_LINE = { STARSHIP_LINE = {
"bash": 'eval "$(starship init bash)"', "bash": (
"zsh": 'eval "$(starship init zsh)"', "if command -v starship >/dev/null 2>&1; then"
' eval "$(starship init bash)"; fi'
),
"zsh": (
"if command -v starship >/dev/null 2>&1; then"
' eval "$(starship init zsh)"; fi'
),
"fish": "starship init fish | source", "fish": "starship init fish | source",
} }

115
script/todo/qemu_install.py
View file

@ -1592,6 +1592,13 @@ class QemuInstallMixin:
Le tout dans un groupe gardé : ni une panne de réseau ni une extension Le tout dans un groupe gardé : ni une panne de réseau ni une extension
retirée du site ne doivent faire échouer une installation d'une heure. retirée du site ne doivent faire échouer une installation d'une heure.
Deux échecs, deux messages. Un téléchargement raté — réseau coupé,
cache qui n'a pas l'archive — n'apprend rien sur la version de GNOME :
le site sert une archive même à une version qu'il ne connaît pas. Seul
un refus de « gnome-extensions install » la met en cause. Et l'appel à
se reconnecter ne vient que si au moins une extension a été posée :
sans quoi il n'y a rien à charger.
""" """
uuids = " ".join(self._QEMU_GNOME_EXT_UUIDS) uuids = " ".join(self._QEMU_GNOME_EXT_UUIDS)
site = self._QEMU_GNOME_EXT_SITE site = self._QEMU_GNOME_EXT_SITE
@ -1617,19 +1624,31 @@ class QemuInstallMixin:
'gx() { if [ -z "$DBUS_SESSION_BUS_ADDRESS" ] && ' 'gx() { if [ -z "$DBUS_SESSION_BUS_ADDRESS" ] && '
"command -v dbus-run-session >/dev/null 2>&1; then " "command -v dbus-run-session >/dev/null 2>&1; then "
'dbus-run-session -- gnome-extensions "$@"; ' 'dbus-run-session -- gnome-extensions "$@"; '
'else gnome-extensions "$@"; fi; }; ' + f"for u in {uuids}; do " 'else gnome-extensions "$@"; fi; }; '
# « || echo » DANS la substitution : un mktemp qui échoue rendrait # `n` compte les extensions réellement posées.
# l'affectation non nulle, et « set -e » couperait toute la suite. + f"n=0; for u in {uuids}; do "
+ "z=$(mktemp -p /var/tmp gext-XXXX.zip || echo /var/tmp/gext.zip); " # L'archive ne va que dans un nom tiré par mktemp, créé par ce
+ 'if curl -fsSL --max-time 120 "' # compte seul : un nom fixe dans /var/tmp, ouvert à tous, pourrait
# y être posé d'avance par un autre. L'affectation est DANS la
# condition, si bien qu'un mktemp qui échoue ne fait pas tomber
# « set -e » ; « z » reste alors vide, et l'extension est sautée
# comme un téléchargement raté, sans rien à effacer.
+ "if ! z=$(mktemp -p /var/tmp gext-XXXX.zip 2>/dev/null); "
+ 'then z=""; fi; '
+ 'if [ -z "$z" ] || ! curl -fsSL --max-time 120 "'
+ site + site
+ '/$u.shell-extension.zip?shell_version=$sv" -o "$z" ' + '/$u.shell-extension.zip?shell_version=$sv" -o "$z"; then '
+ '&& gx install --force "$z" >/dev/null 2>&1; then ' + f'echo " ⚠ {t("download impossible (network or cache):")} '
+ 'gx enable "$u" >/dev/null 2>&1 || true; ' + '$u"; '
+ f'echo " {t("installed and enabled:")} $u"; else ' + 'elif ! gx install --force "$z" >/dev/null 2>&1; then '
+ f'echo " {t("not available for this GNOME, skipped:")} ' + f'echo " {t("not available for this GNOME, skipped:")} '
+ '$u (GNOME $sv)"; fi; rm -f "$z"; done; ' + '$u (GNOME $sv)"; '
+ f'echo " {t("log out and back in to load them")}"; ' + "else "
+ 'gx enable "$u" >/dev/null 2>&1 || true; n=$((n+1)); '
+ f'echo " {t("installed and enabled:")} $u"; '
+ 'fi; if [ -n "$z" ]; then rm -f "$z"; fi; done; '
+ 'if [ "$n" -gt 0 ]; then '
+ f'echo " {t("log out and back in to load them")}"; fi; '
+ "fi; } || true; " + "fi; } || true; "
) )
@ -2210,6 +2229,13 @@ class QemuInstallMixin:
« </dev/null » la lui fait rater tout de suite, « timeout » borne le « </dev/null » la lui fait rater tout de suite, « timeout » borne le
reste. C'est aussi pourquoi starship reçoit « -y ». reste. C'est aussi pourquoi starship reçoit « -y ».
Chaque installateur amont est suivi d'une ligne de verdict : la version
du binaire quand il est là, « ⚠ … non installé (voir ci-dessus) »
sinon. Le code de sortie de la pose ne peut pas la donner : sans
pipefail, « curl | sh » rend celui de sh, 0 sur une entrée vide, si
bien qu'un téléchargement raté passe pour une pose réussie. Seule la
présence du binaire tranche.
Les lignes ajoutées à un fichier du HOME le sont UNE fois : sans le Les lignes ajoutées à un fichier du HOME le sont UNE fois : sans le
« grep » qui précède, chaque redéploiement d'une même VM rallonge son « grep » qui précède, chaque redéploiement d'une même VM rallonge son
~/.bashrc — ou son historique — d'une ligne identique. ~/.bashrc — ou son historique — d'une ligne identique.
@ -2221,10 +2247,12 @@ class QemuInstallMixin:
Ce que le dépôt seul peut donner — hooks git, commandes Claude — n'est Ce que le dépôt seul peut donner — hooks git, commandes Claude — n'est
pas ici : voir _qemu_aidev_after_cmd. pas ici : voir _qemu_aidev_after_cmd.
""" """
commande, repertoire = dev_tools.AGENTS.get( # Une valeur vide ou inconnue retombe sur le défaut : un choix
agent or dev_tools.AGENT_DEFAUT, # inattendu ne doit pas faire tomber un déploiement.
dev_tools.AGENTS[dev_tools.AGENT_DEFAUT], nom_agent = (
agent if agent in dev_tools.AGENTS else dev_tools.AGENT_DEFAUT
) )
commande, repertoire = dev_tools.AGENTS[nom_agent]
repertoire = repertoire.replace("~/", "$HOME/", 1) repertoire = repertoire.replace("~/", "$HOME/", 1)
prompt = dev_tools.STARSHIP_LINE["bash"] prompt = dev_tools.STARSHIP_LINE["bash"]
path_line = f'export PATH="{repertoire}:$PATH"' path_line = f'export PATH="{repertoire}:$PATH"'
@ -2248,6 +2276,22 @@ class QemuInstallMixin:
f" || echo {shlex.quote(ligne)} >> {fichier}; " f" || echo {shlex.quote(ligne)} >> {fichier}; "
) )
def verdict(var, nom, repli=""):
# `var` reçoit le chemin du binaire : « command -v » d'abord, puis
# `repli`, le chemin où l'installateur le pose hors du PATH de ce
# shell. Aucune ligne ne peut faire tomber « set -e » : l'« if »
# absorbe le test, et la version se lit sous « timeout », sans
# entrée, comme une pose.
trouve = f'echo "{repli}"' if repli else "true"
return (
f'{var}="$(command -v {nom} || {trouve})"; '
f'if [ -x "${var}" ]; then echo " {nom}: '
f'$(timeout 10 "${var}" --version </dev/null 2>&1'
' | head -n 1 || true)"; '
f'else echo " ⚠ {nom} {t("not installed (see above)")}"; '
"fi; "
)
return ( return (
f'echo "== {t("AI coding tools")} =="; ' f'echo "== {t("AI coding tools")} =="; '
+ self._qemu_pkg_install_cmd(self._QEMU_AIDEV_PKGS) + self._qemu_pkg_install_cmd(self._QEMU_AIDEV_PKGS)
@ -2256,12 +2300,18 @@ class QemuInstallMixin:
# commande distante a été figé au démarrage du shell SSH, avant que # commande distante a été figé au démarrage du shell SSH, avant que
# l'installateur ne pose le binaire. Le code 127 qu'on obtiendrait # l'installateur ne pose le binaire. Le code 127 qu'on obtiendrait
# sinon ne dirait pas que le hook n'a pas été écrit. # sinon ne dirait pas que le hook n'a pas été écrit.
+ f'RTK="$(command -v rtk || echo "{local_bin}/rtk")"; ' + verdict("RTK", "rtk", f"{local_bin}/rtk")
+ '[ -x "$RTK" ] && timeout 60 "$RTK" init --global' + '[ -x "$RTK" ] && timeout 60 "$RTK" init --global'
" </dev/null >/dev/null 2>&1 || true; " " </dev/null >/dev/null 2>&1 || true; "
+ pose(dev_tools.STARSHIP_UPSTREAM_YES, 300) # En root, starship atterrit dans /usr/local/bin, que le PATH de
# ce shell porte déjà : « command -v » le trouve sans repli. Ces
# 300 s ne bornent que curl : l'installateur root porte sa propre
# borne, plus courte, derrière sudo — voir STARSHIP_UPSTREAM_VM.
+ pose(dev_tools.STARSHIP_UPSTREAM_VM, 300)
+ verdict("STARSHIP", "starship")
+ une_fois(prompt, "starship init bash") + une_fois(prompt, "starship init bash")
+ pose(commande, 600) + pose(commande, 600)
+ verdict("AGENT", nom_agent, f"{repertoire}/{nom_agent}")
+ une_fois(local_line, local_bin) + une_fois(local_line, local_bin)
# Claude Code s'installe DANS ~/.local/bin : la ligne serait la # Claude Code s'installe DANS ~/.local/bin : la ligne serait la
# même, écrite deux fois dans le journal pour un seul effet. # même, écrite deux fois dans le journal pour un seul effet.
@ -2449,7 +2499,20 @@ class QemuInstallMixin:
défaut — même raison que pour cargo et rustc. défaut — même raison que pour cargo et rustc.
Sans mise utilisable, rien n'est écrit : lib_python_provider.sh Sans mise utilisable, rien n'est écrit : lib_python_provider.sh
retombe alors sur pyenv toute seule.""" retombe alors sur pyenv toute seule.
L'installateur est téléchargé dans un fichier, PUIS exécuté. Dans
« curl … | sh || repli », le statut du tube est celui de sh, qui rend
0 sur une entrée vide : sans pipefail, le repli ne se déclencherait
jamais, et un téléchargement raté passerait pour une pose réussie. Le
statut de curl, lu seul, distingue les deux échecs — rien obtenu,
ou un installateur qui a échoué — et chacun a son message. Aucun ne
fait tomber « set -e » : ils sont testés dans un « if ».
Le fichier ne s'obtient que de mktemp : un nom aléatoire, créé par ce
compte seul. Sans mktemp, rien n'est téléchargé. Un nom fixe dans
/tmp, qu'un autre compte peut créer d'avance, serait exécuté par
root."""
if python_provider == "pyenv": if python_provider == "pyenv":
# Explicite : même si mise se trouvait déjà dans l'image, on ne # Explicite : même si mise se trouvait déjà dans l'image, on ne
# l'utilise pas. Sans cela le mode « auto » du dépôt le prendrait. # l'utilise pas. Sans cela le mode « auto » du dépôt le prendrait.
@ -2461,11 +2524,23 @@ class QemuInstallMixin:
"if command -v mise >/dev/null 2>&1; then " "if command -v mise >/dev/null 2>&1; then "
'echo " mise: $(mise --version)"; ' 'echo " mise: $(mise --version)"; '
"else " "else "
# L'affectation est DANS la condition : un mktemp qui échoue prend
# la branche du téléchargement impossible au lieu de faire tomber
# « set -e ». `f` y reste vide, et « rm -f "" » rend 0.
"if ! f=$(mktemp 2>/dev/null) "
'|| ! curl -fsSL https://mise.run -o "$f"; then '
'echo " ⚠ '
+ t(
"mise download impossible (network or cache): "
"pyenv will take over"
)
+ '"; '
# La variable est passée À sudo, pas exportée avant : « sudo -E » # La variable est passée À sudo, pas exportée avant : « sudo -E »
# dépend de env_reset dans sudoers et n'est pas garanti. # dépend de env_reset dans sudoers et n'est pas garanti.
"curl -fsSL https://mise.run " + 'elif ! sudo MISE_INSTALL_PATH=/usr/local/bin/mise sh "$f"'
"| sudo MISE_INSTALL_PATH=/usr/local/bin/mise sh " " </dev/null; then "
'|| echo " mise indisponible ici : pyenv prendra le relais"; ' + f'echo " ⚠ {t("mise installer failed: pyenv will take over")}"; '
+ 'fi; rm -f "$f"; '
"fi; " "fi; "
# « auto », et non « mise » : si l'installation ci-dessus a échoué, # « auto », et non « mise » : si l'installation ci-dessus a échoué,
# lib_python_provider.sh doit pouvoir retomber sur pyenv. # lib_python_provider.sh doit pouvoir retomber sur pyenv.

16
script/todo/todo_i18n.py
View file

@ -6423,6 +6423,22 @@ TRANSLATIONS = {
"fr": "Chemin de la clé publique SSH", "fr": "Chemin de la clé publique SSH",
"en": "SSH public key path", "en": "SSH public key path",
}, },
"not installed (see above)": {
"fr": "non installé (voir ci-dessus)",
"en": "not installed (see above)",
},
"download impossible (network or cache):": {
"fr": "téléchargement impossible (réseau ou cache) :",
"en": "download impossible (network or cache):",
},
"mise download impossible (network or cache): pyenv will take over": {
"fr": "téléchargement de mise impossible (réseau ou cache) : pyenv prendra le relais",
"en": "mise download impossible (network or cache): pyenv will take over",
},
"mise installer failed: pyenv will take over": {
"fr": "l'installateur de mise a échoué : pyenv prendra le relais",
"en": "mise installer failed: pyenv will take over",
},
"timezone not set": { "timezone not set": {
"fr": "fuseau non posé", "fr": "fuseau non posé",
"en": "timezone not set", "en": "timezone not set",

568
test/test_install_pipe_to_shell.py
View file

@ -10,22 +10,186 @@ l'exécute : une page de miroir en panne, un portail captif ou le 504 d'un
cache hors ligne devient une suite de commandes. Le lecteur reçoit alors cache hors ligne devient une suite de commandes. Le lecteur reçoit alors
« command not found » et la cause véritable ne se lit plus nulle part. « command not found » et la cause véritable ne se lit plus nulle part.
Et « -f » ne suffit pas à tout. Le statut d'un tube est celui de son DERNIER
membre : sans pipefail, « curl -f … | sh || repli » rend celui de sh, 0 sur
une entrée vide, et le repli ne se déclenche jamais. Il en va de même de
« tube && a || repli » et du « else » d'un « if » dont le tube est la
condition. Un téléchargement raté passe alors pour une pose réussie. Là où
un échec doit avoir une suite, le téléchargement va dans un fichier, et
c'est le statut de curl qu'on lit.
Le contrôle porte sur la PROPRIÉTÉ et non sur un fichier : tout script du Le contrôle porte sur la PROPRIÉTÉ et non sur un fichier : tout script du
dépôt qui tube un téléchargement dans un interpréteur doit demander à curl dépôt, et toute commande distante que l'hôte compose pour une VM, qui tube
d'échouer. un téléchargement dans un interpréteur doit demander à curl d'échouer, et
ne rien attendre d'un « || » posé derrière le tube.
""" """
import os
import re import re
import shutil
import subprocess
import sys import sys
import tempfile
import unittest import unittest
from pathlib import Path from pathlib import Path
RACINE = Path(__file__).resolve().parent.parent RACINE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(RACINE)) sys.path.insert(0, str(RACINE))
sys.argv = ["todo.py"]
from script.todo import dev_tools # noqa: E402
from script.todo.todo import TODO # noqa: E402
# « curl » suivi de ce qui n'est pas une nouvelle commande, jusqu'à un tube # « curl » suivi de ce qui n'est pas une nouvelle commande, jusqu'à un tube
# vers un interpréteur. Le « \\ » de continuation de ligne est traversé. # vers un interpréteur. Le « \\ » de continuation de ligne est traversé.
TUBE = re.compile(r"curl\s+((?:[^\n|;&]|\\\n)*?)\|\s*(?:sudo\s+)?(?:ba)?sh\b") #
# Entre le tube et l'interpréteur, un « sudo » éventuel, avec ses options
# (« -E », « -n ») et ses affectations (« VAR=valeur ») : c'est la forme d'un
# installateur qui doit écrire hors du HOME. Limite connue : une option qui
# prend son argument à part — « sudo -u compte bash » — n'est pas reconnue,
# faute de savoir, sans la table des options de sudo, quel mot est un argument
# et lequel est la commande.
#
# Puis un « timeout » éventuel, avec ses options et sa durée : la borne d'un
# installateur root se pose derrière sudo, seule place d'où elle peut le tuer.
TUBE = re.compile(
r"curl\s+((?:[^\n|;&]|\\\n)*?)"
r"\|\s*(?:sudo(?:\s+(?:-\S+|\w+=\S+))*\s+)?"
r"(?:timeout(?:\s+(?:-[ks]\s+\S+|-\S+))*\s+\d+(?:\.\d+)?[smhd]?\s+)?"
r"(?:ba)?sh\b"
)
# Ce qui suit le tube jusqu'à la fin de la liste de commandes — « ; », fin de
# ligne non continuée —, puis un « || » et le repli qu'il porte. Un « | » seul
# prolonge le tube, un « & » seul appartient à une redirection (« >&2 »). Un
# « && » est traversé : dans « tube && a || repli », le repli suit aussi un
# tube qui a « réussi » sur une entrée vide, c'est-à-dire jamais.
REPLI = re.compile(r"(?:[^\n;&|]|\\\n|\|(?!\|)|&&|&(?!&))*\|\|\s*([^\n;]*)")
# Un « set » et ses arguments, jusqu'à la fin de la commande.
SET = re.compile(r"\bset((?:[ \t]+[^\s;&|()]+)+)")
# Le tube est-il la condition d'un « if » ou d'un « elif » ? Rien ne sépare
# le mot-clé du tube, sinon un « && » ou un « || ». Groupe 1 : un « ! » qui
# nie la condition.
SI = re.compile(r"\b(?:if|elif)\s+(!\s+)?([^;\n]*)$")
# Jusqu'au « then » de ce « if » : la fin de la condition.
ALORS = re.compile(r"(?:[^\n;]|\\\n)*[;\n]\s*then\b")
MOTS_SI = re.compile(r"\b(if|elif|else|fi)\b")
def bascules_pipefail(texte: str):
"""(position, allumé) de chaque réglage de pipefail, hors commentaire.
« set -o pipefail » et ses formes groupées (« -eo », « -euo ») allument :
le « o » est la dernière lettre d'une option à MOINS, et le mot suivant
est « pipefail ». « set +o pipefail » éteint. Dans un même « set », le
dernier réglage l'emporte, comme pour le shell.
"""
for m in SET.finditer(texte):
debut_ligne = texte.rfind("\n", 0, m.start()) + 1
if re.search(r"(?:^|\s)#", texte[debut_ligne : m.start()]):
continue
mots = m.group(1).split()
for mot, suivant in zip(mots, mots[1:]):
if suivant == "pipefail" and re.fullmatch(r"[-+]\w*o", mot):
yield m.start(), mot[0] == "-"
def debut_de_ligne_logique(src: str, pos: int) -> int:
"""Le début de la ligne qui porte `pos`, continuations « \\ » comprises."""
i = src.rfind("\n", 0, pos)
while i > 0 and src[i - 1] == "\\":
i = src.rfind("\n", 0, i - 1)
return i + 1
def prologue(src: str) -> int:
"""La fin de l'en-tête du fichier : lignes vides, commentaires et « set »
qui précèdent la première autre commande."""
fin = 0
for ligne in src.splitlines(keepends=True):
nue = ligne.strip()
if nue and not nue.startswith("#") and not re.match(r"set\b", nue):
break
fin += len(ligne)
return fin
def pipefail_actif(src: str, pos: int) -> bool:
"""pipefail vaut-il pour le tube qui commence à `pos` ?
Deux portées, et deux seulement. D'abord la ligne logique du tube, avant
lui : le dernier réglage l'emporte. Les sous-shells refermés « ( … ) »,
« $( … ) » et les chaînes refermées « '…' » en sont retirés, parce qu'un
« set » posé là ne vaut que pour eux. Sans réglage sur la ligne, l'en-tête
du fichier : il doit allumer pipefail, et aucun « set +o pipefail » ne
doit l'éteindre entre lui et le tube.
Un pipefail posé ailleurs — dans une fonction, plus bas dans le fichier —
ne compte pas. L'erreur penche du côté du signalement : un repli exempté
à tort serait un repli sans effet que plus rien ne signale.
"""
ligne = src[debut_de_ligne_logique(src, pos) : pos]
retire = 1
while retire:
ligne, retire = re.subn(r"\([^()]*\)", "", ligne)
ligne = re.sub(r"'[^']*'", "", ligne)
sur_la_ligne = list(bascules_pipefail(ligne))
if sur_la_ligne:
return sur_la_ligne[-1][1]
fin = prologue(src)
en_tete = list(bascules_pipefail(src[:fin]))
if not en_tete or not en_tete[-1][1]:
return False
return all(allume for _p, allume in bascules_pipefail(src[fin:pos]))
def branche_de_repli(src: str, debut: int, niee: bool):
"""La branche qu'un « if » réserve à l'échec de sa condition.
`debut` suit le « then ». Condition niée (« if ! … »), c'est la branche
« then » elle-même ; sinon, le « else » ou l'« elif » de ce même « if »,
les « if » imbriqués étant sautés. None s'il n'y en a pas.
"""
profondeur = 0
for m in MOTS_SI.finditer(src, debut):
mot = m.group(1)
if mot == "if":
profondeur += 1
elif mot == "fi" and profondeur:
profondeur -= 1
elif profondeur == 0 and niee:
return src[debut : m.start()].strip(" \t\n;")
elif mot == "fi":
return None
elif profondeur == 0:
fin = src.find("\n", m.end())
corps = src[m.end() : fin if fin >= 0 else len(src)]
corps = corps.strip(" \t;")
return corps if mot == "else" else f"elif {corps}"
return None
def repli_du_tube(src: str, m):
"""Le texte du repli qui lit le statut du tube `m`, ou None.
Deux formes : un « || » derrière le tube, et un « if » dont le tube est
la condition."""
suite = REPLI.match(src, m.end())
if suite:
return suite.group(1)
debut = debut_de_ligne_logique(src, m.start())
si = SI.search(src[debut : m.start()].replace("\\\n", " "))
if not si:
return None
alors = ALORS.match(src, m.end())
if not alors:
return None
niee = bool(si.group(1)) and not re.search(r"&&|\|\|", si.group(2))
return branche_de_repli(src, alors.end(), niee)
def demande_lechec(options: str) -> bool: def demande_lechec(options: str) -> bool:
@ -42,20 +206,71 @@ def demande_lechec(options: str) -> bool:
return False return False
def replis_sans_effet(src: str):
"""Les replis qui lisent le statut d'un tube vers un shell, sans pipefail.
Rend le texte de chaque repli. « || true » et « || : » n'en sont pas :
ils absorbent un échec, ils n'y réagissent pas, et ne promettent donc
rien que le tube ne tienne.
"""
trouves = []
for m in TUBE.finditer(src):
if pipefail_actif(src, m.start()):
continue
repli = repli_du_tube(src, m)
if repli is None:
continue
mots = repli.split()
premier = mots[0].strip("'\")") if mots else ""
if premier not in ("true", ":"):
trouves.append(repli.strip())
return trouves
def scripts(): def scripts():
for chemin in sorted(RACINE.glob("script/**/*.sh")): for chemin in sorted(RACINE.glob("script/**/*.sh")):
yield chemin, chemin.read_text(encoding="utf-8", errors="replace") yield chemin, chemin.read_text(encoding="utf-8", errors="replace")
def commandes_de_lhote():
"""Les commandes que l'hôte compose et envoie à une VM, par leur nom.
Elles ne vivent dans aucun .sh : un tube y est coupé entre plusieurs
littéraux Python, qu'une lecture ligne à ligne du source ne recolle pas.
On les construit donc, comme le fait le déploiement.
"""
todo = TODO.__new__(TODO)
yield "dev_tools.RTK_UPSTREAM", dev_tools.RTK_UPSTREAM
yield "dev_tools.STARSHIP_UPSTREAM", dev_tools.STARSHIP_UPSTREAM
yield "dev_tools.STARSHIP_UPSTREAM_VM", dev_tools.STARSHIP_UPSTREAM_VM
for agent, (commande, _repertoire) in sorted(dev_tools.AGENTS.items()):
yield f"dev_tools.AGENTS[{agent}]", commande
yield (
f"_qemu_aidev_remote_cmd({agent})",
todo._qemu_aidev_remote_cmd(agent),
)
yield "_qemu_mise_remote_cmd(mise)", todo._qemu_mise_remote_cmd("mise")
yield "_qemu_gnome_ext_remote_cmd", todo._qemu_gnome_ext_remote_cmd()
yield "_qemu_pycharm_remote_cmd", todo._qemu_pycharm_remote_cmd()
yield (
"_qemu_android_studio_remote_cmd",
todo._qemu_android_studio_remote_cmd(),
)
def sources():
for chemin, src in scripts():
yield str(chemin.relative_to(RACINE)), src
yield from commandes_de_lhote()
class TestUnTelechargementTubeDansUnShell(unittest.TestCase): class TestUnTelechargementTubeDansUnShell(unittest.TestCase):
def test_curl_doit_echouer_sur_une_erreur_http(self): def test_curl_doit_echouer_sur_une_erreur_http(self):
fautifs = [] fautifs = []
for chemin, src in scripts(): for nom, src in sources():
for options in TUBE.findall(src): for options in TUBE.findall(src):
if not demande_lechec(options): if not demande_lechec(options):
fautifs.append( fautifs.append(f"{nom} : curl {options.strip()}")
f"{chemin.relative_to(RACINE)} : curl {options.strip()}"
)
self.assertEqual( self.assertEqual(
fautifs, fautifs,
[], [],
@ -63,6 +278,21 @@ class TestUnTelechargementTubeDansUnShell(unittest.TestCase):
+ "\n ".join(fautifs), + "\n ".join(fautifs),
) )
def test_aucun_repli_derriere_un_tube_sans_pipefail(self):
"""Le repli d'un tube sans pipefail ne se déclenche jamais : c'est
le statut de l'interpréteur qu'il lit, 0 sur une entrée vide."""
fautifs = [
f"{nom} : || {repli}"
for nom, src in sources()
for repli in replis_sans_effet(src)
]
self.assertEqual(
fautifs,
[],
"repli sans effet, télécharger dans un fichier :\n "
+ "\n ".join(fautifs),
)
def test_les_options_groupees_sont_reconnues(self): def test_les_options_groupees_sont_reconnues(self):
"""« -sSf » vaut « -s -S -f » : les traiter comme un seul jeton """« -sSf » vaut « -s -S -f » : les traiter comme un seul jeton
signalerait à tort la moitié des appels du dépôt.""" signalerait à tort la moitié des appels du dépôt."""
@ -71,6 +301,136 @@ class TestUnTelechargementTubeDansUnShell(unittest.TestCase):
for mauvais in ("-L", "-sSL", "", "--silent"): for mauvais in ("-L", "-sSL", "", "--silent"):
self.assertFalse(demande_lechec(f"{mauvais} https://x"), mauvais) self.assertFalse(demande_lechec(f"{mauvais} https://x"), mauvais)
def test_le_motif_voit_un_sudo_devant_linterpreteur(self):
"""Un installateur qui écrit hors du HOME passe par sudo, avec ses
options et ses variables : le motif doit le voir quand même."""
for tube in (
"curl -sL https://example.invalid/i | sh",
"curl -sL https://example.invalid/i | bash -s -- -y",
"curl -sL https://example.invalid/i | sudo sh",
"curl -sL https://example.invalid/i | sudo -E bash",
"curl -sL https://example.invalid/i | sudo -n -E bash",
"curl -sL https://example.invalid/i | sudo A=/opt/a sh",
"curl -sL https://example.invalid/i | sudo -E A=1 B=2 sh -s",
"curl -sL https://example.invalid/i | sudo timeout 280 sh -s -- -y",
"curl -sL https://example.invalid/i"
" | sudo timeout -k 10 280 sh -s -- -y",
"curl -sL https://example.invalid/i | timeout 60 bash",
"curl -sL https://example.invalid/i | sudo -n timeout -k 5 9m sh",
"curl -sL https://example.invalid/i | sudo timeout --kill-after=5"
" 280 sh",
):
self.assertEqual(
["-sL https://example.invalid/i "], TUBE.findall(tube), tube
)
for pas_un_shell in (
"curl -sL https://example.invalid/i | shasum",
"curl -sL https://example.invalid/i | sudo tee /x",
"curl -sL https://example.invalid/i | sudo timeout 280 tee /x",
"curl -sL https://example.invalid/i -o /tmp/i; sh /tmp/i",
):
self.assertEqual([], TUBE.findall(pas_un_shell), pas_un_shell)
def test_un_repli_derriere_le_tube_est_reconnu(self):
"""Les formes qui promettent un repli qu'elles ne tiennent pas."""
for tube in (
"curl -fsSL https://example.invalid/i | sh || echo repli",
"curl -fsSL https://example.invalid/i | bash >&2 || return 1",
"curl -fsSL https://example.invalid/i | sudo A=1 sh || exit 1",
"curl -fsSL https://example.invalid/i \\\n | bash \\\n || return 1",
"timeout 9 sh -c 'curl -fsSL https://example.invalid/i | sh'"
" </dev/null || echo repli",
# « && » : le repli ne suit qu'un échec de `ok`, jamais du tube.
"curl -fsSL https://example.invalid/i | sh && echo ok"
" || echo repli",
# La condition d'un « if » est le statut de l'interpréteur.
"if curl -fsSL https://example.invalid/i | sh; then echo ok;"
" else echo repli; fi",
"if curl -fsSL https://example.invalid/i | sh\nthen\n echo ok\n"
"else\n echo repli\nfi",
"if curl -fsSL https://example.invalid/i | sh; then echo ok;"
" elif x; then echo repli; fi",
"if ! curl -fsSL https://example.invalid/i | sh; then"
" echo repli; fi",
"if timeout 9 sh -c 'curl -fsSL https://example.invalid/i | sh'"
" </dev/null; then echo ok; else echo repli; fi",
# Le « else » est celui du « if » du tube, pas d'un imbriqué.
"if curl -fsSL https://example.invalid/i | sh; then"
" if a; then b; fi; else echo repli; fi",
# « +o » ÉTEINT pipefail, et le dernier réglage l'emporte.
"set +o pipefail\n"
"curl -fsSL https://example.invalid/i | sh || echo repli",
"set +eo pipefail\n"
"curl -fsSL https://example.invalid/i | sh || echo repli",
"set -euo pipefail\nset +o pipefail\n"
"curl -fsSL https://example.invalid/i | sh || echo repli",
"set -euo pipefail\nx=1\nset +o pipefail\n"
"curl -fsSL https://example.invalid/i | sh || echo repli",
"set -o pipefail; set +o pipefail; "
"curl -fsSL https://example.invalid/i | sh || echo repli",
# Un pipefail qui ne vaut pas pour le tube.
"x=1\nset -o pipefail\n"
"curl -fsSL https://example.invalid/i | sh || echo repli",
"f() { set -o pipefail; }\n"
"curl -fsSL https://example.invalid/i | sh || echo repli",
"(set -o pipefail; true); "
"curl -fsSL https://example.invalid/i | sh || echo repli",
"x=$(set -o pipefail; echo); "
"curl -fsSL https://example.invalid/i | sh || echo repli",
"sh -c 'set -o pipefail; true'; "
"curl -fsSL https://example.invalid/i | sh || echo repli",
):
self.assertEqual(1, len(replis_sans_effet(tube)), tube)
def test_ce_qui_nest_pas_un_repli_passe(self):
"""« || true » n'attend rien du tube ; après « ; » ou une fin de
ligne, le « || » porte sur une autre commande ; pipefail rend le
repli effectif ; un tube hors de la condition d'un « if », ou un
« if » sans « else », ne promet rien ; un téléchargement dans un
fichier n'est pas un tube."""
for correct in (
"curl -fsSL https://example.invalid/i | sh || true",
"timeout 9 sh -c 'curl -fsSL https://example.invalid/i | sh'"
" </dev/null || true",
"curl -fsSL https://example.invalid/i | sh && echo ok || true",
"curl -fsSL https://example.invalid/i | sh; x || echo autre",
"curl -fsSL https://example.invalid/i | sh\nx || echo autre",
"set -euo pipefail\n"
"curl -fsSL https://example.invalid/i | sh || echo repli",
"#!/bin/bash\n# en-tête\n\nset -eu\nset -o pipefail\nx=1\n"
"curl -fsSL https://example.invalid/i | sh || echo repli",
"set -o pipefail; "
"curl -fsSL https://example.invalid/i | sh || echo repli",
"set -e -o pipefail; "
"curl -fsSL https://example.invalid/i | sh || echo repli",
"set -o errexit -o pipefail; "
"curl -fsSL https://example.invalid/i | sh || echo repli",
"set +o pipefail; set -o pipefail; "
"curl -fsSL https://example.invalid/i | sh || echo repli",
"timeout 9 sh -c 'set -o pipefail; "
"curl -fsSL https://example.invalid/i | sh' </dev/null"
" || echo repli",
"set -o pipefail; if curl -fsSL https://example.invalid/i | sh;"
" then echo ok; else echo repli; fi",
"if curl -fsSL https://example.invalid/i | sh; then echo ok; fi",
"if x; then curl -fsSL https://example.invalid/i | sh;"
" else echo autre; fi",
"if curl -fsSL https://example.invalid/i | sh; then"
" if a; then b; else c; fi; fi",
"if ! curl -fsSL https://example.invalid/i | sh; then true; fi",
'if ! curl -fsSL https://example.invalid/i -o "$f"; then'
" echo repli; fi",
):
self.assertEqual([], replis_sans_effet(correct), correct)
def test_un_pipefail_en_commentaire_ne_compte_pas(self):
"""Un commentaire qui parle de pipefail ne le pose pas."""
src = (
"# sans set -o pipefail, voir plus bas\n"
"curl -fsSL https://example.invalid/i | sh || echo repli\n"
)
self.assertEqual(1, len(replis_sans_effet(src)))
def test_le_motif_trouve_bien_les_tubes_existants(self): def test_le_motif_trouve_bien_les_tubes_existants(self):
"""Une expression qui ne trouve rien passerait tous les contrôles. """Une expression qui ne trouve rien passerait tous les contrôles.
@ -80,6 +440,200 @@ class TestUnTelechargementTubeDansUnShell(unittest.TestCase):
trouves = sum(len(TUBE.findall(src)) for _c, src in scripts()) trouves = sum(len(TUBE.findall(src)) for _c, src in scripts())
self.assertGreaterEqual(trouves, 3, f"seulement {trouves} trouvés") self.assertGreaterEqual(trouves, 3, f"seulement {trouves} trouvés")
def test_les_commandes_de_lhote_sont_bien_lues(self):
"""Même garde pour les commandes composées par l'hôte : rtk,
starship et l'agent sont trois tubes par bloc d'outils d'assistance.
Si ce compte tombe, c'est la construction qui ne voit plus rien."""
for agent in dev_tools.AGENTS:
with self.subTest(agent=agent):
cmd = TODO.__new__(TODO)._qemu_aidev_remote_cmd(agent)
self.assertEqual(3, len(TUBE.findall(cmd)))
class TestLeReplisDePyenvSeDeclenche(unittest.TestCase):
"""lib_python_provider.sh pose pyenv par son installateur amont.
Un téléchargement raté doit s'arrêter LÀ, en le disant : sinon la suite
part sur « pyenv: command not found » puis sur un échec de compilation,
deux messages qui accusent la mauvaise étape.
Tout ce que la fonction appelle est remplacé dans un PATH qui ne porte
rien d'autre : curl, pyenv, git, gcc sont des faux. Aucun réseau, aucune
compilation, rien hors du répertoire temporaire du test.
"""
LIB = RACINE / "script" / "install" / "lib_python_provider.sh"
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.dir = Path(self.tmp.name)
self.faux = self.dir / "bin"
self.faux.mkdir()
for outil in ("bash", "mktemp", "rm", "yes", "cat", "touch"):
os.symlink(shutil.which(outil), self.faux / outil)
self.trace = self.dir / "pyenv.trace"
for nom in ("pyenv", "git", "gcc"):
self._faux(nom, f'echo "{nom} $*" >> "{self.trace}"\nexit 1\n')
def _faux(self, nom, corps):
chemin = self.faux / nom
chemin.write_text("#!/bin/bash\n" + corps, encoding="utf-8")
chemin.chmod(0o755)
def _installer(self):
env = {
"PATH": str(self.faux),
"HOME": str(self.dir),
"TMPDIR": str(self.dir),
"PYENV_ROOT": str(self.dir / "pyenv"),
}
return subprocess.run(
[
str(self.faux / "bash"),
"-c",
f'source "{self.LIB}"; el_pyenv_install 3.12.10',
],
env=env,
capture_output=True,
text=True,
timeout=60,
)
def test_un_telechargement_rate_sarrete_et_le_dit(self):
self._faux(
"curl",
'echo "curl: (22) The requested URL returned error: 504" >&2\n'
"exit 22\n",
)
fini = self._installer()
self.assertNotEqual(0, fini.returncode)
self.assertIn("installateur pyenv impossible", fini.stderr)
self.assertFalse(
self.trace.exists(),
"la suite a tourné sans pyenv : "
+ (self.trace.read_text() if self.trace.exists() else ""),
)
def test_un_installateur_qui_echoue_sarrete_et_le_dit(self):
"""Le fichier téléchargé est bien celui qui s'exécute, et son échec
arrête la fonction avant toute compilation."""
marque = self.dir / "installateur.a.tourne"
self._faux(
"curl",
# curl … -o FICHIER URL : on écrit l'installateur dans FICHIER.
'while [ "$#" -gt 0 ]; do [ "$1" = -o ] && dest="$2"; shift;'
" done\n"
f"printf 'touch \"%s\"\\nexit 3\\n' '{marque}' > \"$dest\"\n",
)
fini = self._installer()
self.assertNotEqual(0, fini.returncode)
self.assertTrue(marque.exists(), fini.stderr)
self.assertIn("installateur de pyenv a echoue", fini.stderr)
self.assertFalse(self.trace.exists())
class TestLeReplisDeMiseSeDeclenche(unittest.TestCase):
"""La pose de mise dans une VM, composée par l'hôte.
Deux échecs, deux messages, et aucun ne fait tomber « set -e » : pyenv
prend le relais. La commande tourne pour de vrai dans un PATH où curl et
sudo sont faux — et où mise n'est PAS, pour que la pose soit tentée même
sur un hôte qui l'a.
"""
def setUp(self):
from unittest import mock
with mock.patch("script.todo.qemu_install.t", lambda k: k):
self.cmd = TODO.__new__(TODO)._qemu_mise_remote_cmd("mise")
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.dir = Path(self.tmp.name)
self.faux = self.dir / "bin"
self.faux.mkdir()
for outil in ("sh", "bash", "env", "mktemp", "rm", "cat", "touch"):
os.symlink(shutil.which(outil), self.faux / outil)
self.marque = self.dir / "installateur.a.tourne"
self.trace = self.dir / "sudo.trace"
self._faux(
"curl",
'if [ -n "$ECHEC" ]; then exit 22; fi\n'
'while [ "$#" -gt 0 ]; do [ "$1" = -o ] && dest="$2"; shift;'
" done\n"
f"printf 'touch \"%s\"\\nexit ${{RC_INSTALL:-0}}\\n' "
f"'{self.marque}' > \"$dest\"\n",
)
# Trace, puis exécute SANS privilège ; « env » lit les
# « VAR=valeur » de tête comme le fait sudo.
self._faux("sudo", f'echo "$*" >> "{self.trace}"\nexec env "$@"\n')
def _faux(self, nom, corps):
chemin = self.faux / nom
chemin.write_text("#!/bin/sh\n" + corps, encoding="utf-8")
chemin.chmod(0o755)
def _lancer(self, **env_en_plus):
env = {
"PATH": str(self.faux),
"HOME": str(self.dir),
"TMPDIR": str(self.dir),
**env_en_plus,
}
script = (
"set -e\n" + self.cmd + '\necho "PROVIDER=$EL_PYTHON_PROVIDER"'
)
return subprocess.run(
[str(self.faux / "bash"), "-c", script],
env=env,
capture_output=True,
text=True,
timeout=60,
)
def test_un_telechargement_rate_le_dit_sans_rien_lancer(self):
fini = self._lancer(ECHEC="1")
self.assertEqual(0, fini.returncode, fini.stderr)
self.assertIn("⚠ mise download impossible", fini.stdout)
self.assertIn("PROVIDER=auto", fini.stdout)
self.assertFalse(self.trace.exists(), "sudo lancé sans installateur")
def test_un_installateur_qui_echoue_le_dit(self):
fini = self._lancer(RC_INSTALL="5")
self.assertEqual(0, fini.returncode, fini.stderr)
self.assertTrue(self.marque.exists())
self.assertIn("⚠ mise installer failed", fini.stdout)
self.assertIn("PROVIDER=auto", fini.stdout)
self.assertIn(
"MISE_INSTALL_PATH=/usr/local/bin/mise sh", self.trace.read_text()
)
def test_une_pose_reussie_se_tait(self):
fini = self._lancer()
self.assertEqual(0, fini.returncode, fini.stderr)
self.assertTrue(self.marque.exists())
self.assertNotIn("⚠", fini.stdout)
self.assertIn("PROVIDER=auto", fini.stdout)
def test_sans_mktemp_rien_nest_telecharge_ni_lance(self):
"""Aucun nom de repli ne remplace mktemp : root exécuterait ce qu'un
autre compte aurait pu déposer d'avance sous ce nom."""
# Le lien vers le vrai mktemp est retiré AVANT d'écrire le faux :
# écrire à travers le lien viserait le binaire de l'hôte.
(self.faux / "mktemp").unlink()
self._faux("mktemp", "exit 1\n")
fini = self._lancer()
self.assertEqual(0, fini.returncode, fini.stderr)
self.assertIn("⚠ mise download impossible", fini.stdout)
self.assertIn("PROVIDER=auto", fini.stdout)
self.assertFalse(self.marque.exists(), "installateur lancé")
self.assertFalse(self.trace.exists(), "sudo lancé sans installateur")
def test_aucun_chemin_fixe_dans_tmp(self):
"""Le fichier que root exécute ne vient que de mktemp."""
self.assertNotIn("/tmp/", self.cmd)
self.assertIn("f=$(mktemp 2>/dev/null)", self.cmd)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()

270
test/test_qemu_ai_tools.py
View file

@ -27,6 +27,7 @@ Ce que ces tests gardent :
import importlib.util import importlib.util
import shlex import shlex
import shutil
import subprocess import subprocess
import sys import sys
import unittest import unittest
@ -133,14 +134,21 @@ class LaCommandeDistante(unittest.TestCase):
rallonge son ~/.bashrc — ou son historique — d'une ligne identique. rallonge son ~/.bashrc — ou son historique — d'une ligne identique.
Aucun ajout n'échappe à la règle : on compte les « >> » et non les Aucun ajout n'échappe à la règle : on compte les « >> » et non les
greps, pour qu'une ligne ajoutée sans garde fasse tomber le test.""" greps, pour qu'une ligne ajoutée sans garde fasse tomber le test.
Chaque « >> » doit clore l'idiome ENTIER — grep, puis « || echo » de
la ligne citée —, lu d'un seul motif : la ligne citée peut porter
elle-même des « ; », et un découpage sur « ; » la couperait."""
import re
cite = r"(?:'(?:[^']|'\"'\"')*'|[^\s']+)"
garde = re.compile(
rf"grep -qF {cite} \S+ 2>/dev/null \|\| echo {cite} >> \S+"
)
for agent in ("claude", "opencode"): for agent in ("claude", "opencode"):
cmd = self._cmd(agent) cmd = self._cmd(agent)
with self.subTest(agent=agent): with self.subTest(agent=agent):
self.assertEqual(cmd.count(">> "), cmd.count("grep -qF")) self.assertEqual(cmd.count(">> "), cmd.count("grep -qF"))
for morceau in cmd.split("; "): self.assertEqual(cmd.count(">> "), len(garde.findall(cmd)))
if ">> " in morceau:
self.assertIn("grep -qF", morceau)
def test_the_local_bin_is_on_the_path_for_every_agent(self): def test_the_local_bin_is_on_the_path_for_every_agent(self):
"""rtk se pose dans ~/.local/bin. La ligne de l'agent ne couvre ce """rtk se pose dans ~/.local/bin. La ligne de l'agent ne couvre ce
@ -441,5 +449,259 @@ class LaSpec(unittest.TestCase):
self.assertEqual("qui@exemple.invalid", spec["git_email"]) self.assertEqual("qui@exemple.invalid", spec["git_email"])
# Outils réels prêtés à la commande distante dans un bac à sable : aucun ne
# touche au réseau, aux paquets ni aux droits.
_OUTILS_INOFFENSIFS = (
"sh",
"bash",
"env",
"timeout",
"grep",
"chmod",
"head",
"mktemp",
"rm",
"cat",
"mkdir",
"touch",
)
def _bac_a_sable(test, faux, outils=_OUTILS_INOFFENSIFS):
"""Un PATH qui ne porte QUE des faux et des outils inoffensifs.
`faux` : nom -> corps d'un script sh posé dans ce PATH. curl, sudo et
les gestionnaires de paquets n'y sont que FAUX, ou pas du tout : la
commande distante tourne pour de vrai sans jamais atteindre le réseau
ni l'hôte. Le HOME est jetable. Rend (env, répertoire du PATH, HOME).
"""
import os
import shutil
import tempfile
tmp = tempfile.TemporaryDirectory()
test.addCleanup(tmp.cleanup)
racine = Path(tmp.name)
faux_bin = racine / "bin"
home = racine / "home"
faux_bin.mkdir()
home.mkdir()
for outil in outils:
os.symlink(shutil.which(outil), faux_bin / outil)
for nom, corps in faux.items():
chemin = faux_bin / nom
chemin.write_text("#!/bin/sh\n" + corps, encoding="utf-8")
chemin.chmod(0o755)
env = {
"PATH": str(faux_bin),
"HOME": str(home),
"TMPDIR": str(racine),
"FAUXBIN": str(faux_bin),
}
return env, faux_bin, home
# Faux curl : à chaque installateur amont, un script qui pose un binaire
# factice là où le vrai le poserait — starship dans le PATH, comme en root.
# Avec ECHEC, il rend 22 sans rien écrire, comme sur un 504.
_FAUX_CURL = (
'if [ -n "$ECHEC" ]; then echo "curl: (22) 504" >&2; exit 22; fi\n'
'case "$*" in\n'
" *rtk*) echo 'poser rtk \"$HOME/.local/bin\"' ;;\n"
" *starship*) echo 'poser starship \"$FAUXBIN\"' ;;\n"
" *claude.ai*) echo 'poser claude \"$HOME/.local/bin\"' ;;\n"
" *opencode*) echo 'poser opencode \"$HOME/.opencode/bin\"' ;;\n"
" *) exit 22 ;;\n"
"esac\n"
)
_FAUX_POSER = (
'mkdir -p "$2"\n'
'printf \'#!/bin/sh\\necho "%s 0.0.1"\\n\' "$1" > "$2/$1"\n'
'chmod +x "$2/$1"\n'
)
# Trace, puis exécute SANS privilège ; « env » lit les « VAR=valeur » de
# tête comme le fait sudo.
_FAUX_SUDO = 'echo "$*" >> "$HOME/sudo.trace"\nexec env "$@"\n'
class LeVerdictDesOutilsAmont(unittest.TestCase):
"""Après chaque installateur amont, une ligne qui dit s'il a abouti.
Le code de sortie de la pose ne le peut pas : sans pipefail, un tube
rend le statut de l'interpréteur, 0 sur une entrée vide. La commande
distante tourne ici pour de vrai, sous « set -e », dans un bac à sable
où curl et sudo sont faux.
"""
def _lancer(self, agent, **env_en_plus):
from unittest import mock
with mock.patch("script.todo.qemu_install.t", lambda k: k):
cmd = TODO.__new__(TODO)._qemu_aidev_remote_cmd(agent)
env, faux_bin, home = _bac_a_sable(
self,
{"curl": _FAUX_CURL, "poser": _FAUX_POSER, "sudo": _FAUX_SUDO},
)
env.update(env_en_plus)
fini = subprocess.run(
[str(faux_bin / "bash"), "-c", "set -e\n" + cmd + "\necho FIN"],
env=env,
capture_output=True,
text=True,
timeout=120,
)
return fini, home
def test_without_a_download_each_tool_is_named_missing(self):
"""Et l'installation continue : aucun outil optionnel ne la fait
tomber, pas même sa ligne de verdict."""
for agent in ("claude", "opencode"):
with self.subTest(agent=agent):
fini, _home = self._lancer(agent, ECHEC="1")
self.assertEqual(0, fini.returncode, fini.stderr[-400:])
self.assertIn("FIN", fini.stdout)
for nom in ("rtk", "starship", agent):
self.assertIn(
f"⚠ {nom} not installed (see above)", fini.stdout
)
def test_a_posed_tool_gives_its_version(self):
"""opencode s'installe hors du PATH de ce shell : c'est le chemin
de repli qui le trouve."""
for agent in ("claude", "opencode"):
with self.subTest(agent=agent):
fini, _home = self._lancer(agent)
self.assertEqual(0, fini.returncode, fini.stderr[-400:])
for nom in ("rtk", "starship", agent):
self.assertIn(f"{nom}: {nom} 0.0.1", fini.stdout)
self.assertNotIn("⚠", fini.stdout)
def test_starship_is_installed_as_root(self):
"""En root, /usr/local/bin est inscriptible : l'installateur n'atteint
jamais « sudo -v », que sudo-rs refuse sans mot de passe même à un
compte NOPASSWD. L'hôte, lui, garde son installateur sans sudo."""
self.assertIn(
shlex.quote(dev_tools.STARSHIP_UPSTREAM_VM),
TODO.__new__(TODO)._qemu_aidev_remote_cmd("claude"),
)
self.assertIn("| sudo timeout -k ", dev_tools.STARSHIP_UPSTREAM_VM)
self.assertNotIn("sudo", dev_tools.STARSHIP_UPSTREAM)
_fini, home = self._lancer("claude")
self.assertIn(
f"timeout -k {dev_tools.STARSHIP_ROOT_KILL_AFTER}"
f" {dev_tools.STARSHIP_ROOT_TIMEOUT} sh -s -- -y",
(home / "sudo.trace").read_text(),
)
def test_the_root_installer_is_bounded_behind_sudo(self):
"""Un « timeout » sans privilège ne tue pas un processus root : la
borne de l'installateur doit passer DERRIÈRE sudo. « -k » envoie
SIGKILL à un installateur qui ignore SIGTERM ; borne et délai
additionnés tombent avant la borne de la pose, qui ne tient plus que
curl."""
import re
vm = dev_tools.STARSHIP_UPSTREAM_VM
dedans = re.search(r"\| sudo timeout -k (\d+) (\d+) sh -s -- -y$", vm)
self.assertIsNotNone(dedans, vm)
cmd = TODO.__new__(TODO)._qemu_aidev_remote_cmd("claude")
dehors = re.search(
r"timeout (\d+) sh -c " + re.escape(shlex.quote(vm)), cmd
)
self.assertIsNotNone(dehors, "la pose de starship n'est plus bornée")
self.assertLess(
int(dedans.group(1)) + int(dedans.group(2)), int(dehors.group(1))
)
class LeHookDuPrompt(unittest.TestCase):
"""La ligne que le fichier du shell reçoit pour starship, lue par un vrai
shell.
Le fichier est lu SEUL, sous « set -e », et rien ne suit la lecture : la
ligne est la dernière du fichier, dont le statut est donc le sien. Une
commande placée après masquerait ce statut.
"""
def _sourcer(self, faux, shell="bash"):
binaire = shutil.which(shell)
if binaire is None:
self.skipTest(f"{shell} absent de cet hôte")
env, faux_bin, home = _bac_a_sable(self, faux, outils=())
rc = home / f".{shell}rc"
rc.write_text(dev_tools.STARSHIP_LINE[shell] + "\n", encoding="utf-8")
# Ni ~/.bashrc ni ~/.zshrc lus d'office : seul `rc` est lu.
options = ["--norc"] if shell == "bash" else ["-f"]
return subprocess.run(
[binaire, *options, "-c", f'set -e; . "{rc}"'],
env=env,
capture_output=True,
text=True,
timeout=30,
)
def test_without_the_binary_it_stays_silent_and_returns_zero(self):
"""Une pose ratée, un binaire retiré : sans garde, chaque shell
ouvert écrirait « command not found ». Et la lecture rend 0 : un
script sous « set -e » qui lit ce fichier ne s'y arrête pas."""
for shell in ("bash", "zsh"):
with self.subTest(shell=shell):
fini = self._sourcer({}, shell)
self.assertEqual("", fini.stderr)
self.assertEqual(0, fini.returncode)
def test_with_the_binary_it_starts_starship(self):
for shell in ("bash", "zsh"):
with self.subTest(shell=shell):
fini = self._sourcer(
{"starship": 'echo "echo INIT-$2"\n'}, shell
)
self.assertEqual(0, fini.returncode, fini.stderr)
self.assertIn(f"INIT-{shell}", fini.stdout)
def test_the_guard_is_an_if_in_every_posix_shell(self):
"""Le même « if … fi » pour bash et zsh : un hôte sans zsh ne lit
la ligne de zsh par aucun shell, sa forme reste donc vérifiée."""
for shell in ("bash", "zsh"):
with self.subTest(shell=shell):
ligne = dev_tools.STARSHIP_LINE[shell]
self.assertTrue(ligne.startswith("if command -v starship "))
self.assertTrue(ligne.endswith("; fi"), ligne)
def test_the_dedup_pattern_survives_the_guard(self):
"""L'hôte comme la VM reconnaissent une ligne déjà écrite à
« starship init » : la garde ne doit pas le masquer."""
for shell, ligne in dev_tools.STARSHIP_LINE.items():
with self.subTest(shell=shell):
self.assertIn(f"starship init {shell}", ligne)
class UneApostropheTraduite(unittest.TestCase):
def test_the_tool_blocks_stay_valid_shell(self):
"""Les messages sont traduits, et le français est plein
d'apostrophes : une seule mal placée casse la commande distante
ENTIÈRE. On remplace la traduction elle-même — « set_lang » la
persisterait dans env_var.sh."""
from unittest import mock
# Un nombre IMPAIR d'apostrophes : entre apostrophes, un nombre pair
# se referme de lui-même, et « bash -n » ne verrait rien.
piege = "l'outil n'a pas « fini » aujourd'hui"
with mock.patch("script.todo.qemu_install.t", lambda k: piege):
todo = TODO.__new__(TODO)
blocs = {
f"aidev({agent})": todo._qemu_aidev_remote_cmd(agent)
for agent in dev_tools.AGENTS
}
blocs["mise"] = todo._qemu_mise_remote_cmd("mise")
for nom, cmd in blocs.items():
with self.subTest(bloc=nom):
self.assertIn(piege, cmd)
fini = subprocess.run(
["bash", "-n"], input=cmd, text=True, capture_output=True
)
self.assertEqual(0, fini.returncode, fini.stderr[:400])
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()

134
test/test_qemu_desktop_tools.py
View file

@ -1329,6 +1329,134 @@ class TestGnomeSiteExtensions(unittest.TestCase):
l'activation ne peut rien ecrire dans dconf.""" l'activation ne peut rien ecrire dans dconf."""
self.assertIn("dbus-run-session", self.block) self.assertIn("dbus-run-session", self.block)
def _lancer(self, **env_en_plus):
"""Le bloc, pour de vrai, sous « set -e », dans un PATH où
gnome-shell, curl, mktemp et gnome-extensions sont faux : ni
réseau, ni session, ni fichier hors du répertoire du test."""
import os
import shutil
import tempfile
with mock.patch("script.todo.qemu_install.t", lambda k: k):
bloc = TODO.__new__(TODO)._qemu_gnome_ext_remote_cmd()
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
racine = pathlib.Path(tmp.name)
self.racine = racine
faux = racine / "bin"
faux.mkdir()
for outil in ("sh", "bash", "awk", "cut", "rm"):
os.symlink(shutil.which(outil), faux / outil)
corps = {
"gnome-shell": 'echo "GNOME Shell 50.1"\n',
"mktemp": 'if [ -n "$MKTEMP_ECHEC" ]; then exit 1; fi\n'
'f="$TMPDIR/gext.$$"; : > "$f"; echo "$f"\n',
"curl": 'echo "$*" >> "$HOME/curl.trace"\n'
'if [ -n "$ECHEC" ]; then exit 22; fi\n'
'while [ "$#" -gt 0 ]; do [ "$1" = -o ] && dest="$2"; shift;'
' done\n: > "$dest"\n',
"gnome-extensions": '[ "$1" = install ] && exit "${GX_RC:-0}"\n'
"exit 0\n",
}
for nom, texte in corps.items():
chemin = faux / nom
chemin.write_text("#!/bin/sh\n" + texte, encoding="utf-8")
chemin.chmod(0o755)
env = {
"PATH": str(faux),
"HOME": str(racine),
"TMPDIR": str(racine),
**env_en_plus,
}
fini = subprocess.run(
[str(faux / "bash"), "-c", "set -e\n" + bloc + "\necho FIN"],
env=env,
capture_output=True,
text=True,
timeout=60,
)
self.assertEqual(0, fini.returncode, fini.stderr[-400:])
self.assertIn("FIN", fini.stdout)
return fini.stdout
def test_a_failed_download_does_not_blame_gnome(self):
"""Le site sert une archive même à un GNOME qu'il ne connaît pas :
un téléchargement raté ne dit rien de la version."""
sortie = self._lancer(ECHEC="1")
self.assertEqual(
len(TODO._QEMU_GNOME_EXT_UUIDS),
sortie.count("⚠ download impossible (network or cache):"),
)
self.assertNotIn("not available for this GNOME", sortie)
def test_a_failed_mktemp_skips_like_a_failed_download(self):
"""Sans nom tiré par mktemp, rien n'est téléchargé : un nom fixe dans
/var/tmp, ouvert à tous, pourrait y être posé d'avance."""
sortie = self._lancer(MKTEMP_ECHEC="1")
self.assertEqual(
len(TODO._QEMU_GNOME_EXT_UUIDS),
sortie.count("⚠ download impossible (network or cache):"),
)
self.assertNotIn("installed and enabled:", sortie)
self.assertFalse(
(self.racine / "curl.trace").exists(), "curl lancé sans fichier"
)
def test_no_fixed_path_in_the_shared_tmp(self):
"""/var/tmp n'apparaît que comme répertoire donné à mktemp, et le
bloc reste valide sous une traduction à apostrophe impaire."""
import re
piege = "l'extension n'a pas « fini » aujourd'hui"
with mock.patch("script.todo.qemu_install.t", lambda k: piege):
bloc = TODO.__new__(TODO)._qemu_gnome_ext_remote_cmd()
self.assertEqual(
{"/var/tmp"}, set(re.findall(r"/var/tmp[^\s;\"')|]*", bloc))
)
self.assertIn("mktemp -p /var/tmp gext-XXXX.zip", bloc)
res = subprocess.run(
["bash", "-n"], input=bloc, text=True, capture_output=True
)
self.assertEqual(0, res.returncode, res.stderr[:400])
def test_a_refused_install_names_the_gnome_version(self):
sortie = self._lancer(GX_RC="1")
self.assertEqual(
len(TODO._QEMU_GNOME_EXT_UUIDS),
sortie.count("not available for this GNOME, skipped:"),
)
self.assertIn("(GNOME 50)", sortie)
self.assertNotIn("download impossible", sortie)
def test_log_out_is_asked_only_when_something_was_installed(self):
"""Se reconnecter pour charger des extensions jamais posées est
une consigne qui ment."""
for echec in ({"ECHEC": "1"}, {"GX_RC": "1"}):
with self.subTest(echec=echec):
sortie = self._lancer(**echec)
self.assertNotIn("log out and back in", sortie)
sortie = self._lancer()
self.assertEqual(
len(TODO._QEMU_GNOME_EXT_UUIDS),
sortie.count("installed and enabled:"),
)
self.assertEqual(1, sortie.count("log out and back in to load them"))
def test_a_translated_apostrophe_keeps_the_block_valid(self):
"""Une apostrophe mal placée casse la commande distante ENTIÈRE. On
remplace la traduction — « set_lang » la persisterait.
Un nombre IMPAIR d'apostrophes : entre apostrophes, un nombre pair
se referme de lui-même, et « bash -n » ne verrait rien."""
piege = "l'extension n'a pas « fini » aujourd'hui"
with mock.patch("script.todo.qemu_install.t", lambda k: piege):
bloc = TODO.__new__(TODO)._qemu_gnome_ext_remote_cmd()
self.assertIn(piege, bloc)
res = subprocess.run(
["bash", "-n"], input=bloc, text=True, capture_output=True
)
self.assertEqual(0, res.returncode, res.stderr[:400])
class TestLeServiceDeLAgentInvite(unittest.TestCase): class TestLeServiceDeLAgentInvite(unittest.TestCase):
"""Le service détaché qui pose qemu-guest-agent, lancé par cloud-init. """Le service détaché qui pose qemu-guest-agent, lancé par cloud-init.
@ -1486,10 +1614,14 @@ class TestLeVerrouAptNeCoutePasDesMinutes(unittest.TestCase):
PERSISTE dans env_var.sh, et un test qui la déplace fait échouer PERSISTE dans env_var.sh, et un test qui la déplace fait échouer
tout ce qui suit. On remplace la traduction elle-même, le temps du tout ce qui suit. On remplace la traduction elle-même, le temps du
contrôle, par une chaîne qui porte le caractère dangereux. contrôle, par une chaîne qui porte le caractère dangereux.
Le nombre d'apostrophes est IMPAIR : avec deux, un message emballé
entre guillemets simples se referme sur lui-même et le shell reste
valide, si bien que le contrôle ne verrait rien.
""" """
import tempfile import tempfile
piege = "l'agent n'a pas fini « attendre »" piege = "l'agent n'a pas fini aujourd'hui « attendre »"
with mock.patch("script.todo.qemu_install.t", lambda k: piege): with mock.patch("script.todo.qemu_install.t", lambda k: piege):
todo = TODO.__new__(TODO) todo = TODO.__new__(TODO)
cmd = ( cmd = (

22
test/test_todo_shell_tools.py
View file

@ -25,9 +25,17 @@ from unittest.mock import MagicMock, patch
from script.todo.todo import TODO from script.todo.todo import TODO
LIGNE_BASH = 'eval "$(starship init bash)"' LIGNE_BASH = (
LIGNE_ZSH = 'eval "$(starship init zsh)"' "if command -v starship >/dev/null 2>&1; then"
' eval "$(starship init bash)"; fi'
)
LIGNE_ZSH = (
"if command -v starship >/dev/null 2>&1; then"
' eval "$(starship init zsh)"; fi'
)
LIGNE_FISH = "starship init fish | source" LIGNE_FISH = "starship init fish | source"
# La ligne telle qu'elle s'écrivait avant sa garde : des fichiers la portent.
LIGNE_BASH_SANS_GARDE = 'eval "$(starship init bash)"'
def refuse_input(*args, **kwargs): def refuse_input(*args, **kwargs):
@ -214,6 +222,16 @@ class TestHookStarship(ShellFixture):
) )
self.assertIn("✅", sortie) self.assertIn("✅", sortie)
def test_a_line_written_before_the_guard_is_still_recognised(self):
"""Un fichier écrit avant la garde porte la ligne nue : « starship
init » la reconnaît, et une seconde ligne n'est pas ajoutée."""
self.cree("bash", contenu=f"{LIGNE_BASH_SANS_GARDE}\n")
self.hook()
self.assertEqual(
self.fichiers["bash"].read_text(encoding="utf-8"),
f"{LIGNE_BASH_SANS_GARDE}\n",
)
def test_no_file_at_all_creates_the_bash_one(self): def test_no_file_at_all_creates_the_bash_one(self):
self.hook() self.hook()
self.assertEqual( self.assertEqual(