[FIX] LongTest : sh au lieu de bash, et --detruire trop large

Attaqué par trois lentilles sur le code écrit, avant de le lancer pour de
vrai. Deux fautes valaient à elles seules l'exercice.

Il n'aurait JAMAIS fonctionné. L'installeur était lancé par « sh », or il
porte « set -euo pipefail » et un shebang bash : sur Debian /bin/sh est dash,
qui répond « set: Illegal option -o pipefail » et sort à la PREMIÈRE ligne.
Chaque étage aurait échoué sur l'installation, à tous les coups.

Et « --detruire » pouvait emporter une machine étrangère. Il prenait toute
entrée ssh dont le nom CONTENAIT « deep-pve », puis sur son rebond détruisait
toute VM dont le nom contenait « deep-pve » — une « deep-pve-lab » de
production tombait dedans, et « --purge » emporte les disques. Son tri « du
plus profond au plus haut » comptait les « + » de l'alias, or alias_etage
remplace le « + » du parent par un « - » : chaque alias en portait exactement
UN, le tri ne triait rien, et la destruction partait du plus HAUT — le disque
du parent emportait ses enfants sans qu'on les ait nommés. Il ignorait
« --dry-run », ne lisait aucun code de retour, concluait « ✓ défait », et le
menu le lançait d'une touche.

Il ne détruit plus que ce que le RAPPORT nomme : un couple (parent, VMID) par
étage, du plus profond d'après le niveau lu, égalité stricte du nom, arrêt
CONSTATÉ avant destruction, codes de retour lus, et une confirmation par
« OUI » après la liste.

Six autres constats, tous réels. Le redémarrage se prouve par btime et non par
le seul noyau — rejoué sur un étage déjà installé, on validait un redémarrage
qui n'avait pas eu lieu, exactement le piège corrigé la semaine dernière dans
le suivi. La sonde de disponibilité ne demande plus sudo, sinon un sudo lent
se lisait « jamais joignable en ssh ». Les délais suivent la profondeur : le
script existe pour mesurer un ralentissement de 36x, et un plafond fixe
déclarait échouée une installation qui avançait. L'adresse fixe est contrôlée
AVANT de télécharger une image et de démarrer une VM. Le DNS de l'hôte suit la
spec, sinon apt meurt sans rien expliquer. Et l'essai à blanc ne prétend plus
avoir atteint quoi que ce soit — son rapport était indiscernable d'une
réussite, JSON compris.

L'algorithme aussi : profondeur 0 rendait un plan d'UN étage, donc
« --depth 0 » créait une VM ; et sur un hôte de quatre cœurs le premier étage
recevait UN vCPU quand son invité en recevait deux — un parent plus étroit que
son enfant.

Les tests mordent, prouvé par mutation : remplacer le calcul du premier étage
par la valeur imbriquée les laissait verts.

--- EN ---

Attacked by three lenses on the written code, before running it for real. Two
faults alone justified the exercise.

It would NEVER have worked. The installer was run by "sh", yet it carries "set
-euo pipefail" and a bash shebang: on Debian /bin/sh is dash, which answers
"set: Illegal option -o pipefail" and exits on the FIRST line. Every level
would have failed at install, every time.

And "--detruire" could take a stranger's machine. It took every ssh entry
whose name CONTAINED "deep-pve", then on its jump host destroyed every VM
whose name contained "deep-pve" — a production "deep-pve-lab" fell in, and
"--purge" takes the disks. Its "deepest first" sort counted the "+" in the
alias, yet alias_etage replaces the parent's "+" with a "-": every alias had
exactly ONE, the sort sorted nothing, and destruction started from the TOP —
the parent's disk took its children with it, unnamed. It ignored "--dry-run",
read no return code, concluded "✓ done", and the menu fired it on one key.

It now destroys only what the REPORT names: a (parent, VMID) pair per level,
deepest first by the recorded level, strict name equality, shutdown VERIFIED
before destruction, return codes read, and a "OUI" confirmation after the
list.

Six more findings, all real. The reboot is proven by btime, not by the kernel
alone — replayed on an already-installed level, we validated a reboot that
never happened, exactly the trap fixed last week in the monitor. The liveness
probe no longer asks for sudo, or a slow sudo read as "never reachable by
ssh". Timeouts follow the depth: the script exists to measure a 36x slowdown,
and a fixed ceiling declared failed an install that was progressing. The
static address is checked BEFORE downloading an image and starting a VM. The
host's DNS follows the spec, or apt dies explaining nothing. And the dry run
no longer claims to have reached anything — its report was indistinguishable
from a success, JSON included.

The algorithm too: depth 0 returned a ONE-level plan, so "--depth 0" created a
VM; and on a four-core host the first level got ONE vCPU while its guest got
two — a parent narrower than its child.

The tests bite, proven by mutation: replacing the first level's computation
with the nested value left them green.

Assisted-by: Claude Opus 5
(cherry picked from commit 64b8e5063bd7f420cdeb27b88f94043190b5ecd4)
This commit is contained in:
Mathieu Benoit 2026-08-26 07:00:30 -04:00
parent 7199a7cbb2
commit 9bce1a8503
6 changed files with 551 additions and 102 deletions

View file

