[REF] long_test : moteur commun, sûreté déclarée, sixième étape

deep_proxmox.py passe de 1245 à 474 lignes : tout ce qui ne connaît ni « qm »
ni pmxcfs vit désormais dans descente.py, prêt pour un second test long.

L'extraction a mis à nu ce qui protégeait un hôte qu'on n'a pas créé : rien.
a_defaire exigeait « vmid » et « parent_alias », deux clés que seule une
descente écrit — la protection tenait parce qu'aucun champ ne décrivait un
hôte emprunté. Un champ « cree », écrit à l'instant de la création, la rend
explicite et ferme trois portes : la liste de destruction, le repli par NOM de
detruire_etage1, et le retrait des entrées ~/.ssh/config de l'utilisateur.

Quatrième porte : le dossier des rapports est partagé. « deep_qemu --detruire »
aurait pris le rapport le plus récent, fût-il celui d'une descente Proxmox. Le
rapport porte son outil ; un rapport plus ancien, qui n'en a pas, est placé par
le préfixe de son nom de fichier plutôt que d'être rendu indéfaisable.

detruire_etage1 ne devine plus le nom de sa cible : il est obligatoire. Et une
sixième étape est née — « cet étage peut-il héberger le suivant ? » — parce que
le contrôle du stockage était celui du DÉBUT de l'étage suivant.

64 tests, les cinq garde-fous morts sous mutation. Au passage : la classe
LongTestMenuMixin, que mon renommage de répertoire avait rebaptisée
long_testMenuMixin sans qu'aucun test le voie.

--- EN ---

deep_proxmox.py drops from 1245 to 474 lines: everything that knows neither
"qm" nor pmxcfs now lives in descente.py, ready for a second long test.

The extraction laid bare what protected a host we did not create: nothing.
a_defaire required "vmid" and "parent_alias", two keys only a descent writes —
the protection held because no field described a borrowed host. A "cree" field,
written the instant a machine is created, makes it explicit and closes three
doors: the destroy list, detruire_etage1's fallback to the NAME, and the
removal of the user's own ~/.ssh/config entries.

Fourth door: the report directory is shared. "deep_qemu --detruire" would have
taken the most recent report, Proxmox's included. Reports now carry their tool;
an older one without it is placed by its filename prefix rather than made
undestroyable.

detruire_etage1 no longer guesses its target's name: it is mandatory. And a
sixth step is born — "can this level host the next?" — because the storage
check was the one at the START of the next level.

64 tests, all five guards die under mutation. Along the way: the class
LongTestMenuMixin, which my directory rename had turned into
long_testMenuMixin without any test noticing.

Assisted-by: claude-opus-5
(cherry picked from commit e1bc9ae3cfacd36a502bc88dc0789a4e86ce986b)
This commit is contained in:
Mathieu Benoit 2026-08-28 01:31:36 -04:00
parent 84ec78a61d
commit b46615f3cf
5 changed files with 1410 additions and 923 deletions

File diff suppressed because it is too large Load diff

1039
long_test/descente.py Normal file

File diff suppressed because it is too large Load diff

View file

@ -24,7 +24,7 @@ from script.todo.todo_i18n import t
LONGTEST_DIR = "long_test" LONGTEST_DIR = "long_test"
class long_testMenuMixin: class LongTestMenuMixin:
def _longtest_script(self, nom): def _longtest_script(self, nom):
"""Chemin d'un test long, ou "" s'il n'est pas là.""" """Chemin d'un test long, ou "" s'il n'est pas là."""
chemin = os.path.join(os.getcwd(), LONGTEST_DIR, nom) chemin = os.path.join(os.getcwd(), LONGTEST_DIR, nom)

View file

@ -28,7 +28,7 @@ from script.config import config_file
from script.execute import execute from script.execute import execute
from script.todo import todo_prefs from script.todo import todo_prefs
from script.todo.database_manager import DatabaseManager from script.todo.database_manager import DatabaseManager
from script.todo.longtest_menu import long_testMenuMixin from script.todo.longtest_menu import LongTestMenuMixin
from script.todo.proxmox_menu import ProxmoxMenuMixin from script.todo.proxmox_menu import ProxmoxMenuMixin
from script.todo.qemu_access import QemuAccessMixin from script.todo.qemu_access import QemuAccessMixin
from script.todo.qemu_deploy import QemuDeployMixin from script.todo.qemu_deploy import QemuDeployMixin
@ -97,7 +97,7 @@ class TODO(
QemuManageMixin, QemuManageMixin,
QemuAccessMixin, QemuAccessMixin,
ProxmoxMenuMixin, ProxmoxMenuMixin,
long_testMenuMixin, LongTestMenuMixin,
): ):
def __init__(self): def __init__(self):
self.dir_path = None self.dir_path = None

