[FIX] LongTest : ne pas détruire sous une descente vivante ; blocs ssh
Relecture adversaire du commit précédent : il avait CRÉÉ un danger. Le rapport
s'écrivant maintenant VM par VM, celui de la descente EN COURS est le plus
récent, et « --detruire » l'aurait choisi — qm destroy --purge sur l'arbre que
le processus installait encore. Deux garde-fous : un PID dans le rapport, et
un refus net tant qu'un autre deep_proxmox.py tourne. Le second est nécessaire
car une descente déjà lancée a l'ancien module en mémoire.
Reconnu par ARGUMENT, pas par sous-chaîne : mon propre « pgrep -f
deep_proxmox.py » de surveillance donnait deux faux positifs sur trois.
Trois autres, mêmes preuves :
- un rapport vide plus récent masquait celui qui nommait les VM réelles ;
- detruire() ne créditait jamais l'étage 1 : le décompte était décalé de un
dans tous les cas, donc l'avertissement sortait toujours ;
- _ssh_config_drop_hosts prenait l'indentation pour de la syntaxe. Sur un bloc
au corps non indenté, seule la ligne Host partait et ssh rattachait
« StrictHostKeyChecking no » au bloc précédent — un serveur de production.
Et un bloc partagé (« Host prod-db vm-a ») partait en entier.
Les cinq correctifs meurent sous mutation.
--- EN ---
Adversarial review of the previous commit: it had CREATED a hazard. With the
report now written VM by VM, the RUNNING descent's is the most recent, and
"--detruire" would have picked it — qm destroy --purge on the tree the process
was still installing. Two guards: a PID in the report, and a flat refusal
while another deep_proxmox.py runs. The second is needed because an
already-running descent holds the old module in memory.
Matched by ARGUMENT, not substring: my own monitoring "pgrep -f
deep_proxmox.py" produced two false positives out of three.
Three more, same evidence:
- a newer empty report masked the one naming the real VMs;
- detruire() never credited level 1: the count was off by one in every case,
so the warning always fired;
- _ssh_config_drop_hosts took indentation for syntax. On a block with an
unindented body only the Host line went, and ssh attached
"StrictHostKeyChecking no" to the preceding block — a production server.
And a shared block ("Host prod-db vm-a") went entirely.
All five fixes die under mutation.
Assisted-by: claude-opus-5
(cherry picked from commit 7d348d976f96e420b4fec4d879911b658a730ffa)
This commit is contained in:
parent
d0a06c03b7
commit
c046e027e9
4 changed files with 512 additions and 27 deletions
|
|
@ -628,12 +628,13 @@ class Descente:
|
|||
return
|
||||
temporaire = self.chemin_json + ".tmp"
|
||||
try:
|
||||
etat = self._etat(interrompu=True, en_cours=en_cours)
|
||||
# Le PID de la descente qui écrit : c'est ce qui distingue un
|
||||
# rapport ABANDONNÉ d'un rapport en cours d'écriture. Le rapport
|
||||
# final, lui, n'en porte pas — la descente est finie.
|
||||
etat["pid"] = os.getpid()
|
||||
with open(temporaire, "w", encoding="utf-8") as fh:
|
||||
json.dump(
|
||||
self._etat(interrompu=True, en_cours=en_cours),
|
||||
fh,
|
||||
indent=2,
|
||||
)
|
||||
json.dump(etat, fh, indent=2)
|
||||
os.replace(temporaire, self.chemin_json)
|
||||
except OSError as err:
|
||||
self.dire(f" ⚠ rapport non écrit : {err}")
|
||||
|
|
@ -664,13 +665,84 @@ class Descente:
|
|||
return etat
|
||||
|
||||
|
||||
def _lance_ce_script(pid):
|
||||
"""`pid` exécute-t-il CE script — et non pas seulement le nomme-t-il ?
|
||||
|
||||
Par ARGUMENT, jamais par sous-chaîne. Constaté sur cette machine : un
|
||||
« pgrep -f deep_proxmox.py » posé dans une boucle de surveillance donne un
|
||||
shell dont la ligne de commande contient le motif, et le contrôle comptait
|
||||
ce shell comme une descente — deux faux positifs sur trois. Un argument
|
||||
qui SE TERMINE par le nom du fichier, lui, ne peut venir que d'un
|
||||
interpréteur qu'on a lancé dessus.
|
||||
"""
|
||||
try:
|
||||
with open(f"/proc/{int(pid)}/cmdline", "rb") as fh:
|
||||
arguments = fh.read().split(b"\0")
|
||||
except (OSError, ValueError):
|
||||
return False
|
||||
return any(a.endswith(b"deep_proxmox.py") for a in arguments)
|
||||
|
||||
|
||||
def descente_vivante(pid):
|
||||
"""Le processus `pid` est-il une descente EN COURS ?
|
||||
|
||||
Le PID seul ne suffirait pas : les numéros se réutilisent, et rien ne dit
|
||||
qu'un rapport vieux d'une semaine ne porte pas le PID d'un shell
|
||||
d'aujourd'hui. La ligne de commande est donc lue aussi.
|
||||
"""
|
||||
return bool(pid) and _lance_ce_script(pid)
|
||||
|
||||
|
||||
def autre_deep_proxmox():
|
||||
"""Les PID des AUTRES deep_proxmox.py vivants. Le sien est exclu.
|
||||
|
||||
Le garde-fou du rapport — un PID dans le fichier — ne protège que les
|
||||
descentes lancées APRÈS son écriture : celle qui tournait déjà avait
|
||||
chargé l'ancien module en mémoire et n'écrira jamais de PID. Constaté sur
|
||||
une descente réelle de dix étages, à l'étage 4. Ce contrôle-ci ne dépend
|
||||
d'aucun rapport : détruire pendant qu'une descente tourne n'est jamais
|
||||
juste, quel que soit le rapport choisi.
|
||||
|
||||
/proc plutôt que pgrep : « pgrep -f deep_proxmox » attrape le shell qui
|
||||
l'invoque, et on croit alors voir survivre un processus qui n'existe pas.
|
||||
"""
|
||||
moi = os.getpid()
|
||||
vivants = []
|
||||
try:
|
||||
entrees = os.listdir("/proc")
|
||||
except OSError:
|
||||
return vivants
|
||||
for entree in entrees:
|
||||
if not entree.isdigit() or int(entree) == moi:
|
||||
continue
|
||||
if _lance_ce_script(entree):
|
||||
vivants.append(int(entree))
|
||||
return vivants
|
||||
|
||||
|
||||
def dernier_rapport():
|
||||
"""Le rapport JSON le plus récent, ou {}.
|
||||
"""Le rapport le plus récent qui NOMME quelque chose à défaire, ou {}.
|
||||
|
||||
C'est le SEUL enregistrement de ce que la descente a créé : un couple
|
||||
(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.
|
||||
|
||||
Deux rapports sont ÉCARTÉS, et chacun l'est pour un accident précis :
|
||||
|
||||
* celui d'une descente VIVANTE. Depuis que le rapport s'écrit VM par VM,
|
||||
la descente en cours en a un sur le disque, et c'est le plus récent :
|
||||
« --detruire » aurait détruit l'arbre sous le processus qui installait
|
||||
encore, emportant des heures de mesure. Avant, la descente en cours
|
||||
n'avait aucun rapport et la question ne se posait pas — le correctif a
|
||||
créé le danger.
|
||||
|
||||
* celui qui n'a RIEN créé. Un second lancement qui meurt à l'étage 1 —
|
||||
« le disque existe déjà » — écrit un rapport vide sous un horodatage
|
||||
plus tardif. Il masquait le partiel qui nommait les VM réelles :
|
||||
« 0 VM imbriquée(s) », puis « virsh undefine --remove-all-storage » sur
|
||||
l'étage 1, dont le disque contient les étages 2 et suivants — jamais
|
||||
arrêtés, jamais nommés.
|
||||
"""
|
||||
dossier = os.path.expanduser("~/.erplibre/longtest")
|
||||
try:
|
||||
|
|
@ -680,14 +752,20 @@ def dernier_rapport():
|
|||
except OSError:
|
||||
return {}
|
||||
for nom in reversed(fichiers):
|
||||
chemin = os.path.join(dossier, nom)
|
||||
try:
|
||||
with open(os.path.join(dossier, nom), encoding="utf-8") as fh:
|
||||
with open(chemin, 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)
|
||||
if descente_vivante(rapport.get("pid")):
|
||||
dire(f" ⏳ descente EN COURS ({rapport['pid']}) : {nom} ignoré")
|
||||
continue
|
||||
if not (rapport.get("etages") or []):
|
||||
continue # rien créé : ne pas masquer un rapport qui nomme des VM
|
||||
rapport["fichier"] = chemin
|
||||
return rapport
|
||||
return {}
|
||||
|
||||
|
|
@ -820,6 +898,18 @@ def detruire(journal=None, dry_run=False):
|
|||
dont le nom contenait « deep-pve » — une machine de labo appelée
|
||||
« deep-pve-lab » sur un hyperviseur de production tombait dedans.
|
||||
"""
|
||||
# Avant tout : refuser tant qu'une descente tourne. Elle installe encore
|
||||
# sur les machines qu'on s'apprête à détruire, et son rapport peut être
|
||||
# celui qu'on vient de choisir.
|
||||
autres = autre_deep_proxmox()
|
||||
if autres:
|
||||
dire(
|
||||
f" ⛔ une descente tourne ({', '.join(map(str, autres))}) :"
|
||||
" rien ne sera détruit.",
|
||||
journal,
|
||||
)
|
||||
dire(" Attendre qu'elle finisse, ou l'arrêter d'abord.", journal)
|
||||
return 1
|
||||
rapport = dernier_rapport()
|
||||
if not rapport:
|
||||
dire(" aucun rapport de descente : rien à défaire.", journal)
|
||||
|
|
@ -851,8 +941,14 @@ def detruire(journal=None, dry_run=False):
|
|||
for niveau, parent_alias, vmid, nom in liste
|
||||
if detruire_une(parent_alias, vmid, nom, journal)
|
||||
)
|
||||
if not detruire_etage1(journal):
|
||||
faits -= 1
|
||||
# « if not … : faits -= 1 » : un succès de l'étage 1 n'ajoutait RIEN,
|
||||
# alors que le total est len(liste) + 1. Le décompte était décalé de un
|
||||
# dans TOUS les cas — une destruction complète annonçait « il reste des
|
||||
# machines » et sortait 1, si bien que le seul avertissement censé
|
||||
# prévenir qu'un disque de plusieurs dizaines de Go reste alloué
|
||||
# s'affichait toujours, et qu'on apprenait à ne plus le lire.
|
||||
if detruire_etage1(journal):
|
||||
faits += 1
|
||||
dire(
|
||||
f"\n {faits} / {len(liste) + 1} défait(s)."
|
||||
+ (
|
||||
|
|
|
|||
|
|
@ -1398,34 +1398,69 @@ class TODO(
|
|||
|
||||
@staticmethod
|
||||
def _ssh_config_drop_hosts(content, names):
|
||||
"""Retire les blocs « Host … » qui déclarent l'un de `names`.
|
||||
"""Retire de ~/.ssh/config ce qui déclare l'un de `names`.
|
||||
|
||||
On découpe en blocs plutôt que de substituer par expression
|
||||
régulière : une ligne Host peut porter PLUSIEURS noms, et il faut
|
||||
alors retirer le bloc entier dès qu'un seul de ses noms est repris —
|
||||
sinon le même nom se retrouverait défini deux fois, et ssh
|
||||
appliquerait la première définition rencontrée."""
|
||||
régulière : une ligne Host peut porter PLUSIEURS noms.
|
||||
|
||||
Deux règles, chacune corrigeant une perte de données CONSTATÉE dans
|
||||
le fichier d'un utilisateur.
|
||||
|
||||
1. Seuls « Host » et « Match » clôturent un bloc. La règle d'avant —
|
||||
« une ligne non indentée clôt le bloc » — prenait l'indentation
|
||||
pour de la syntaxe, alors qu'elle est cosmétique dans ce format et
|
||||
qu'un fichier écrit à la main s'en passe souvent. Sur un bloc au
|
||||
corps non indenté, seule la ligne « Host » partait : HostName,
|
||||
User, IdentityFile et « StrictHostKeyChecking no » restaient, sans
|
||||
Host au-dessus, et ssh les rattachait au bloc PRÉCÉDENT. La
|
||||
vérification de clé d'hôte se retrouvait désactivée sur un serveur
|
||||
de production.
|
||||
|
||||
2. Un bloc qui déclare AUSSI des noms qu'on ne retire pas survit,
|
||||
amputé de ceux-là seulement. Il partait en entier : « Host prod-db
|
||||
vm-a » perdait le prod-db de l'utilisateur, et le surnom qu'on
|
||||
ajoute à un bloc généré disparaissait au déploiement suivant.
|
||||
|
||||
La queue du bloc — lignes vides et commentaires — n'est pas emportée :
|
||||
elle précède le plus souvent le bloc SUIVANT, et l'utilisateur y met
|
||||
ses propres notes.
|
||||
"""
|
||||
drop = set(names)
|
||||
out, block, block_names = [], [], set()
|
||||
out, block, block_names = [], [], []
|
||||
|
||||
def flush():
|
||||
if block and not (block_names & drop):
|
||||
if not block:
|
||||
return
|
||||
restants = [n for n in block_names if n not in drop]
|
||||
if restants == block_names:
|
||||
out.extend(block)
|
||||
return
|
||||
fin = len(block)
|
||||
while fin > 1 and (
|
||||
not block[fin - 1].strip()
|
||||
or block[fin - 1].lstrip().startswith("#")
|
||||
):
|
||||
fin -= 1
|
||||
if restants:
|
||||
tete = block[0]
|
||||
marge = tete[: len(tete) - len(tete.lstrip())]
|
||||
out.append(f"{marge}Host {' '.join(restants)}\n")
|
||||
out.extend(block[1:fin])
|
||||
out.extend(block[fin:])
|
||||
|
||||
for line in content.splitlines(keepends=True):
|
||||
if re.match(r"^[ \t]*Host[ \t]+", line):
|
||||
if re.match(r"^[ \t]*Host[ \t]+", line, re.I):
|
||||
flush()
|
||||
block = [line]
|
||||
block_names = set(line.split()[1:])
|
||||
block_names = line.split()[1:]
|
||||
elif re.match(r"^[ \t]*Match[ \t]+", line, re.I):
|
||||
# Match ouvre une section qui n'appartient à aucun Host : la
|
||||
# garder telle quelle, quel que soit le sort du bloc d'avant.
|
||||
flush()
|
||||
block, block_names = [], []
|
||||
out.append(line)
|
||||
elif block:
|
||||
# Une ligne non indentée et non vide clôt le bloc (Match,
|
||||
# directive globale…) : elle n'appartient à personne.
|
||||
if line.strip() and not line[:1].isspace():
|
||||
flush()
|
||||
block, block_names = [], set()
|
||||
out.append(line)
|
||||
else:
|
||||
block.append(line)
|
||||
block.append(line)
|
||||
else:
|
||||
out.append(line)
|
||||
flush()
|
||||
|
|
|
|||
|
|
@ -1019,6 +1019,124 @@ class TestUnSeulNomDansSshConfig(unittest.TestCase):
|
|||
)
|
||||
|
||||
|
||||
class TestNeRienPerdreDansSshConfig(unittest.TestCase):
|
||||
"""~/.ssh/config contient les entrées PERSONNELLES de l'utilisateur.
|
||||
|
||||
Ce fichier est réécrit en entier à chaque déploiement de VM. Deux pertes
|
||||
de données y ont été constatées, l'une capable de désactiver la
|
||||
vérification de clé d'hôte sur un serveur de production."""
|
||||
|
||||
def setUp(self):
|
||||
import sys
|
||||
|
||||
sys.argv = ["todo.py"]
|
||||
from script.todo.todo import TODO
|
||||
|
||||
self.retirer = TODO._ssh_config_drop_hosts
|
||||
|
||||
def test_a_block_with_an_unindented_body_goes_entirely(self):
|
||||
"""L'indentation est COSMÉTIQUE dans ce format, et un fichier écrit à
|
||||
la main s'en passe souvent. La règle d'avant la prenait pour de la
|
||||
syntaxe : seule la ligne « Host » partait."""
|
||||
avant = (
|
||||
"Host prod\n"
|
||||
" HostName prod.example.com\n"
|
||||
" User root\n"
|
||||
"\n"
|
||||
"Host deep-1\n"
|
||||
"HostName 10.0.0.1\n"
|
||||
"User erplibre\n"
|
||||
"StrictHostKeyChecking no\n"
|
||||
"UserKnownHostsFile /dev/null\n"
|
||||
"IdentityFile ~/.ssh/id_deep\n"
|
||||
)
|
||||
apres = self.retirer(avant, ["deep-1"])
|
||||
# Rien du bloc retiré ne subsiste : sans Host au-dessus, ssh
|
||||
# rattacherait ces lignes à « prod » et la production perdrait sa
|
||||
# vérification de clé d'hôte.
|
||||
for orphelin in (
|
||||
"10.0.0.1",
|
||||
"StrictHostKeyChecking",
|
||||
"UserKnownHostsFile",
|
||||
"id_deep",
|
||||
):
|
||||
self.assertNotIn(orphelin, apres, orphelin)
|
||||
# Et le bloc de l'utilisateur est intact.
|
||||
self.assertIn("HostName prod.example.com", apres)
|
||||
self.assertIn("User root", apres)
|
||||
|
||||
def test_a_shared_host_line_keeps_the_names_not_dropped(self):
|
||||
"""« Host prod-db vm-a » perdait le prod-db de l'utilisateur : le bloc
|
||||
partait en entier dès qu'UN de ses noms était repris."""
|
||||
avant = (
|
||||
"Host prod-db vm-a\n"
|
||||
" HostName db.interne\n"
|
||||
" ProxyJump pve9\n"
|
||||
)
|
||||
apres = self.retirer(avant, ["vm-a"])
|
||||
self.assertIn("Host prod-db\n", apres)
|
||||
self.assertNotIn("vm-a", apres)
|
||||
# Le corps suit le nom qui reste : sinon prod-db perd son rebond.
|
||||
self.assertIn("HostName db.interne", apres)
|
||||
self.assertIn("ProxyJump pve9", apres)
|
||||
|
||||
def test_a_nickname_added_by_hand_survives_a_redeploy(self):
|
||||
avant = "Host pve9+vm-a webtest\n HostName 10.10.10.5\n"
|
||||
apres = self.retirer(avant, ["pve9+vm-a"])
|
||||
self.assertIn("Host webtest\n", apres)
|
||||
self.assertIn("HostName 10.10.10.5", apres)
|
||||
|
||||
def test_all_names_dropped_removes_the_block(self):
|
||||
avant = "Host a b\n HostName 1.2.3.4\n\nHost garde\n User x\n"
|
||||
apres = self.retirer(avant, ["a", "b"])
|
||||
self.assertNotIn("1.2.3.4", apres)
|
||||
self.assertIn("Host garde", apres)
|
||||
|
||||
def test_a_match_section_is_never_swallowed(self):
|
||||
avant = (
|
||||
"Host part\n"
|
||||
" HostName 10.0.0.9\n"
|
||||
"\n"
|
||||
"Match host *.interne\n"
|
||||
" User admin\n"
|
||||
)
|
||||
apres = self.retirer(avant, ["part"])
|
||||
self.assertIn("Match host *.interne", apres)
|
||||
self.assertIn("User admin", apres)
|
||||
self.assertNotIn("10.0.0.9", apres)
|
||||
|
||||
def test_comments_before_the_next_block_are_not_swallowed(self):
|
||||
avant = (
|
||||
"Host part\n"
|
||||
" HostName 10.0.0.9\n"
|
||||
"\n"
|
||||
"# la machine du client, ne pas toucher\n"
|
||||
"Host client\n"
|
||||
" HostName 10.0.0.10\n"
|
||||
)
|
||||
apres = self.retirer(avant, ["part"])
|
||||
self.assertIn("# la machine du client, ne pas toucher", apres)
|
||||
self.assertIn("Host client", apres)
|
||||
|
||||
def test_global_directives_above_the_first_host_stay(self):
|
||||
avant = "ServerAliveInterval 60\n\nHost part\n HostName 10.0.0.9\n"
|
||||
apres = self.retirer(avant, ["part"])
|
||||
self.assertIn("ServerAliveInterval 60", apres)
|
||||
self.assertNotIn("10.0.0.9", apres)
|
||||
|
||||
def test_the_keyword_is_read_case_insensitively(self):
|
||||
# ssh lit ses mots-clés sans égard à la casse ; nous aussi, sinon un
|
||||
# « host » minuscule échappe au retrait et le nom vit deux fois.
|
||||
avant = "host part\n HostName 10.0.0.9\n"
|
||||
self.assertNotIn("10.0.0.9", self.retirer(avant, ["part"]))
|
||||
|
||||
def test_hostname_is_not_mistaken_for_a_host_line(self):
|
||||
avant = "Host garde\n HostName part\n"
|
||||
apres = self.retirer(avant, ["part"])
|
||||
self.assertIn("Host garde", apres)
|
||||
self.assertIn("HostName part", apres)
|
||||
|
||||
|
||||
class TestLAncienNomSEnVa(unittest.TestCase):
|
||||
"""La convention a changé : les entrées écrites AVANT portent le nom
|
||||
court, et rien ne les retirerait — elles ne portent pas le nom qu'on
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import shutil
|
|||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
|
||||
sys.argv = ["todo.py"]
|
||||
|
|
@ -390,6 +391,241 @@ class TestUnRapportQuiSurvitAuProcessus(unittest.TestCase):
|
|||
self.assertFalse(os.path.exists(chemin))
|
||||
|
||||
|
||||
class TestNeJamaisDetruireSousUneDescenteVivante(unittest.TestCase):
|
||||
"""Le correctif du rapport partiel a CRÉÉ ce danger.
|
||||
|
||||
Avant, la descente en cours n'avait aucun rapport sur le disque et
|
||||
« --detruire » retombait sur la précédente, terminée. Depuis qu'il s'écrit
|
||||
VM par VM, le rapport de la descente VIVANTE est le plus récent : détruire
|
||||
aurait emporté l'arbre sous le processus qui installait encore."""
|
||||
|
||||
def setUp(self):
|
||||
sys.path.insert(0, os.path.join(RACINE, "LongTest"))
|
||||
import deep_proxmox
|
||||
|
||||
self.dp = deep_proxmox
|
||||
self.maison = tempfile.mkdtemp(prefix="longtest-maison-")
|
||||
self.dossier = os.path.join(self.maison, ".erplibre/longtest")
|
||||
os.makedirs(self.dossier)
|
||||
self._vrai = os.environ.get("HOME")
|
||||
os.environ["HOME"] = self.maison
|
||||
self.addCleanup(shutil.rmtree, self.maison, ignore_errors=True)
|
||||
# Un bouchon posé par un test et non repris fausse les SUIVANTS : la
|
||||
# première version de ce fichier remplaçait dernier_rapport et le
|
||||
# laissait en place, et le test d'après lisait le bouchon.
|
||||
self._vrais = {
|
||||
nom: getattr(deep_proxmox, nom)
|
||||
for nom in ("autre_deep_proxmox", "dernier_rapport")
|
||||
}
|
||||
|
||||
def tearDown(self):
|
||||
if self._vrai is not None:
|
||||
os.environ["HOME"] = self._vrai
|
||||
for nom, vrai in self._vrais.items():
|
||||
setattr(self.dp, nom, vrai)
|
||||
|
||||
def _ecrire(self, nom, rapport):
|
||||
with open(
|
||||
os.path.join(self.dossier, nom), "w", encoding="utf-8"
|
||||
) as fh:
|
||||
json.dump(rapport, fh)
|
||||
|
||||
def test_a_living_descent_is_recognised_by_its_pid(self):
|
||||
# Ce processus-ci exécute bien un test, pas deep_proxmox.py : c'est
|
||||
# justement ce que le contrôle doit savoir distinguer.
|
||||
self.assertFalse(self.dp.descente_vivante(os.getpid()))
|
||||
self.assertFalse(self.dp.descente_vivante(None))
|
||||
self.assertFalse(self.dp.descente_vivante(999999999))
|
||||
|
||||
def test_a_shell_that_merely_names_the_script_is_not_a_descent(self):
|
||||
"""Constaté sur la machine : un « pgrep -f deep_proxmox.py » posé dans
|
||||
une boucle de surveillance donnait un shell dont la ligne de commande
|
||||
contient le motif, et deux faux positifs sur trois."""
|
||||
import subprocess
|
||||
|
||||
faux = subprocess.Popen(
|
||||
[
|
||||
"sh",
|
||||
"-c",
|
||||
"echo deep_proxmox.py --depth 10 >/dev/null; sleep 30",
|
||||
]
|
||||
)
|
||||
self.addCleanup(faux.kill)
|
||||
self.assertFalse(self.dp._lance_ce_script(faux.pid))
|
||||
self.assertNotIn(faux.pid, self.dp.autre_deep_proxmox())
|
||||
|
||||
def _fausse_descente(self):
|
||||
"""Un processus qui exécute VRAIMENT un « deep_proxmox.py ».
|
||||
|
||||
Un PID inventé ne prouverait rien : le contrôle lit /proc, et la seule
|
||||
façon honnête de l'éprouver est de lui donner un processus à voir."""
|
||||
faux = os.path.join(self.maison, "deep_proxmox.py")
|
||||
with open(faux, "w", encoding="utf-8") as fh:
|
||||
fh.write("import time\ntime.sleep(60)\n")
|
||||
proc = subprocess.Popen([sys.executable, faux])
|
||||
self.addCleanup(proc.kill)
|
||||
for _ in range(60):
|
||||
if self.dp._lance_ce_script(proc.pid):
|
||||
return proc.pid
|
||||
time.sleep(0.05)
|
||||
self.skipTest("le processus témoin n'a pas démarré")
|
||||
|
||||
def test_a_living_descent_is_seen_in_proc(self):
|
||||
pid = self._fausse_descente()
|
||||
self.assertTrue(self.dp.descente_vivante(pid))
|
||||
self.assertIn(pid, self.dp.autre_deep_proxmox())
|
||||
|
||||
def test_the_report_of_a_living_descent_is_skipped(self):
|
||||
"""Sans ce filtre, « --detruire » choisissait le rapport de la
|
||||
descente EN COURS — le plus récent — et détruisait l'arbre sous le
|
||||
processus qui installait encore."""
|
||||
pid = self._fausse_descente()
|
||||
self._ecrire(
|
||||
"deep-pve-20260101-000000.json",
|
||||
{
|
||||
"dry_run": False,
|
||||
"etages": [{"niveau": 2, "vmid": 102, "parent_alias": "a"}],
|
||||
},
|
||||
)
|
||||
self._ecrire(
|
||||
"deep-pve-20260102-000000.json",
|
||||
{
|
||||
"dry_run": False,
|
||||
"pid": pid,
|
||||
"etages": [{"niveau": 5, "vmid": 105, "parent_alias": "vif"}],
|
||||
},
|
||||
)
|
||||
with contextlib.redirect_stdout(io.StringIO()) as sortie:
|
||||
rapport = self.dp.dernier_rapport()
|
||||
# Celui de la descente vivante est écarté, et on le DIT.
|
||||
self.assertIn("descente EN COURS", sortie.getvalue())
|
||||
self.assertEqual(rapport["etages"][0]["vmid"], 102)
|
||||
|
||||
def test_an_empty_later_report_never_masks_one_that_names_vms(self):
|
||||
"""Un second lancement qui meurt à l'étage 1 — « le disque existe
|
||||
déjà » — écrivait un rapport VIDE sous un horodatage plus tardif.
|
||||
« --detruire » annonçait « 0 VM imbriquée(s) » puis effaçait le disque
|
||||
de l'étage 1, où vivaient les étages 2 et suivants : jamais arrêtés,
|
||||
jamais nommés."""
|
||||
self._ecrire(
|
||||
"deep-pve-20260101-000000.json",
|
||||
{
|
||||
"dry_run": False,
|
||||
"etages": [
|
||||
{"niveau": 3, "vmid": 103, "parent_alias": "a+b"},
|
||||
{"niveau": 2, "vmid": 102, "parent_alias": "a"},
|
||||
],
|
||||
},
|
||||
)
|
||||
self._ecrire(
|
||||
"deep-pve-20260102-000000.json", {"dry_run": False, "etages": []}
|
||||
)
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
rapport = self.dp.dernier_rapport()
|
||||
self.assertEqual(len(self.dp.a_defaire(rapport)), 2)
|
||||
self.assertTrue(rapport["fichier"].endswith("20260101-000000.json"))
|
||||
|
||||
def test_a_dry_run_report_still_never_wins(self):
|
||||
self._ecrire(
|
||||
"deep-pve-20260101-000000.json",
|
||||
{
|
||||
"dry_run": False,
|
||||
"etages": [{"niveau": 2, "vmid": 102, "parent_alias": "a"}],
|
||||
},
|
||||
)
|
||||
self._ecrire(
|
||||
"deep-pve-20260103-000000.json",
|
||||
{
|
||||
"dry_run": True,
|
||||
"etages": [{"niveau": 9, "vmid": 900, "parent_alias": "z"}],
|
||||
},
|
||||
)
|
||||
with contextlib.redirect_stdout(io.StringIO()):
|
||||
rapport = self.dp.dernier_rapport()
|
||||
self.assertEqual(rapport["etages"][0]["vmid"], 102)
|
||||
|
||||
def test_destroying_refuses_while_a_descent_runs(self):
|
||||
appels = []
|
||||
self.dp.autre_deep_proxmox = lambda: [4242]
|
||||
self.dp.dernier_rapport = lambda: appels.append("lu") or {}
|
||||
with contextlib.redirect_stdout(io.StringIO()) as sortie:
|
||||
code = self.dp.detruire(None, dry_run=False)
|
||||
self.assertEqual(code, 1)
|
||||
# Le rapport n'est même pas LU : on ne demande rien, on ne propose
|
||||
# rien, et surtout on n'attend pas un « OUI » sur un arbre vivant.
|
||||
self.assertEqual(appels, [])
|
||||
self.assertIn("descente tourne", sortie.getvalue())
|
||||
|
||||
|
||||
class TestLeDecompteDeLaDestruction(unittest.TestCase):
|
||||
"""« if not detruire_etage1(…) : faits -= 1 » — un succès de l'étage 1
|
||||
n'ajoutait RIEN, alors que le total est len(liste) + 1.
|
||||
|
||||
Le décompte était décalé de un dans TOUS les cas : une destruction
|
||||
complète annonçait « il reste des machines » et sortait 1. Le seul
|
||||
avertissement censé prévenir qu'un disque de plusieurs dizaines de Go
|
||||
reste alloué s'affichait toujours — on apprend à ne plus le lire."""
|
||||
|
||||
def setUp(self):
|
||||
sys.path.insert(0, os.path.join(RACINE, "LongTest"))
|
||||
import deep_proxmox
|
||||
|
||||
self.dp = deep_proxmox
|
||||
self._vrais = {
|
||||
nom: getattr(deep_proxmox, nom)
|
||||
for nom in (
|
||||
"autre_deep_proxmox",
|
||||
"dernier_rapport",
|
||||
"detruire_une",
|
||||
"detruire_etage1",
|
||||
)
|
||||
}
|
||||
deep_proxmox.autre_deep_proxmox = lambda: []
|
||||
deep_proxmox.dernier_rapport = lambda: {
|
||||
"fichier": "/x.json",
|
||||
"etages": [
|
||||
{"niveau": 3, "vmid": 103, "parent_alias": "a+b"},
|
||||
{"niveau": 2, "vmid": 102, "parent_alias": "a"},
|
||||
],
|
||||
}
|
||||
self._entree = (
|
||||
__builtins__["input"]
|
||||
if isinstance(__builtins__, dict)
|
||||
else __builtins__.input
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
for nom, vrai in self._vrais.items():
|
||||
setattr(self.dp, nom, vrai)
|
||||
import builtins
|
||||
|
||||
builtins.input = self._entree
|
||||
|
||||
def _lancer(self, etage1_ok):
|
||||
import builtins
|
||||
|
||||
builtins.input = lambda _prompt="": "OUI"
|
||||
self.dp.detruire_une = lambda *a, **k: True
|
||||
self.dp.detruire_etage1 = lambda *a, **k: etage1_ok
|
||||
with contextlib.redirect_stdout(io.StringIO()) as sortie:
|
||||
code = self.dp.detruire(None, dry_run=False)
|
||||
return code, sortie.getvalue()
|
||||
|
||||
def test_a_complete_destruction_reports_success(self):
|
||||
code, texte = self._lancer(etage1_ok=True)
|
||||
self.assertEqual(code, 0)
|
||||
self.assertIn("3 / 3", texte)
|
||||
self.assertNotIn("il reste des machines", texte)
|
||||
|
||||
def test_a_surviving_level_one_is_still_warned_about(self):
|
||||
# L'avertissement doit rester CRÉDIBLE : il ne sort que quand il a
|
||||
# quelque chose à dire.
|
||||
code, texte = self._lancer(etage1_ok=False)
|
||||
self.assertEqual(code, 1)
|
||||
self.assertIn("2 / 3", texte)
|
||||
self.assertIn("il reste des machines", texte)
|
||||
|
||||
|
||||
class TestLeMenu(unittest.TestCase):
|
||||
def test_the_mixin_is_wired_into_TODO(self):
|
||||
todo = TODO.__new__(TODO)
|
||||
|
|
|
|||
Loading…
Reference in a new issue