@ -135,10 +135,28 @@ class Descente:
self.journal = journal self.journal = journal
self.dry_run = dry_run self.dry_run = dry_run
self.etages = [] self.etages = []
self.interrompu = False
self.niveau_courant = 1
def dire(self, msg): def dire(self, msg):
dire(msg, self.journal) dire(msg, self.journal)
def delai(self, etape):
"""Le délai de cette étape, à l'étage courant.
Constant, il contredisait la raison d'être du script : au quatrième
étage un invité tournait 36 fois moins vite. Une installation de dix
minutes au premier étage en demande des heures au quatrième, et le
plafond fixe la déclarait échouée — en concluant à un mur
d'imbrication là où il n'y avait qu'un délai trop court.
Le facteur est CARRÉ et borné : chaque étage ajoute une couche
d'hyperviseur à traverser, mais un facteur illimité rendrait un
échec réel indiscernable d'une attente sans fin.
"""
facteur = min(max(1, self.niveau_courant), 5) ** 2
return DELAIS[etape] * facteur
# ---------------------------------------------------------------- # # ---------------------------------------------------------------- #
# Parler aux machines # Parler aux machines
# ---------------------------------------------------------------- # # ---------------------------------------------------------------- #
@ -171,13 +189,26 @@ class Descente:
if self.dry_run: if self.dry_run:
return 0 return 0
debut = time.time() debut = time.time()
# SANS privilège : wrap_privilege transformerait « true » en
# « sudo sh -c true », et un sudo qui réclame un mot de passe — le
# temps que cloud-init écrive /etc/sudoers.d — se lisait « jamais
# joignable en ssh ». Le transport marchait ; c'est le diagnostic qui
# était faux.
sonde = dict(hote, sudo="")
while time.time() - debut < delai: while time.time() - debut < delai:
code, _o = pve.run(hote, "true", 30) code, _o = pve.run(sonde, "true", 60)
if code == 0: if code == 0:
return int(time.time() - debut) return int(time.time() - debut)
time.sleep(15) time.sleep(15)
return None return None
def sudo_pret(self, hote):
"""sudo répond-il sans mot de passe ? Nommé à part de ssh."""
if self.dry_run:
return True
code, _o = pve.run(hote, "true", 60)
return code == 0
# ---------------------------------------------------------------- # # ---------------------------------------------------------------- #
# Les six étapes, les mêmes à chaque étage # Les six étapes, les mêmes à chaque étage
# ---------------------------------------------------------------- # # ---------------------------------------------------------------- #
@ -187,7 +218,7 @@ class Descente:
distant = "/tmp/install_proxmox.sh" distant = "/tmp/install_proxmox.sh"
if self.dry_run: if self.dry_run:
print(f" scp {local} <hôte>:{distant}") print(f" scp {local} <hôte>:{distant}")
print(f" sh {distant}") print(f" bash {distant}")
return True return True
argv = pve.ssh_argv(hote, "")[:-1] # les options, sans la commande argv = pve.ssh_argv(hote, "")[:-1] # les options, sans la commande
cible = argv[-1] cible = argv[-1]
@ -201,10 +232,14 @@ class Descente:
if res.returncode: if res.returncode:
self.dire(f" ✗ scp : {res.stderr.strip()[:200]}") self.dire(f" ✗ scp : {res.stderr.strip()[:200]}")
return False return False
# « bash » et non « sh » : le script porte « set -euo pipefail » et un
# shebang bash. Sur Debian /bin/sh est dash, qui répond « set: Illegal
# option -o pipefail » et sort à la PREMIÈRE ligne — vérifié. Chaque
# étage aurait échoué sur l'installation, à tous les coups.
code, _o = self.executer( code, _o = self.executer(
dict(hote, sudo=""), dict(hote, sudo=""),
f"sh {distant}", f"bash {distant}",
DELAIS["install"], self.delai("install"),
"install_proxmox.sh", "install_proxmox.sh",
montrer=True, montrer=True,
) )
@ -219,19 +254,36 @@ class Descente:
dépouillé de tout netfilter : ni pont NAT, ni invité. dépouillé de tout netfilter : ni pont NAT, ni invité.
""" """
if self.dry_run: if self.dry_run:
print(" reboot puis attente de *-pve dans uname -r") print(" reboot, puis btime changé ET *-pve dans uname -r")
return True return True
# L'instant de démarrage AVANT : le noyau seul ne prouve rien. Rejoué
# sur un étage déjà installé, le script est idempotent et ne redémarre
# pas ; vingt secondes après l'ordre, sshd répond encore et la machine
# tourne DÉJÀ sur -pve. On validait donc un redémarrage qui n'avait pas
# eu lieu, et l'étape suivante tombait sur une machine en train de
# s'éteindre — avec un diagnostic sans rapport. Même piège que celui
# corrigé dans le suivi d'installation, refait ici.
_c, out = pve.run(dict(hote, sudo=""), "stat -c %Y /proc/1", 60)
avant = pve.strip_ssh_noise(out).strip()
pve.run(hote, "systemctl reboot", 60) pve.run(hote, "systemctl reboot", 60)
debut = time.time() debut = time.time()
while time.time() - debut < DELAIS["reboot"]: while time.time() - debut < self.delai("reboot"):
time.sleep(20) time.sleep(20)
code, out = pve.run(dict(hote, sudo=""), "uname -r", 30) code, out = pve.run(
noyau = pve.strip_ssh_noise(out).strip() dict(hote, sudo=""), "uname -r; stat -c %Y /proc/1", 60
if code == 0 and "-pve" in noyau: )
self.dire( lignes = pve.strip_ssh_noise(out).strip().splitlines()
f" noyau {noyau} après {int(time.time() - debut)} s" if code or len(lignes) < 2:
) continue
return True noyau, apres = lignes[0].strip(), lignes[-1].strip()
if "-pve" not in noyau:
continue
if avant and apres == avant:
continue # elle n'a pas encore redémarré
self.dire(
f" noyau {noyau} après {int(time.time() - debut)} s"
)
return True
self.dire(" ✗ pas revenue sur un noyau -pve") self.dire(" ✗ pas revenue sur un noyau -pve")
return False return False
@ -252,7 +304,7 @@ class Descente:
(pve.hosts_repair_cmd(ip), "/etc/hosts"), (pve.hosts_repair_cmd(ip), "/etc/hosts"),
): ):
code, sortie = self.executer( code, sortie = self.executer(
hote, cmd, DELAIS["reparation"], etiquette hote, cmd, self.delai("reparation"), etiquette
) )
if code or "-KO" in pve.strip_ssh_noise(sortie): if code or "-KO" in pve.strip_ssh_noise(sortie):
self.dire(f" ✗ {etiquette}") self.dire(f" ✗ {etiquette}")
@ -262,7 +314,7 @@ class Descente:
hote, pve.pve_unit_cmd(unite, remonte=True), 300, unite hote, pve.pve_unit_cmd(unite, remonte=True), 300, unite
) )
_c, out = self.executer( _c, out = self.executer(
hote, pve.mount_wait_cmd(), DELAIS["reparation"], "montage" hote, pve.mount_wait_cmd(), self.delai("reparation"), "montage"
) )
vu = pve.parse_mount_wait(out) vu = pve.parse_mount_wait(out)
self.dire(f" /etc/pve : {vu['verdict']}") self.dire(f" /etc/pve : {vu['verdict']}")
@ -281,12 +333,15 @@ class Descente:
self.dire(" ✗ aucun stockage sur le parent") self.dire(" ✗ aucun stockage sur le parent")
return None return None
_c, out = self.executer( _c, out = self.executer(
parent, "ip -o link show type bridge", DELAIS["controle"], "ponts" parent,
"ip -o link show type bridge",
self.delai("controle"),
"ponts",
) )
ponts = pve.parse_bridges(out) ponts = pve.parse_bridges(out)
if not ponts: if not ponts:
_c, nets = self.executer( _c, nets = self.executer(
parent, pve.USED_NETS_CMD, DELAIS["controle"], "réseaux" parent, pve.USED_NETS_CMD, self.delai("controle"), "réseaux"
) )
cidr = pve.pick_internal_cidr(nets) or pve.INTERNAL_CIDR cidr = pve.pick_internal_cidr(nets) or pve.INTERNAL_CIDR
_c, rt = self.executer( _c, rt = self.executer(
@ -311,10 +366,18 @@ class Descente:
DELAIS["controle"], DELAIS["controle"],
"interfaces", "interfaces",
) )
# Le DNS de l'hôte. « --ipconfig0 » ne le porte PAS : une VM en
# adresse fixe route mais ne résout rien, et install_proxmox.sh meurt
# sur « apt update » sans que rien ne l'explique. Le rapport imputerait
# à l'installation ce qui est un défaut de résolveur.
_c, resolv = self.executer(
parent, pve.RESOLV_CMD, self.delai("controle"), "resolv"
)
return ( return (
stockage or "local", stockage or "local",
ponts[0], ponts[0],
pve.parse_bridge_config(cfg).get(ponts[0], {}), pve.parse_bridge_config(cfg).get(ponts[0], {}),
pve.parse_nameservers(resolv),
) )
def creer_etage1(self, res): def creer_etage1(self, res):
@ -348,18 +411,34 @@ class Descente:
def creer_enfant(self, parent, niveau, res, prepare): def creer_enfant(self, parent, niveau, res, prepare):
"""« qm create » sur le parent. Rend (vmid, adresse) ou (None, None).""" """« qm create » sur le parent. Rend (vmid, adresse) ou (None, None)."""
stockage, pont, info_pont = prepare stockage, pont, info_pont, dns = prepare
mod = module_qemu() mod = module_qemu()
version = mod.DISTROS[DISTRO][1] version = mod.DISTROS[DISTRO][1]
code_img = mod.DISTROS[DISTRO][0][version][0] code_img = mod.DISTROS[DISTRO][0][version][0]
url = mod.image_url(DISTRO, code_img, "amd64", version) url = mod.image_url(DISTRO, code_img, "amd64", version)
image = mod.default_image_name(DISTRO, code_img, "amd64", version) image = mod.default_image_name(DISTRO, code_img, "amd64", version)
_c, out = self.executer( _c, out = self.executer(
parent, "qm list", DELAIS["controle"], "qm list" parent, "qm list", self.delai("controle"), "qm list"
) )
vmid = pve.next_vmid(pve.parse_qm_list(out)) vmid = pve.next_vmid(pve.parse_qm_list(out))
ipconfig = pve.ipconfig_for(info_pont, vmid) ipconfig = pve.ipconfig_for(info_pont, vmid)
adresse = pve.ip_from_ipconfig(ipconfig) adresse = pve.ip_from_ipconfig(ipconfig)
# AVANT de télécharger l'image et de démarrer quoi que ce soit : sur un
# pont relié au LAN, ipconfig_for rend « ip=dhcp » et l'adresse est
# vide. Le contrôle venait après la création : on laissait une VM
# allumée, un disque alloué, et une machine que --detruire ne
# connaissait pas.
if not adresse and self.dry_run:
# En essai à blanc on n'a rien lu du parent : conclure « pas de
# pont interne » serait une affirmation tirée d'une mesure qui
# n'a pas eu lieu. On prend une adresse plausible pour dérouler
# le plan jusqu'au bout.
adresse = "10.10.10.150"
ipconfig = f"ip={adresse}/24,gw=10.10.10.1"
if not adresse:
self.dire(" ✗ pas d'adresse fixe : le parent n'a pas de")
self.dire(" pont interne, et l'enfant serait injoignable")
return None, None
spec = { spec = {
"name": nom_etage(niveau), "name": nom_etage(niveau),
"storage": stockage, "storage": stockage,
@ -371,6 +450,7 @@ class Descente:
"user": "erplibre", "user": "erplibre",
"ipconfig": ipconfig, "ipconfig": ipconfig,
"sshkey_path": "/root/.ssh/longtest.pub", "sshkey_path": "/root/.ssh/longtest.pub",
"nameservers": dns,
"start": True, "start": True,
} }
# La clé publique doit être un FICHIER sur le parent : « --sshkeys » # La clé publique doit être un FICHIER sur le parent : « --sshkeys »
@ -390,7 +470,7 @@ class Descente:
vmid, spec vmid, spec
): ):
code, _o = self.executer( code, _o = self.executer(
parent, cmd, DELAIS["creation"], "qm create" parent, cmd, self.delai("creation"), "qm create"
) )
if code and not self.dry_run: if code and not self.dry_run:
return None, None return None, None
@ -404,6 +484,7 @@ class Descente:
parent_alias = "" parent_alias = ""
for res in self.plan["niveaux"]: for res in self.plan["niveaux"]:
niveau = res["niveau"] niveau = res["niveau"]
self.niveau_courant = niveau
debut = time.time() debut = time.time()
etage = { etage = {
"niveau": niveau, "niveau": niveau,
@ -419,6 +500,7 @@ class Descente:
nom = self.creer_etage1(res) nom = self.creer_etage1(res)
if not nom: if not nom:
self.etages.append(etage) self.etages.append(etage)
self.interrompu = True
break break
alias = nom alias = nom
else: else:
@ -426,32 +508,31 @@ class Descente:
if not prepare: if not prepare:
etage["etape"] = "parent" etage["etape"] = "parent"
self.etages.append(etage) self.etages.append(etage)
self.interrompu = True
break break
vmid, adresse = self.creer_enfant(parent, niveau, res, prepare) vmid, adresse = self.creer_enfant(parent, niveau, res, prepare)
if vmid is None: if vmid is None:
self.etages.append(etage) self.etages.append(etage)
self.interrompu = True
break break
etage["vmid"] = vmid etage["vmid"] = vmid
# Le parent est noté AVANT tout autre contrôle : c'est le seul
# enregistrement de ce qu'on vient de créer, et --detruire s'en
# sert. Sans lui, une VM abandonnée juste après « qm create »
# n'était nommée nulle part.
etage["parent_alias"] = parent_alias
alias = alias_etage(niveau, parent_alias) alias = alias_etage(niveau, parent_alias)
if not self.dry_run: if not self.dry_run:
if not adresse:
# Sur un pont interne l'adresse est FIXE et dérivée du
# VMID. Vide, c'est que le parent n'a pas de pont
# interne — écrire un alias sans HostName donnerait
# une entrée qui ne mène nulle part.
self.dire(" ✗ pas d'adresse fixe pour l'enfant")
etage["etape"] = "adresse"
self.etages.append(etage)
break
self.ecrire_alias(alias, adresse, parent_alias) self.ecrire_alias(alias, adresse, parent_alias)
cible = {"target": alias, "sudo": "sudo ", "jump": ""} cible = {"target": alias, "sudo": "sudo ", "jump": ""}
etage["alias"] = alias etage["alias"] = alias
etage["etape"] = "ssh" etage["etape"] = "ssh"
attente = self.attendre_ssh(cible, DELAIS["ssh"]) attente = self.attendre_ssh(cible, self.delai("ssh"))
if attente is None: if attente is None:
self.dire(" ✗ jamais joignable en ssh") self.dire(" ✗ jamais joignable en ssh")
self.etages.append(etage) self.etages.append(etage)
self.interrompu = True
break break
etage["ssh_secondes"] = attente etage["ssh_secondes"] = attente
self.dire(f" ssh après {attente} s") self.dire(f" ssh après {attente} s")
@ -466,13 +547,16 @@ class Descente:
self.etages.append(etage) self.etages.append(etage)
return self.rapport(interrompu=True) return self.rapport(interrompu=True)
etage["etape"] = "termine" # En dry-run, aucune étape n'a été mesurée : les marquer
etage["ok"] = True # « atteintes » produisait un rapport indiscernable d'une vraie
# réussite, JSON compris, et un code de sortie 0.
etage["etape"] = "plan" if self.dry_run else "termine"
etage["ok"] = not self.dry_run
etage["secondes"] = int(time.time() - debut) etage["secondes"] = int(time.time() - debut)
self.etages.append(etage) self.etages.append(etage)
self.dire(f" ✓ étage {niveau} en {etage['secondes']} s") self.dire(f" ✓ étage {niveau} en {etage['secondes']} s")
parent, parent_alias = cible, alias parent, parent_alias = cible, alias
return self.rapport() return self.rapport(interrompu=self.interrompu)
def ecrire_alias(self, alias, adresse, parent_alias): def ecrire_alias(self, alias, adresse, parent_alias):
"""Une entrée ~/.ssh/config pour joindre l'enfant à travers le parent.""" """Une entrée ~/.ssh/config pour joindre l'enfant à travers le parent."""
@ -491,69 +575,236 @@ class Descente:
def rapport(self, interrompu=False): def rapport(self, interrompu=False):
atteint = sum(1 for e in self.etages if e["ok"]) atteint = sum(1 for e in self.etages if e["ok"])
print("") print("")
self.dire( if self.dry_run:
f" profondeur atteinte : {atteint} / {self.plan['demandee']}" self.dire(
) f" plan annoncé sur {len(self.etages)} étage(s) —"
" rien n'a été créé"
)
else:
self.dire(
f" profondeur atteinte : {atteint}"
f" / {self.plan['demandee']}"
)
for e in self.etages: for e in self.etages:
marque = "✓" if e["ok"] else "✗" if self.dry_run:
detail = f"{e.get('secondes', '—')} s" if e["ok"] else e["etape"] marque, detail = "·", "plan"
else:
marque = "✓" if e["ok"] else "✗"
detail = (
f"{e.get('secondes', '—')} s" if e["ok"] else e["etape"]
)
self.dire(f" {marque} étage {e['niveau']:2d} {detail}") self.dire(f" {marque} étage {e['niveau']:2d} {detail}")
return { return {
"demandee": self.plan["demandee"], "demandee": self.plan["demandee"],
"atteignable": self.plan["atteignable"], "atteignable": self.plan["atteignable"],
"atteinte": atteint, "atteinte": atteint,
"interrompu": interrompu, "interrompu": interrompu,
# Sans ce champ, un rapport d'essai à blanc se lisait comme une
# descente réussie — et « --detruire » s'en servait.
"dry_run": self.dry_run,
"etages": self.etages, "etages": self.etages,
} }
def detruire(journal=None): def dernier_rapport():
"""Défait ce que la descente a posé, du plus profond au plus haut. """Le rapport JSON le plus récent, ou {}.
Du plus profond : détruire un parent d'abord emporterait ses enfants sans C'est le SEUL enregistrement de ce que la descente a créé : un couple
qu'on ait pu les nommer, et laisserait des entrées ssh vers rien. (alias du parent, VMID) par étage. Détruire d'après lui, et non d'après
les noms, est toute la différence entre défaire son propre travail et
effacer une machine qui se trouve porter un nom voisin.
""" """
from script.todo.todo import TODO dossier = os.path.expanduser("~/.erplibre/longtest")
try:
fichiers = sorted(
f for f in os.listdir(dossier) if f.endswith(".json")
)
except OSError:
return {}
for nom in reversed(fichiers):
try:
with open(os.path.join(dossier, nom), encoding="utf-8") as fh:
rapport = json.load(fh)
except (OSError, ValueError):
continue
if rapport.get("dry_run"):
continue # un plan n'a rien créé
rapport["fichier"] = os.path.join(dossier, nom)
return rapport
return {}
hosts = [h for h in TODO._ssh_config_hosts() if NOM_BASE in h]
hosts.sort(key=lambda h: -h.count("+")) def a_defaire(rapport):
dire(f" {len(hosts)} entrée(s) ssh à défaire", journal) """[(niveau, parent_alias, vmid, nom)] du plus PROFOND au plus haut.
for alias in hosts:
bloc = TODO._ssh_config_block(alias) Trié sur le niveau LU dans le rapport, pas déduit du nom. La version
saut = (bloc or {}).get("proxyjump") d'avant comptait les « + » de l'alias — or `alias_etage` remplace le « + »
if saut: du parent par un « - », donc chaque alias en portait exactement UN et le
parent = {"target": saut, "sudo": "sudo ", "jump": ""} tri ne triait rien. La destruction partait du plus HAUT : « qm destroy
_c, out = pve.run(parent, "qm list", 120) --purge » sur l'étage 2 emportait le disque contenant les étages 3 et
for vm in pve.parse_qm_list(out): suivants, sans les avoir arrêtés ni nommés.
if NOM_BASE in (vm.get("name") or ""): """
dire(f" qm destroy {vm['vmid']} sur {saut}", journal) etages = [
pve.run( e
parent, for e in (rapport.get("etages") or [])
f"qm stop {vm['vmid']} --skiplock 1 || true;" if e.get("vmid") and e.get("parent_alias")
f" qm destroy {vm['vmid']} --purge 1", ]
300, etages.sort(key=lambda e: -int(e["niveau"]))
) return [
for niveau in range(1, 30): (
nom = nom_etage(niveau) int(e["niveau"]),
if nom in hosts or niveau == 1: e["parent_alias"],
subprocess.run( int(e["vmid"]),
["sudo", "virsh", "destroy", nom], nom_etage(int(e["niveau"])),
capture_output=True, )
) for e in etages
subprocess.run( ]
[
"sudo",
"virsh", def detruire_une(parent_alias, vmid, nom, journal):
"undefine", """Arrête puis détruit UNE VM, par son VMID. Rend True si elle a disparu.
nom,
"--nvram", Par le VMID et par égalité stricte du nom : un filtre par sous-chaîne
"--remove-all-storage", aurait pris une « deep-pve-lab » de production, et « --purge » emporte les
], disques ET les entrées de sauvegarde.
capture_output=True,
) L'arrêt est CONSTATÉ avant la destruction : « qm stop » rend la main dès
dire( que la tâche est lancée, et sur un hyperviseur imbriqué mesuré 36 fois
" ✓ défait. Les entrées ssh orphelines : menu de nettoyage.", journal plus lent, « qm destroy » arrivait alors que la VM tournait encore et
refusait avec « VM is running ».
"""
parent = {"target": parent_alias, "sudo": "sudo ", "jump": ""}
code, out = pve.run(parent, "qm list", 180)
if code:
dire(f" ✗ {parent_alias} injoignable : rien touché", journal)
return False
presentes = {
int(v["vmid"]): (v.get("name") or "") for v in pve.parse_qm_list(out)
}
if vmid not in presentes:
dire(f" — {vmid} déjà absente de {parent_alias}", journal)
return True
if presentes[vmid] != nom:
dire(
f" ✗ {vmid} sur {parent_alias} s'appelle"
f" « {presentes[vmid]} », pas « {nom} » : rien touché",
journal,
)
return False
pve.run(parent, f"qm stop {vmid} --skiplock 1 || true", 300)
for _ in range(20):
_c, etat = pve.run(parent, f"qm status {vmid}", 120)
if "stopped" in pve.strip_ssh_noise(etat):
break
time.sleep(6)
code, out = pve.run(parent, f"qm destroy {vmid} --purge 1", 600)
if code:
dire(f" ✗ qm destroy {vmid} : code {code}", journal)
for ligne in pve.strip_ssh_noise(out).strip().splitlines()[-3:]:
dire(f" {ligne}", journal)
return False
dire(f" ✓ {nom} ({vmid}) sur {parent_alias}", journal)
return True
def detruire_etage1(journal, dry_run=False):
"""Le domaine libvirt du premier étage — le SEUL qui en soit un.
La boucle d'avant tournait sur trente niveaux avec une condition morte, et
sa branche « niveau == 1 » était vraie même quand la descente n'avait
jamais rien créé : « virsh undefine --remove-all-storage » partait alors
sur un domaine qui pouvait être n'importe quoi, sortie capturée, sans un
mot.
"""
nom = nom_etage(1)
existe = subprocess.run(
["sudo", "-n", "virsh", "dominfo", nom],
capture_output=True,
text=True,
) )
if existe.returncode:
dire(f" — {nom} : aucun domaine libvirt", journal)
return True
if dry_run:
dire(
f" [à blanc] virsh undefine {nom} --remove-all-storage", journal
)
return True
subprocess.run(
["sudo", "virsh", "destroy", nom], capture_output=True, text=True
)
res = subprocess.run(
[
"sudo",
"virsh",
"undefine",
nom,
"--nvram",
"--remove-all-storage",
],
capture_output=True,
text=True,
)
if res.returncode:
dire(
f" ✗ virsh undefine {nom} : {res.stderr.strip()[:160]}", journal
)
return False
dire(f" ✓ {nom} (libvirt)", journal)
return True
def detruire(journal=None, dry_run=False):
"""Défait ce que le DERNIER rapport dit avoir créé, du plus profond.
Rien d'autre. La version d'avant prenait toute entrée ~/.ssh/config dont
le nom contenait « deep-pve », puis sur son rebond détruisait toute VM
dont le nom contenait « deep-pve » — une machine de labo appelée
« deep-pve-lab » sur un hyperviseur de production tombait dedans.
"""
rapport = dernier_rapport()
if not rapport:
dire(" aucun rapport de descente : rien à défaire.", journal)
dire(
" (les entrées ~/.ssh/config orphelines : menu de nettoyage)",
journal,
)
return 0
liste = a_defaire(rapport)
dire(f" rapport : {rapport.get('fichier')}", journal)
dire(f" {len(liste)} VM imbriquée(s) + l'étage 1 :", journal)
for niveau, parent_alias, vmid, nom in liste:
dire(
f" étage {niveau:2d} {nom} ({vmid}) sur {parent_alias}",
journal,
)
dire(f" étage 1 {nom_etage(1)} (libvirt)", journal)
if dry_run:
dire("\n --dry-run : rien ne sera détruit.", journal)
return 0
# Une confirmation, parce que « --purge » emporte les disques et que le
# menu lançait cette option d'une seule touche.
reponse = input("\n Détruire tout cela ? (tapez OUI) : ").strip()
if reponse != "OUI":
dire(" annulé.", journal)
return 1
faits = sum(
1
for niveau, parent_alias, vmid, nom in liste
if detruire_une(parent_alias, vmid, nom, journal)
)
if not detruire_etage1(journal):
faits -= 1
dire(
f"\n {faits} / {len(liste) + 1} défait(s)."
+ (
""
if faits == len(liste) + 1
else " ⚠ il reste des machines : voir plus haut."
),
journal,
)
return 0 if faits == len(liste) + 1 else 1
def principal(argv=None): def principal(argv=None):
@ -570,8 +821,9 @@ def principal(argv=None):
) )
os.makedirs(os.path.dirname(journal), exist_ok=True) os.makedirs(os.path.dirname(journal), exist_ok=True)
if args.detruire: if args.detruire:
detruire(journal) # « --dry-run » était ignoré ici : la prudence naturelle avant une
return 0 # destruction détruisait pour de vrai.
return detruire(journal, dry_run=args.dry_run)
coeurs, ram, disque = capacite_hote() coeurs, ram, disque = capacite_hote()
print( print(
@ -591,6 +843,9 @@ def principal(argv=None):
f" {plan['atteignable']} — manque de {plan['arret']}" f" {plan['atteignable']} — manque de {plan['arret']}"
) )
if not plan["niveaux"]: if not plan["niveaux"]:
if args.depth < 1:
print(f"\n profondeur demandée : {args.depth} — rien à faire.\n")
return 0
print("\n ✗ pas même un étage ne tient sur cette machine.\n") print("\n ✗ pas même un étage ne tient sur cette machine.\n")
return 1 return 1
print(f"\n journal : {journal}") print(f"\n journal : {journal}")
@ -598,11 +853,16 @@ def principal(argv=None):
print(" --dry-run : rien ne sera créé.\n") print(" --dry-run : rien ne sera créé.\n")
descente = Descente(plan, journal, args.dry_run) descente = Descente(plan, journal, args.dry_run)
rapport = descente.parcourir() rapport = descente.parcourir()
chemin = journal[:-4] + ".json" chemin = journal[:-4] + ("-dryrun.json" if args.dry_run else ".json")
with open(chemin, "w", encoding="utf-8") as fh: with open(chemin, "w", encoding="utf-8") as fh:
json.dump(rapport, fh, indent=2) json.dump(rapport, fh, indent=2)
print(f"\n rapport : {chemin}\n") print(f"\n rapport : {chemin}\n")
return 0 if rapport["atteinte"] else 1 # En essai à blanc, c'est le PLAN qui est complet ou non — aucune
# profondeur n'a été atteinte. Hors essai, « non nul » ne suffisait pas :
# une descente morte au deuxième étage sur dix rendait 0.
if args.dry_run:
return 0 if rapport["atteignable"] == rapport["demandee"] else 1
return 0 if rapport["atteinte"] == rapport["demandee"] else 1
if __name__ == "__main__": if __name__ == "__main__":

View file

@ -81,7 +81,10 @@ def nesting_plan(
ram = ((int(ram_dispo_mo) - HOTE_RESERVE_RAM_MO) // 1024) * 1024 ram = ((int(ram_dispo_mo) - HOTE_RESERVE_RAM_MO) // 1024) * 1024
disque = int(disque_libre_go) - HOTE_RESERVE_DISQUE_GO disque = int(disque_libre_go) - HOTE_RESERVE_DISQUE_GO
niveaux, arret = [], "" niveaux, arret = [], ""
for niveau in range(1, max(1, int(profondeur)) + 1): # « max(1, …) » forçait un tour : profondeur 0 rendait un plan d'UN
# étage, et « --depth 0 » créait donc une VM. range(1, 1) est déjà vide,
# et un plan vide est la bonne réponse à une demande vide.
for niveau in range(1, int(profondeur) + 1):
if niveau > 1: if niveau > 1:
ram -= PVE_RAM_MO ram -= PVE_RAM_MO
disque -= PVE_DISQUE_GO disque -= PVE_DISQUE_GO
@ -94,8 +97,16 @@ def nesting_plan(
niveaux.append( niveaux.append(
{ {
"niveau": niveau, "niveau": niveau,
# Le plancher est VCPU_IMBRIQUE et non 1 : sur un hôte de
# quatre cœurs, « // 4 » donnait UN vCPU au premier étage —
# l'hyperviseur parent — alors que son invité en recevait
# deux. Un parent plus étroit que son enfant est absurde, et
# c'est tout l'inverse de ce que ce module raconte.
"vcpu": ( "vcpu": (
max(1, min(VCPU_NIVEAU1_MAX, int(cpu_hote) // 4)) max(
VCPU_IMBRIQUE,
min(VCPU_NIVEAU1_MAX, int(cpu_hote) // 4),
)
if niveau == 1 if niveau == 1
else VCPU_IMBRIQUE else VCPU_IMBRIQUE
), ),

View file

@ -76,7 +76,13 @@ class LongTestMenuMixin:
"deep_proxmox.py", f"--depth {self._longtest_depth()}" "deep_proxmox.py", f"--depth {self._longtest_depth()}"
) )
elif status == "3": elif status == "3":
self._longtest_run("deep_proxmox.py", "--detruire") # Le script demande « OUI » avant de détruire, mais il liste
# d'abord : on lui fait faire cette liste À BLANC pour que le
# choix « 3 » d'une touche ne mène pas directement à un
# « qm destroy --purge ».
self._longtest_run("deep_proxmox.py", "--detruire --dry-run")
if self._is_yes(input(f"\n{t('Destroy all that? (y/N): ')}")):
self._longtest_run("deep_proxmox.py", "--detruire")
else: else:
print(t("Command not found !")) print(t("Command not found !"))

View file

@ -3422,6 +3422,10 @@ TRANSLATIONS = {
"fr": "L'adresse écrite n'est peut-être pas celle qu'il faut à pmxcfs.", "fr": "L'adresse écrite n'est peut-être pas celle qu'il faut à pmxcfs.",
"en": "The address written may not be the one pmxcfs needs.", "en": "The address written may not be the one pmxcfs needs.",
}, },
"Destroy all that? (y/N): ": {
"fr": "Détruire tout cela ? (o/N) : ",
"en": "Destroy all that? (y/N): ",
},
"Long tests - real VMs, hours": { "Long tests - real VMs, hours": {
"fr": "⏳ Tests longs - vraies VM, des heures", "fr": "⏳ Tests longs - vraies VM, des heures",
"en": "⏳ Long tests - real VMs, hours", "en": "⏳ Long tests - real VMs, hours",

View file

@ -42,10 +42,29 @@ class TestLePlanDesEtages(unittest.TestCase):
de démarrage ; les mêmes 2 vCPU avançaient. Amener douze processeurs de démarrage ; les mêmes 2 vCPU avançaient. Amener douze processeurs
en ligne demande autant d'allers-retours à travers la pile.""" en ligne demande autant d'allers-retours à travers la pile."""
niveaux = nesting.nesting_plan(4, **self.HOTE)["niveaux"] niveaux = nesting.nesting_plan(4, **self.HOTE)["niveaux"]
self.assertGreater(niveaux[0]["vcpu"], nesting.VCPU_IMBRIQUE - 1) # STRICTEMENT plus grand. « > VCPU_IMBRIQUE - 1 » était satisfait par
# la valeur imbriquée elle-même : remplacer tout le calcul du premier
# étage par VCPU_IMBRIQUE laissait les tests verts, donc cpu_hote
# n'était couvert par rien.
self.assertGreater(niveaux[0]["vcpu"], nesting.VCPU_IMBRIQUE)
for n in niveaux[1:]: for n in niveaux[1:]:
self.assertEqual(n["vcpu"], nesting.VCPU_IMBRIQUE) self.assertEqual(n["vcpu"], nesting.VCPU_IMBRIQUE)
def test_a_parent_is_never_narrower_than_its_child(self):
"""Sur un hôte de quatre cœurs, « // 4 » donnait UN vCPU au premier
étage — l'hyperviseur — alors que son invité en recevait deux."""
for coeurs in (2, 4, 8, 12, 28):
with self.subTest(coeurs=coeurs):
niveaux = nesting.nesting_plan(
3,
cpu_hote=coeurs,
ram_dispo_mo=32768,
disque_libre_go=300,
)["niveaux"]
self.assertGreaterEqual(
niveaux[0]["vcpu"], niveaux[1]["vcpu"], f"{coeurs} cœurs"
)
def test_running_out_of_ram_is_named(self): def test_running_out_of_ram_is_named(self):
plan = nesting.nesting_plan( plan = nesting.nesting_plan(
10, cpu_hote=8, ram_dispo_mo=12288, disque_libre_go=500 10, cpu_hote=8, ram_dispo_mo=12288, disque_libre_go=500
@ -65,6 +84,13 @@ class TestLePlanDesEtages(unittest.TestCase):
for n in plan["niveaux"]: for n in plan["niveaux"]:
self.assertGreaterEqual(n["disque"], nesting.DISQUE_MIN_GO) self.assertGreaterEqual(n["disque"], nesting.DISQUE_MIN_GO)
def test_a_depth_of_zero_asks_for_nothing(self):
for profondeur in (0, -1, -7):
with self.subTest(profondeur=profondeur):
plan = nesting.nesting_plan(profondeur, **self.HOTE)
self.assertEqual(plan["niveaux"], [])
self.assertEqual(plan["atteignable"], 0)
def test_a_machine_too_small_for_even_one_level(self): def test_a_machine_too_small_for_even_one_level(self):
plan = nesting.nesting_plan( plan = nesting.nesting_plan(
3, cpu_hote=2, ram_dispo_mo=4096, disque_libre_go=200 3, cpu_hote=2, ram_dispo_mo=4096, disque_libre_go=200
@ -76,10 +102,14 @@ class TestLePlanDesEtages(unittest.TestCase):
def test_a_plan_is_never_promised_beyond_what_fits(self): def test_a_plan_is_never_promised_beyond_what_fits(self):
# Mieux vaut annoncer six étages et en réussir six que d'en promettre # Mieux vaut annoncer six étages et en réussir six que d'en promettre
# dix et mourir au septième sans savoir pourquoi. # dix et mourir au septième sans savoir pourquoi.
for profondeur in range(1, 13): # Depuis 0 et depuis les négatifs : « max(1, …) » forçait un tour,
# donc « --depth 0 » rendait un plan d'UN étage et créait une VM.
for profondeur in range(-2, 13):
plan = nesting.nesting_plan(profondeur, **self.HOTE) plan = nesting.nesting_plan(profondeur, **self.HOTE)
self.assertEqual(len(plan["niveaux"]), plan["atteignable"]) self.assertEqual(len(plan["niveaux"]), plan["atteignable"])
self.assertLessEqual(plan["atteignable"], profondeur) # « max(0, …) » : une demande négative ne peut pas donner un
# nombre d'étages négatif, elle donne zéro.
self.assertLessEqual(plan["atteignable"], max(0, profondeur))
class TestLaProfondeurDUnHote(unittest.TestCase): class TestLaProfondeurDUnHote(unittest.TestCase):

View file

@ -36,14 +36,22 @@ class TestLaFrontiere(unittest.TestCase):
# LongTest, sinon la suite unitaire créerait des VM. # LongTest, sinon la suite unitaire créerait des VM.
self.assertNotIn("LongTest", lanceur) self.assertNotIn("LongTest", lanceur)
def test_the_naming_rule_is_written_where_it_is_read(self): def test_the_runner_only_looks_under_test(self):
# Un fichier hors préfixe tombe dans le même silence qu'un fichier """Le lanceur balaie TOUT test/test_*.py depuis qu'une liste de
# absent : douze tests écrits, jamais lancés. préfixes a laissé 2400 tests hors de la suite.
La frontière n'est donc plus un nom mais un RÉPERTOIRE : ce qui doit
rester hors de la suite doit vivre ailleurs que dans test/. C'est
exactement pourquoi LongTest est à la racine."""
with open( with open(
os.path.join(RACINE, "script/test/run_unit_test.sh"), os.path.join(RACINE, "script/test/run_unit_test.sh"),
encoding="utf-8", encoding="utf-8",
) as fh: ) as fh:
self.assertIn("NOMMER UN NOUVEAU FICHIER", fh.read()) lanceur = fh.read()
self.assertIn("test/test_*.py", lanceur)
# Aucun chemin du lanceur ne sort de test/ : sinon LongTest y
# entrerait par la porte de service.
self.assertNotIn("LongTest", lanceur)
def test_the_script_is_executable_and_documented(self): def test_the_script_is_executable_and_documented(self):
script = os.path.join(RACINE, "LongTest/deep_proxmox.py") script = os.path.join(RACINE, "LongTest/deep_proxmox.py")
@ -64,6 +72,11 @@ class TestLEssaiABlanc(unittest.TestCase):
@classmethod @classmethod
def setUpClass(cls): def setUpClass(cls):
import tempfile
# HOME temporaire : la suite unitaire tourne souvent, et elle n'a pas
# à semer un rapport dans ~/.erplibre à chaque passage.
cls.maison = tempfile.mkdtemp()
cls.res = subprocess.run( cls.res = subprocess.run(
[ [
PYTHON, PYTHON,
@ -76,19 +89,34 @@ class TestLEssaiABlanc(unittest.TestCase):
text=True, text=True,
timeout=180, timeout=180,
cwd=RACINE, cwd=RACINE,
env=dict(os.environ, PYTHONPATH=RACINE), env=dict(os.environ, PYTHONPATH=RACINE, HOME=cls.maison),
) )
@classmethod
def tearDownClass(cls):
import shutil
shutil.rmtree(cls.maison, ignore_errors=True)
def test_it_exits_cleanly(self): def test_it_exits_cleanly(self):
self.assertEqual(self.res.returncode, 0, self.res.stderr[-800:]) self.assertEqual(self.res.returncode, 0, self.res.stderr[-800:])
def test_it_announces_the_plan_before_anything(self): def test_it_announces_the_plan_before_anything(self):
sortie = self.res.stdout """Les LIGNES du plan, pas les chiffres.
self.assertIn("étage", sortie)
# Quatre étages demandés, quatre lignes de plan. La version d'avant cherchait « 1 », « 2 », « 3 », « 4 » dans la
for niveau in ("1", "2", "3", "4"): sortie : l'en-tête « 28 cœurs, 29128 Mo, 138 Go » et l'horodatage du
self.assertIn(niveau, sortie) journal les fournissent tous. Elle passait même à --depth 1, avec une
self.assertIn("dry-run", sortie) seule ligne de plan — elle ne prouvait rien."""
import re
plan = re.findall(
r"^\s+(\d+)\s+(\d+)\s+(\d+) Mo\s+(\d+) Go\s*$",
self.res.stdout,
re.M,
)
self.assertEqual([int(p[0]) for p in plan], [1, 2, 3, 4])
self.assertIn("dry-run", self.res.stdout)
def test_it_shows_the_commands_it_would_send(self): def test_it_shows_the_commands_it_would_send(self):
# Une étape affichée est une étape rejouable à la main : c'est ainsi # Une étape affichée est une étape rejouable à la main : c'est ainsi
@ -96,6 +124,43 @@ class TestLEssaiABlanc(unittest.TestCase):
self.assertIn("qm create", self.res.stdout) self.assertIn("qm create", self.res.stdout)
self.assertIn("install_proxmox.sh", self.res.stdout) self.assertIn("install_proxmox.sh", self.res.stdout)
def test_the_installer_is_run_by_bash_not_sh(self):
"""Le script porte « set -euo pipefail » et un shebang bash.
Sur Debian /bin/sh est dash, qui répond « set: Illegal option -o
pipefail » et sort à la PREMIÈRE ligne — vérifié. Lancé par sh, chaque
étage aurait échoué sur l'installation, à tous les coups."""
# Sur la LIGNE, pas dans le texte : « bash /tmp/… » contient
# « sh /tmp/… », donc un assertNotIn naïf échouait sur lui-même.
lignes = [
ligne.strip()
for ligne in self.res.stdout.splitlines()
if "install_proxmox.sh" in ligne
and not ligne.strip().startswith("scp")
]
self.assertTrue(lignes)
for ligne in lignes:
self.assertTrue(
ligne.startswith("bash "), f"lancé par autre chose : {ligne}"
)
def test_the_dry_run_claims_nothing_reached(self):
"""Le rapport d'un essai à blanc était indiscernable d'une réussite —
JSON compris — et « --detruire » s'en servait."""
import glob
import json
fichiers = glob.glob(
os.path.join(self.maison, ".erplibre/longtest/*.json")
)
self.assertEqual(len(fichiers), 1, fichiers)
self.assertIn("dryrun", fichiers[0])
with open(fichiers[0], encoding="utf-8") as fh:
rapport = json.load(fh)
self.assertTrue(rapport["dry_run"])
self.assertEqual(rapport["atteinte"], 0)
self.assertTrue(all(not e["ok"] for e in rapport["etages"]))
def test_the_first_level_is_wide_and_the_others_are_not(self): def test_the_first_level_is_wide_and_the_others_are_not(self):
# 12 vCPU au quatrième étage ont gelé un noyau invité ; deux # 12 vCPU au quatrième étage ont gelé un noyau invité ; deux
# avançaient. # avançaient.
@ -115,6 +180,79 @@ class TestLEssaiABlanc(unittest.TestCase):
self.assertEqual(niveaux[niveau], 2, f"étage {niveau}") self.assertEqual(niveaux[niveau], 2, f"étage {niveau}")
class TestDefaireSansEffacerAutreChose(unittest.TestCase):
"""« --detruire » effaçait par SOUS-CHAÎNE de nom, dans le mauvais ordre,
sans confirmation et sans honorer --dry-run.
Quatre défauts trouvés en attaquant le code écrit, chacun capable
d'emporter une machine qui n'appartient pas au test. « qm destroy --purge »
emporte les disques ET les entrées de sauvegarde."""
def setUp(self):
sys.path.insert(0, os.path.join(RACINE, "LongTest"))
import deep_proxmox
self.dp = deep_proxmox
def test_the_deepest_level_goes_first(self):
"""Le tri comptait les « + » de l'alias — or alias_etage remplace le
« + » du parent par un « - », donc chaque alias en portait
exactement UN. Le tri ne triait rien, et la destruction partait du
plus HAUT : « qm destroy --purge » sur l'étage 2 emportait le disque
contenant les étages 3 et suivants."""
rapport = {
"etages": [
{"niveau": 2, "vmid": 100, "parent_alias": "a"},
{"niveau": 4, "vmid": 100, "parent_alias": "c"},
{"niveau": 3, "vmid": 100, "parent_alias": "b"},
]
}
niveaux = [n for n, _p, _v, _nom in self.dp.a_defaire(rapport)]
self.assertEqual(niveaux, [4, 3, 2])
def test_a_level_without_a_vmid_is_not_guessed(self):
# Un étage abandonné avant « qm create » n'a rien créé : ne rien
# inventer à sa place.
rapport = {"etages": [{"niveau": 2}, {"niveau": 3, "vmid": 101}]}
self.assertEqual(len(self.dp.a_defaire(rapport)), 0)
def test_the_alias_chain_really_flattens_the_plus(self):
# La cause du tri mort, énoncée pour qu'on ne la réintroduise pas.
alias, precedent = "deep-pve-1", "deep-pve-1"
for niveau in (2, 3, 4):
alias = self.dp.alias_etage(niveau, precedent)
precedent = alias
self.assertEqual(alias.count("+"), 1, alias)
def test_an_exact_name_is_required(self):
"""Le filtre était « NOM_BASE in name » : une VM de labo appelée
« deep-pve-lab » sur un hyperviseur de production tombait dedans."""
import inspect
src = inspect.getsource(self.dp.detruire_une)
self.assertIn("!= nom", src)
self.assertNotIn("in presentes[vmid]", src)
def test_dry_run_reports_are_never_used_to_destroy(self):
"""Un rapport d'essai à blanc n'a rien créé : s'en servir ferait
détruire d'après un plan."""
import inspect
src = inspect.getsource(self.dp.dernier_rapport)
self.assertIn('rapport.get("dry_run")', src)
def test_destruction_honours_dry_run_and_asks(self):
import inspect
src = inspect.getsource(self.dp.detruire)
self.assertIn("dry_run", src)
# Une confirmation explicite, pas un « o/N » : le menu lançait cette
# option d'une seule touche.
self.assertIn("OUI", src)
principal = inspect.getsource(self.dp.principal)
self.assertIn("dry_run=args.dry_run", principal)
class TestLeMenu(unittest.TestCase): class TestLeMenu(unittest.TestCase):
def test_the_mixin_is_wired_into_TODO(self): def test_the_mixin_is_wired_into_TODO(self):
todo = TODO.__new__(TODO) todo = TODO.__new__(TODO)