View file

@ -27,6 +27,12 @@ from script.todo.todo import TODO # noqa: E402
RACINE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) RACINE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PYTHON = os.path.join(RACINE, ".venv.erplibre/bin/python") PYTHON = os.path.join(RACINE, ".venv.erplibre/bin/python")
# Le moteur vit dans son propre module depuis qu'il est partagé entre
# deep_proxmox et deep_qemu. Bouchonner « deep_proxmox.dernier_rapport » ne
# ferait plus rien : c'est descente.detruire qui appelle descente.dernier_rapport.
sys.path.insert(0, os.path.join(RACINE, "long_test"))
import descente as moteur # noqa: E402
class TestLaFrontiere(unittest.TestCase): class TestLaFrontiere(unittest.TestCase):
"""long_test est hors de portée du lanceur unitaire, et ce n'est pas un """long_test est hors de portée du lanceur unitaire, et ce n'est pas un
@ -301,14 +307,17 @@ class TestDefaireSansEffacerAutreChose(unittest.TestCase):
{"niveau": 3, "vmid": 100, "parent_alias": "b"}, {"niveau": 3, "vmid": 100, "parent_alias": "b"},
] ]
} }
niveaux = [n for n, _p, _v, _nom in self.dp.a_defaire(rapport)] niveaux = [
n
for n, _p, _v, _nom in self.dp.a_defaire(rapport, self.dp.NOM_BASE)
]
self.assertEqual(niveaux, [4, 3, 2]) self.assertEqual(niveaux, [4, 3, 2])
def test_a_level_without_a_vmid_is_not_guessed(self): def test_a_level_without_a_vmid_is_not_guessed(self):
# Un étage abandonné avant « qm create » n'a rien créé : ne rien # Un étage abandonné avant « qm create » n'a rien créé : ne rien
# inventer à sa place. # inventer à sa place.
rapport = {"etages": [{"niveau": 2}, {"niveau": 3, "vmid": 101}]} rapport = {"etages": [{"niveau": 2}, {"niveau": 3, "vmid": 101}]}
self.assertEqual(len(self.dp.a_defaire(rapport)), 0) self.assertEqual(len(self.dp.a_defaire(rapport, self.dp.NOM_BASE)), 0)
def test_the_alias_chain_really_flattens_the_plus(self): 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. # La cause du tri mort, énoncée pour qu'on ne la réintroduise pas.
@ -394,7 +403,9 @@ class TestUnRapportQuiSurvitAuProcessus(unittest.TestCase):
d.ecrire_alias = lambda *a, **k: None d.ecrire_alias = lambda *a, **k: None
d.attendre_ssh = lambda cible, delai, parent=None: 1 d.attendre_ssh = lambda cible, delai, parent=None: 1
d.redemarrer_et_verifier = lambda cible: True d.redemarrer_et_verifier = lambda cible: True
d.reparer_pmxcfs = lambda cible: True d.remettre_debout = lambda cible: True
d.preparer_systeme = lambda cible: True
d.controler = lambda cible: True
def installer(cible): def installer(cible):
appels.append(cible) appels.append(cible)
@ -402,7 +413,7 @@ class TestUnRapportQuiSurvitAuProcessus(unittest.TestCase):
raise KeyboardInterrupt("descente tuée") raise KeyboardInterrupt("descente tuée")
return True return True
d.installer_proxmox = installer d.installer = installer
with contextlib.redirect_stdout(io.StringIO()): with contextlib.redirect_stdout(io.StringIO()):
with self.assertRaises(KeyboardInterrupt): with self.assertRaises(KeyboardInterrupt):
d.parcourir() d.parcourir()
@ -414,10 +425,10 @@ class TestUnRapportQuiSurvitAuProcessus(unittest.TestCase):
# Le couple (parent, VMID) des étages imbriqués créés : c'est de lui # Le couple (parent, VMID) des étages imbriqués créés : c'est de lui
# seul que « --detruire » se sert. # seul que « --detruire » se sert.
self.assertEqual( self.assertEqual(
self.dp.a_defaire(rapport), self.dp.a_defaire(rapport, self.dp.NOM_BASE),
[ [
(3, "deep-pve-1+deep-pve-2", 103, self.dp.nom_etage(3)), (3, "deep-pve-1+deep-pve-2", "103", self.dp.nom_etage(3)),
(2, "deep-pve-1", 102, self.dp.nom_etage(2)), (2, "deep-pve-1", "102", self.dp.nom_etage(2)),
], ],
) )
@ -427,7 +438,7 @@ class TestUnRapportQuiSurvitAuProcessus(unittest.TestCase):
ne le regardait même pas.""" ne le regardait même pas."""
rapport = self._descente_tuee(a_l_etage=1) rapport = self._descente_tuee(a_l_etage=1)
self.assertTrue(rapport["etages"]) self.assertTrue(rapport["etages"])
self.assertEqual(self.dp.a_defaire(rapport), []) self.assertEqual(self.dp.a_defaire(rapport, self.dp.NOM_BASE), [])
def test_a_partial_report_never_reads_as_a_finished_descent(self): def test_a_partial_report_never_reads_as_a_finished_descent(self):
rapport = self._descente_tuee(a_l_etage=3) rapport = self._descente_tuee(a_l_etage=3)
@ -476,7 +487,7 @@ class TestNeJamaisDetruireSousUneDescenteVivante(unittest.TestCase):
# laissait en place, et le test d'après lisait le bouchon. # laissait en place, et le test d'après lisait le bouchon.
self._vrais = { self._vrais = {
nom: getattr(deep_proxmox, nom) nom: getattr(deep_proxmox, nom)
for nom in ("autre_deep_proxmox", "dernier_rapport") for nom in ("autre_descente", "dernier_rapport")
} }
def tearDown(self): def tearDown(self):
@ -512,8 +523,8 @@ class TestNeJamaisDetruireSousUneDescenteVivante(unittest.TestCase):
] ]
) )
self.addCleanup(faux.kill) self.addCleanup(faux.kill)
self.assertFalse(self.dp._lance_ce_script(faux.pid)) self.assertFalse(self.dp._lance_une_descente(faux.pid))
self.assertNotIn(faux.pid, self.dp.autre_deep_proxmox()) self.assertNotIn(faux.pid, self.dp.autre_descente())
def _fausse_descente(self): def _fausse_descente(self):
"""Un processus qui exécute VRAIMENT un « deep_proxmox.py ». """Un processus qui exécute VRAIMENT un « deep_proxmox.py ».
@ -526,7 +537,7 @@ class TestNeJamaisDetruireSousUneDescenteVivante(unittest.TestCase):
proc = subprocess.Popen([sys.executable, faux]) proc = subprocess.Popen([sys.executable, faux])
self.addCleanup(proc.kill) self.addCleanup(proc.kill)
for _ in range(60): for _ in range(60):
if self.dp._lance_ce_script(proc.pid): if self.dp._lance_une_descente(proc.pid):
return proc.pid return proc.pid
time.sleep(0.05) time.sleep(0.05)
self.skipTest("le processus témoin n'a pas démarré") self.skipTest("le processus témoin n'a pas démarré")
@ -534,7 +545,7 @@ class TestNeJamaisDetruireSousUneDescenteVivante(unittest.TestCase):
def test_a_living_descent_is_seen_in_proc(self): def test_a_living_descent_is_seen_in_proc(self):
pid = self._fausse_descente() pid = self._fausse_descente()
self.assertTrue(self.dp.descente_vivante(pid)) self.assertTrue(self.dp.descente_vivante(pid))
self.assertIn(pid, self.dp.autre_deep_proxmox()) self.assertIn(pid, self.dp.autre_descente())
def test_the_report_of_a_living_descent_is_skipped(self): def test_the_report_of_a_living_descent_is_skipped(self):
"""Sans ce filtre, « --detruire » choisissait le rapport de la """Sans ce filtre, « --detruire » choisissait le rapport de la
@ -583,7 +594,7 @@ class TestNeJamaisDetruireSousUneDescenteVivante(unittest.TestCase):
) )
with contextlib.redirect_stdout(io.StringIO()): with contextlib.redirect_stdout(io.StringIO()):
rapport = self.dp.dernier_rapport() rapport = self.dp.dernier_rapport()
self.assertEqual(len(self.dp.a_defaire(rapport)), 2) self.assertEqual(len(self.dp.a_defaire(rapport, self.dp.NOM_BASE)), 2)
self.assertTrue(rapport["fichier"].endswith("20260101-000000.json")) self.assertTrue(rapport["fichier"].endswith("20260101-000000.json"))
def test_a_dry_run_report_still_never_wins(self): def test_a_dry_run_report_still_never_wins(self):
@ -607,10 +618,10 @@ class TestNeJamaisDetruireSousUneDescenteVivante(unittest.TestCase):
def test_destroying_refuses_while_a_descent_runs(self): def test_destroying_refuses_while_a_descent_runs(self):
appels = [] appels = []
self.dp.autre_deep_proxmox = lambda: [4242] moteur.autre_descente = lambda: [4242]
self.dp.dernier_rapport = lambda: appels.append("lu") or {} moteur.dernier_rapport = lambda outil="": appels.append("lu") or {}
with contextlib.redirect_stdout(io.StringIO()) as sortie: with contextlib.redirect_stdout(io.StringIO()) as sortie:
code = self.dp.detruire(None, dry_run=False) code = self.dp.detruire(self.dp.FAMILLE, None, dry_run=False)
self.assertEqual(code, 1) self.assertEqual(code, 1)
# Le rapport n'est même pas LU : on ne demande rien, on ne propose # 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. # rien, et surtout on n'attend pas un « OUI » sur un arbre vivant.
@ -637,10 +648,13 @@ class TestLEtage1SIdentifiePasParSonNom(unittest.TestCase):
# ferait une méthode d'instance, et « self.uuid_libvirt(nom) » # ferait une méthode d'instance, et « self.uuid_libvirt(nom) »
# passerait deux arguments à une fonction qui en prend un. La fuite # passerait deux arguments à une fonction qui en prend un. La fuite
# tombait sur les tests SUIVANTS. # tombait sur les tests SUIVANTS.
self.vrai_uuid = deep_proxmox.Descente.__dict__["uuid_libvirt"] # Le crochet vit sur la classe de BASE, dans descente.py : c'est
# elle qu'il faut détourner, pas la sous-classe Proxmox.
self.moteur = moteur
self.vrai_uuid = moteur.Descente.__dict__["uuid_libvirt"]
self.addCleanup(setattr, deep_proxmox.subprocess, "run", self.vrai_run) self.addCleanup(setattr, deep_proxmox.subprocess, "run", self.vrai_run)
self.addCleanup( self.addCleanup(
setattr, deep_proxmox.Descente, "uuid_libvirt", self.vrai_uuid setattr, moteur.Descente, "uuid_libvirt", self.vrai_uuid
) )
self.lances = [] self.lances = []
@ -656,9 +670,13 @@ class TestLEtage1SIdentifiePasParSonNom(unittest.TestCase):
def test_a_homonym_with_another_uuid_is_left_alone(self): def test_a_homonym_with_another_uuid_is_left_alone(self):
self._virsh() self._virsh()
self.dp.Descente.uuid_libvirt = staticmethod(lambda nom: "AUTRE-UUID") self.moteur.Descente.uuid_libvirt = staticmethod(
lambda nom: "AUTRE-UUID"
)
with contextlib.redirect_stdout(io.StringIO()) as sortie: with contextlib.redirect_stdout(io.StringIO()) as sortie:
res = self.dp.detruire_etage1(None, attendu="LE-NOTRE") res = self.dp.detruire_etage1(
None, "deep-pve-1", attendu="LE-NOTRE"
)
self.assertFalse(res) self.assertFalse(res)
self.assertIn("PAS notre machine", sortie.getvalue()) self.assertIn("PAS notre machine", sortie.getvalue())
# Aucun undefine, aucun destroy : seule la lecture a eu lieu. # Aucun undefine, aucun destroy : seule la lecture a eu lieu.
@ -666,9 +684,13 @@ class TestLEtage1SIdentifiePasParSonNom(unittest.TestCase):
def test_our_own_machine_is_destroyed(self): def test_our_own_machine_is_destroyed(self):
self._virsh() self._virsh()
self.dp.Descente.uuid_libvirt = staticmethod(lambda nom: "LE-NOTRE") self.moteur.Descente.uuid_libvirt = staticmethod(
lambda nom: "LE-NOTRE"
)
with contextlib.redirect_stdout(io.StringIO()): with contextlib.redirect_stdout(io.StringIO()):
res = self.dp.detruire_etage1(None, attendu="LE-NOTRE") res = self.dp.detruire_etage1(
None, "deep-pve-1", attendu="LE-NOTRE"
)
self.assertTrue(res) self.assertTrue(res)
self.assertTrue( self.assertTrue(
any( any(
@ -683,14 +705,16 @@ class TestLEtage1SIdentifiePasParSonNom(unittest.TestCase):
plutôt que de laisser croire qu'on a vérifié.""" plutôt que de laisser croire qu'on a vérifié."""
self._virsh() self._virsh()
with contextlib.redirect_stdout(io.StringIO()) as sortie: with contextlib.redirect_stdout(io.StringIO()) as sortie:
res = self.dp.detruire_etage1(None) res = self.dp.detruire_etage1(None, "deep-pve-1")
self.assertTrue(res) self.assertTrue(res)
self.assertIn("identifié par son NOM", sortie.getvalue()) self.assertIn("identifié par son NOM", sortie.getvalue())
def test_an_absent_domain_is_not_an_error(self): def test_an_absent_domain_is_not_an_error(self):
self._virsh(dominfo=1) self._virsh(dominfo=1)
with contextlib.redirect_stdout(io.StringIO()): with contextlib.redirect_stdout(io.StringIO()):
self.assertTrue(self.dp.detruire_etage1(None, attendu="X")) self.assertTrue(
self.dp.detruire_etage1(None, "deep-pve-1", attendu="X")
)
def test_the_name_comes_from_the_report_not_from_the_level(self): def test_the_name_comes_from_the_report_not_from_the_level(self):
"""Le déduire du numéro d'étage supposait que nom_etage ne changera """Le déduire du numéro d'étage supposait que nom_etage ne changera
@ -706,15 +730,15 @@ class TestLEtage1SIdentifiePasParSonNom(unittest.TestCase):
] ]
} }
self.assertEqual( self.assertEqual(
self.dp.a_defaire(rapport), self.dp.a_defaire(rapport, self.dp.NOM_BASE),
[(2, "a", 102, "nom-ecrit-a-la-creation")], [(2, "a", "102", "nom-ecrit-a-la-creation")],
) )
def test_a_report_without_a_name_falls_back_on_the_level(self): def test_a_report_without_a_name_falls_back_on_the_level(self):
rapport = {"etages": [{"niveau": 3, "vmid": 103, "parent_alias": "b"}]} rapport = {"etages": [{"niveau": 3, "vmid": 103, "parent_alias": "b"}]}
self.assertEqual( self.assertEqual(
self.dp.a_defaire(rapport), self.dp.a_defaire(rapport, self.dp.NOM_BASE),
[(3, "b", 103, self.dp.nom_etage(3))], [(3, "b", "103", self.dp.nom_etage(3))],
) )
@ -759,7 +783,7 @@ class TestLaCauseDUnMontageAbsent(unittest.TestCase):
} }
self.addCleanup(setattr, self.dp.pve, "parse_mount_wait", vrai) self.addCleanup(setattr, self.dp.pve, "parse_mount_wait", vrai)
with contextlib.redirect_stdout(io.StringIO()) as sortie: with contextlib.redirect_stdout(io.StringIO()) as sortie:
res = self.d.reparer_pmxcfs({"target": "h", "sudo": "sudo "}) res = self.d.remettre_debout({"target": "h", "sudo": "sudo "})
return res, sortie.getvalue() return res, sortie.getvalue()
def test_a_failing_unit_is_named_with_its_journal(self): def test_a_failing_unit_is_named_with_its_journal(self):
@ -889,9 +913,11 @@ class TestUneVmCreeeEstToujoursNommee(unittest.TestCase):
) )
d.ecrire_alias = lambda *a, **k: None d.ecrire_alias = lambda *a, **k: None
d.attendre_ssh = lambda cible, delai, parent=None: 1 d.attendre_ssh = lambda cible, delai, parent=None: 1
d.installer_proxmox = lambda cible: True d.installer = lambda cible: True
d.redemarrer_et_verifier = lambda cible: True d.redemarrer_et_verifier = lambda cible: True
d.reparer_pmxcfs = lambda cible: True d.remettre_debout = lambda cible: True
d.preparer_systeme = lambda cible: True
d.controler = lambda cible: True
# La création note son VMID, puis MEURT — comme « qm resize » sur un # La création note son VMID, puis MEURT — comme « qm resize » sur un
# stockage plein. # stockage plein.
@ -905,8 +931,8 @@ class TestUneVmCreeeEstToujoursNommee(unittest.TestCase):
with open(chemin, encoding="utf-8") as fh: with open(chemin, encoding="utf-8") as fh:
rapport = json.load(fh) rapport = json.load(fh)
self.assertEqual( self.assertEqual(
self.dp.a_defaire(rapport), self.dp.a_defaire(rapport, self.dp.NOM_BASE),
[(2, "deep-pve-1", 142, self.dp.nom_etage(2))], [(2, "deep-pve-1", "142", self.dp.nom_etage(2))],
) )
def test_the_vmid_is_announced_before_the_creating_commands(self): def test_the_vmid_is_announced_before_the_creating_commands(self):
@ -1003,6 +1029,187 @@ class TestNePasAttendreUneMaisonDisparue(unittest.TestCase):
self.assertEqual(self.d.attendre_ssh(enfant, 60), 0) self.assertEqual(self.d.attendre_ssh(enfant, 60), 0)
class TestDeuxPilesNeSeMelangentPas(unittest.TestCase):
"""Le dossier des rapports et le motif « *.json » sont PARTAGÉS.
Depuis qu'il y a deux tests longs, « deep_qemu --detruire » prendrait le
rapport le plus récent — pouvant être celui d'une descente Proxmox — et
lancerait « virsh undefine » d'après des VMID de Proxmox."""
def setUp(self):
self.maison = tempfile.mkdtemp(prefix="longtest-piles-")
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)
def tearDown(self):
if self._vrai is not None:
os.environ["HOME"] = self._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 _etage(self):
return [{"niveau": 2, "identite": "102", "parent_alias": "a"}]
def test_each_tool_only_sees_its_own_reports(self):
self._ecrire(
"deep-pve-20260101-000000.json",
{
"dry_run": False,
"outil": "deep_proxmox",
"etages": self._etage(),
},
)
self._ecrire(
"deep-qemu-20260102-000000.json",
{"dry_run": False, "outil": "deep_qemu", "etages": self._etage()},
)
with contextlib.redirect_stdout(io.StringIO()):
pve = moteur.dernier_rapport("deep_proxmox", "deep-pve")
qemu = moteur.dernier_rapport("deep_qemu", "deep-qemu")
self.assertTrue(
pve["fichier"].endswith("deep-pve-20260101-000000.json")
)
self.assertTrue(
qemu["fichier"].endswith("deep-qemu-20260102-000000.json")
)
def test_an_older_report_without_a_tool_is_placed_by_its_filename(self):
"""Les rapports écrits avant que ce champ existe n'ont pas d'outil.
Les refuser les rendrait indéfaisables ; les accepter sans regarder
ramènerait le danger. Le nom de fichier tranche."""
self._ecrire(
"deep-pve-20260101-000000.json",
{"dry_run": False, "etages": self._etage()},
)
with contextlib.redirect_stdout(io.StringIO()):
self.assertTrue(moteur.dernier_rapport("deep_proxmox", "deep-pve"))
self.assertFalse(moteur.dernier_rapport("deep_qemu", "deep-qemu"))
def test_a_report_is_never_handed_to_the_wrong_tool(self):
self._ecrire(
"deep-pve-20260103-000000.json",
{
"dry_run": False,
"outil": "deep_proxmox",
"etages": self._etage(),
},
)
with contextlib.redirect_stdout(io.StringIO()):
self.assertFalse(moteur.dernier_rapport("deep_qemu", "deep-qemu"))
def test_the_lock_looks_for_every_descent_script(self):
"""Deux descentes de piles différentes se disputent la RAM, le disque
et ~/.ssh/config aussi sûrement que deux de la même."""
import inspect
src = inspect.getsource(moteur._lance_une_descente)
self.assertIn("SCRIPTS", src)
self.assertIn("deep_proxmox.py", moteur.SCRIPTS)
self.assertIn("deep_qemu.py", moteur.SCRIPTS)
class TestNeDetruirePasCeQuOnNaPasCree(unittest.TestCase):
"""Une descente peut PARTIR d'une machine existante.
Ce qui protégeait jusqu'ici un hôte non créé était un effet de bord :
a_defaire exigeait deux clés que seule une descente écrit. Depuis qu'un
hôte emprunté peut figurer au rapport, il faut le DIRE."""
def setUp(self):
sys.path.insert(0, os.path.join(RACINE, "long_test"))
import deep_proxmox
self.dp = deep_proxmox
def test_a_borrowed_level_is_never_in_the_destroy_list(self):
rapport = {
"etages": [
{
"niveau": 1,
"identite": "9",
"parent_alias": "x",
"cree": False,
},
{
"niveau": 2,
"identite": "102",
"parent_alias": "a",
"cree": True,
},
]
}
niveaux = [
n for n, _p, _i, _nom in moteur.a_defaire(rapport, "deep-pve")
]
self.assertEqual(niveaux, [2])
def test_a_borrowed_root_is_not_undefined_by_name(self):
"""Une descente partie d'un hôte existant n'a JAMAIS d'UUID libvirt
local. Le repli par le nom aurait effacé un homonyme, disques
compris."""
lances = []
def faux(argv, **kw):
import types
lances.append(" ".join(argv))
return types.SimpleNamespace(returncode=0, stdout="", stderr="")
vrai = moteur.subprocess.run
moteur.subprocess.run = faux
self.addCleanup(setattr, moteur.subprocess, "run", vrai)
with contextlib.redirect_stdout(io.StringIO()) as sortie:
res = moteur.detruire_etage1(None, "machine-a-moi", cree=False)
self.assertTrue(res)
self.assertIn("pas créé par cette descente", sortie.getvalue())
self.assertFalse(
[c for c in lances if "undefine" in c or "destroy" in c], lances
)
def test_a_borrowed_alias_stays_in_the_users_ssh_config(self):
vus = {}
import script.todo.todo as module_todo
vrai = module_todo.TODO._write_ssh_config_entry
module_todo.TODO._write_ssh_config_entry = (
lambda self, host, user, ip, **kw: vus.update(
drop=kw.get("also_drop")
)
)
self.addCleanup(
setattr, module_todo.TODO, "_write_ssh_config_entry", vrai
)
with contextlib.redirect_stdout(io.StringIO()):
moteur.retirer_alias(
{
"etages": [
{"niveau": 1, "alias": "mon-proxmox", "cree": False},
{"niveau": 2, "alias": "mon-proxmox+deep-pve-2"},
]
},
nom_base="deep-pve",
)
self.assertEqual(vus["drop"], ("mon-proxmox+deep-pve-2",))
def test_destroying_the_level_one_requires_a_name(self):
"""« virsh undefine --remove-all-storage » ne devine pas sa cible. Le
repli nom_etage(1) désignait la machine numéro 1 de la pile, quelle
que soit celle dont parlait le rapport."""
import inspect
signature = inspect.signature(moteur.detruire_etage1)
self.assertIs(
signature.parameters["nom"].default, inspect.Parameter.empty
)
class TestLeDecompteDeLaDestruction(unittest.TestCase): class TestLeDecompteDeLaDestruction(unittest.TestCase):
"""« if not detruire_etage1(…) : faits -= 1 » — un succès de l'étage 1 """« if not detruire_etage1(…) : faits -= 1 » — un succès de l'étage 1
n'ajoutait RIEN, alors que le total est len(liste) + 1. n'ajoutait RIEN, alors que le total est len(liste) + 1.
@ -1017,19 +1224,26 @@ class TestLeDecompteDeLaDestruction(unittest.TestCase):
import deep_proxmox import deep_proxmox
self.dp = deep_proxmox self.dp = deep_proxmox
# Pris ET rendus sur le MOTEUR. La première version les prenait sur
# deep_proxmox et les rendait là aussi, alors qu'elle les posait sur
# descente : les bouchons fuyaient sur tous les tests suivants, qui
# inspectaient une lambda au lieu de la vraie fonction.
self._vrais = { self._vrais = {
nom: getattr(deep_proxmox, nom) nom: getattr(moteur, nom)
for nom in ( for nom in (
"autre_deep_proxmox", "autre_descente",
"dernier_rapport", "dernier_rapport",
"detruire_une",
"detruire_etage1", "detruire_etage1",
"retirer_alias", "retirer_alias",
) )
} }
deep_proxmox.autre_deep_proxmox = lambda: [] self._vraie_detruire_une = self.dp.FAMILLE.detruire_une
deep_proxmox.retirer_alias = lambda *a, **k: None self.addCleanup(
deep_proxmox.dernier_rapport = lambda: { setattr, self.dp.FAMILLE, "detruire_une", self._vraie_detruire_une
)
moteur.autre_descente = lambda: []
moteur.retirer_alias = lambda *a, **k: None
moteur.dernier_rapport = lambda outil="", prefixe="": {
"fichier": "/x.json", "fichier": "/x.json",
"etages": [ "etages": [
{"niveau": 3, "vmid": 103, "parent_alias": "a+b"}, {"niveau": 3, "vmid": 103, "parent_alias": "a+b"},
@ -1044,7 +1258,7 @@ class TestLeDecompteDeLaDestruction(unittest.TestCase):
def tearDown(self): def tearDown(self):
for nom, vrai in self._vrais.items(): for nom, vrai in self._vrais.items():
setattr(self.dp, nom, vrai) setattr(moteur, nom, vrai)
import builtins import builtins
builtins.input = self._entree builtins.input = self._entree
@ -1053,10 +1267,10 @@ class TestLeDecompteDeLaDestruction(unittest.TestCase):
import builtins import builtins
builtins.input = lambda _prompt="": "OUI" builtins.input = lambda _prompt="": "OUI"
self.dp.detruire_une = lambda *a, **k: une self.dp.FAMILLE.detruire_une = lambda *a, **k: une
self.dp.detruire_etage1 = lambda *a, **k: etage1_ok moteur.detruire_etage1 = lambda *a, **k: etage1_ok
with contextlib.redirect_stdout(io.StringIO()) as sortie: with contextlib.redirect_stdout(io.StringIO()) as sortie:
code = self.dp.detruire(None, dry_run=False) code = self.dp.detruire(self.dp.FAMILLE, None, dry_run=False)
return code, sortie.getvalue() return code, sortie.getvalue()
def test_a_complete_destruction_reports_success(self): def test_a_complete_destruction_reports_success(self):
@ -1071,7 +1285,7 @@ class TestLeDecompteDeLaDestruction(unittest.TestCase):
machines ». Or « virsh undefine --remove-all-storage » sur l'étage 1 machines ». Or « virsh undefine --remove-all-storage » sur l'étage 1
efface le disque où ils VIVENT. L'avertissement était faux dans efface le disque où ils VIVENT. L'avertissement était faux dans
l'autre sens, et un avertissement faux ne se lit plus.""" l'autre sens, et un avertissement faux ne se lit plus."""
self.dp.detruire_une = lambda *a, **k: False self.dp.FAMILLE.detruire_une = lambda *a, **k: False
code, texte = self._lancer(etage1_ok=True, une=False) code, texte = self._lancer(etage1_ok=True, une=False)
self.assertEqual(code, 0) self.assertEqual(code, 0)
self.assertIn("3 / 3", texte) self.assertIn("3 / 3", texte)
@ -1088,8 +1302,10 @@ class TestLeDecompteDeLaDestruction(unittest.TestCase):
"""Elles survivaient aux machines : des entrées mortes dont le """Elles survivaient aux machines : des entrées mortes dont le
ProxyJump désigne un hôte qui n'existe plus.""" ProxyJump désigne un hôte qui n'existe plus."""
retires = [] retires = []
self.dp.retirer_alias = lambda rapport, journal=None: retires.append( moteur.retirer_alias = (
[e.get("alias") for e in rapport["etages"]] lambda rapport, journal=None, nom_base="": retires.append(
[e.get("alias") for e in rapport["etages"]]
)
) )
self._lancer(etage1_ok=True) self._lancer(etage1_ok=True)
self.assertEqual(len(retires), 1) self.assertEqual(len(retires), 1)
@ -1113,12 +1329,13 @@ class TestLeDecompteDeLaDestruction(unittest.TestCase):
# La VRAIE fonction : setUp en a posé un bouchon pour les autres # La VRAIE fonction : setUp en a posé un bouchon pour les autres
# tests de cette classe. # tests de cette classe.
self._vrais["retirer_alias"]( self._vrais["retirer_alias"](
{ nom_base=self.dp.NOM_BASE,
rapport={
"etages": [ "etages": [
{"niveau": 1, "alias": "deep-pve-1"}, {"niveau": 1, "alias": "deep-pve-1"},
{"niveau": 2, "parent_alias": "deep-pve-1"}, {"niveau": 2, "parent_alias": "deep-pve-1"},
] ]
} },
) )
self.assertEqual( self.assertEqual(
vus["drop"], vus["drop"],