From c046e027e9f52bd8371a7a5e4d3f36a507c83486 Mon Sep 17 00:00:00 2001 From: Mathieu Benoit Date: Thu, 27 Aug 2026 06:56:35 -0400 Subject: [PATCH] =?UTF-8?q?[FIX]=20LongTest=20:=20ne=20pas=20d=C3=A9truire?= =?UTF-8?q?=20sous=20une=20descente=20vivante=20;=20blocs=20ssh?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- LongTest/deep_proxmox.py | 116 ++++++++++++++++-- script/todo/todo.py | 69 ++++++++--- test/test_proxmox_form.py | 118 +++++++++++++++++++ test/test_todo_longtest.py | 236 +++++++++++++++++++++++++++++++++++++ 4 files changed, 512 insertions(+), 27 deletions(-) diff --git a/LongTest/deep_proxmox.py b/LongTest/deep_proxmox.py index 4f5fc94..76fbcdb 100755 --- a/LongTest/deep_proxmox.py +++ b/LongTest/deep_proxmox.py @@ -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)." + ( diff --git a/script/todo/todo.py b/script/todo/todo.py index 2a2dd43..0a5fb21 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -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() diff --git a/test/test_proxmox_form.py b/test/test_proxmox_form.py index 7276470..34e1b71 100644 --- a/test/test_proxmox_form.py +++ b/test/test_proxmox_form.py @@ -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 diff --git a/test/test_todo_longtest.py b/test/test_todo_longtest.py index 5be410b..0e7e043 100644 --- a/test/test_todo_longtest.py +++ b/test/test_todo_longtest.py @@ -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)