#!/usr/bin/env python3 # © 2026 TechnoLibre (http://www.technolibre.ca) # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) """Le test long QEMU-dans-QEMU, et le garde qui donne un sens à sa mesure. `deploy_qemu.py` n'échoue PAS quand KVM manque : il pose « --virt-type qemu » et crée une VM entièrement émulée, sept minutes et demie de démarrage, sans qu'aucun code de retour ne le dise. Une descente qui ne le vérifierait pas irait plus « profond » en mesurant de la TCG empilée — un chiffre plus flatteur, et faux. Ces tests-ci ne créent aucune machine : ils lisent des sorties de `virsh` et de `/sys` telles qu'elles arrivent vraiment, et vérifient ce qu'on en conclut. """ import contextlib import io import os import sys import unittest RACINE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, RACINE) sys.path.insert(0, os.path.join(RACINE, "long_test")) sys.argv = ["todo.py"] import deep_qemu # noqa: E402 class TestLireCeQueVirshEcrit(unittest.TestCase): """Des sorties RÉELLES, prises sur la machine, pas inventées.""" def test_the_address_table_has_a_header_and_a_mask(self): vrai = ( " Name MAC address Protocol Address\n" "----------------------------------------------------------\n" " vnet3 52:54:00:79:78:a4 ipv4 192.168.123.118/24\n" ) self.assertEqual(deep_qemu.parse_domifaddr(vrai), "192.168.123.118") def test_a_domain_without_a_lease_yet_gives_nothing(self): vide = ( " Name MAC address Protocol Address\n" "----------------------------------------------------------\n" " vnet0 52:54:00:aa:bb:cc N/A N/A\n" ) self.assertEqual(deep_qemu.parse_domifaddr(vide), "") self.assertEqual(deep_qemu.parse_domifaddr(""), "") self.assertEqual(deep_qemu.parse_domifaddr(None), "") def test_an_emulated_domain_is_recognised(self): emule = "\n deep-qemu-2\n" self.assertEqual(deep_qemu.parse_domaine(emule)["type"], "qemu") def test_an_accelerated_domain_and_its_cpu_mode(self): vrai = ( "\n" " deep-qemu-2\n" " \n" ) vu = deep_qemu.parse_domaine(vrai) self.assertEqual(vu["type"], "kvm") self.assertEqual(vu["cpu"], "host-passthrough") def test_an_unreadable_dumpxml_claims_nothing(self): vu = deep_qemu.parse_domaine("error: failed to get domain") self.assertEqual(vu["type"], "") self.assertEqual(vu["cpu"], "") class TestUnEtageQuiNeSaitPasHeberger(unittest.TestCase): """Ce qui n'a pas été lu vaut NON. Si /sys/module/kvm_amd/parameters/nested n'existe pas, c'est que le module n'est pas chargé — et l'étage suivant serait émulé. Traiter l'absence comme un oui rendrait le contrôle décoratif.""" def test_a_missing_nested_line_is_not_a_yes(self): vu = deep_qemu.parse_controle("KVM=oui\nDISQUE=120G\n") self.assertTrue(vu["kvm"]) self.assertFalse(vu["nested"]) self.assertEqual(vu["disque_go"], 120) def test_both_spellings_of_yes_are_accepted(self): for valeur in ("Y", "1"): with self.subTest(valeur=valeur): vu = deep_qemu.parse_controle(f"KVM=oui\nNESTED={valeur}\n") self.assertTrue(vu["nested"]) def test_nested_off_is_read_as_off(self): for valeur in ("N", "0"): with self.subTest(valeur=valeur): vu = deep_qemu.parse_controle(f"KVM=oui\nNESTED={valeur}\n") self.assertFalse(vu["nested"]) def test_no_kvm_device_at_all(self): vu = deep_qemu.parse_controle("KVM=non\nNESTED=Y\nDISQUE=50G\n") self.assertFalse(vu["kvm"]) def test_an_empty_answer_asserts_nothing(self): vu = deep_qemu.parse_controle("") self.assertFalse(vu["kvm"]) self.assertFalse(vu["nested"]) self.assertEqual(vu["disque_go"], 0) def test_the_probe_is_written_for_dash(self): """/bin/sh est dash sur Debian : « set -o pipefail » y répond « Illegal option » et sort à la première ligne.""" for interdit in ("[[", "pipefail", "$(", "&&\n"): self.assertNotIn(interdit, deep_qemu.CONTROLE_CMD, interdit) class TestLeControleArreteLaDescente(unittest.TestCase): """Un étage sans KVM ne casse pas : il bascule en émulation et continue. C'est ce silence-là que le contrôle doit rompre.""" def setUp(self): self.d = deep_qemu.Descente.__new__(deep_qemu.Descente) self.d.dry_run = False self.d.journal = None self.d.niveau_courant = 2 def _repond(self, sortie, code=0): self.d.executer = lambda h, c, delai, etiquette="", **k: (code, sortie) def test_a_level_without_kvm_stops_the_descent(self): self._repond("KVM=non\nNESTED=Y\nDISQUE=90G\n") with contextlib.redirect_stdout(io.StringIO()) as sortie: self.assertFalse(self.d.controler({"target": "h"})) self.assertIn("serait ÉMULÉ", sortie.getvalue()) def test_a_level_without_nesting_stops_the_descent(self): self._repond("KVM=oui\nNESTED=N\nDISQUE=90G\n") with contextlib.redirect_stdout(io.StringIO()) as sortie: self.assertFalse(self.d.controler({"target": "h"})) self.assertIn("imbriquée absente", sortie.getvalue()) def test_an_unreadable_probe_concludes_nothing(self): # Une lecture qui échoue ne dit pas « pas de KVM » : elle ne dit rien. self._repond("", code=255) with contextlib.redirect_stdout(io.StringIO()) as sortie: self.assertFalse(self.d.controler({"target": "h"})) self.assertIn("illisible", sortie.getvalue()) def test_a_healthy_level_passes(self): """Le contrôle NÉGATIF : sans lui, ce garde interdirait toute descente.""" self._repond("KVM=oui\nNESTED=Y\nDISQUE=90G\n") with contextlib.redirect_stdout(io.StringIO()): self.assertTrue(self.d.controler({"target": "h"})) class TestUneVmEmuleeNestPasUneMesure(unittest.TestCase): """« deploy_qemu » rend 0 en créant une VM émulée. La descente doit s'en apercevoir à la création, pas après sept minutes de démarrage.""" def setUp(self): self.d = deep_qemu.Descente.__new__(deep_qemu.Descente) self.d.dry_run = False self.d.journal = None self.d.niveau_courant = 2 self.d._envoyer_cli = lambda hote: True def _machine(self, xml, adresse=" x y ipv4 10.0.0.9/24"): def executer(hote, cmd, delai, etiquette="", **k): if "dumpxml" in cmd: return 0, xml if "domifaddr" in cmd: return 0, adresse return 0, "" self.d.executer = executer def test_an_emulated_child_is_refused(self): self._machine("x") with contextlib.redirect_stdout(io.StringIO()) as sortie: identite, adresse = self.d.creer_enfant( {"target": "p"}, 2, {"vcpu": 2, "ram": 2048, "disque": 20}, ("default",), ) self.assertIsNone(identite) self.assertIsNone(adresse) self.assertIn("ÉMULÉE", sortie.getvalue()) def test_an_accelerated_child_is_kept(self): self._machine( "x" "" ) with contextlib.redirect_stdout(io.StringIO()): identite, adresse = self.d.creer_enfant( {"target": "p"}, 2, {"vcpu": 2, "ram": 2048, "disque": 20}, ("default",), ) self.assertEqual(identite, "deep-qemu-2") self.assertEqual(adresse, "10.0.0.9") def test_the_identity_is_noted_before_anything_is_created(self): """Une création échouée à mi-chemin laisserait sinon une machine que le rapport ne nomme nulle part — et que --detruire ne peut pas défaire.""" vus = [] self._machine("") with contextlib.redirect_stdout(io.StringIO()): self.d.creer_enfant( {"target": "p"}, 4, {"vcpu": 2, "ram": 2048, "disque": 20}, ("default",), noter=vus.append, ) # Notée, alors même que la création a été REFUSÉE ensuite. self.assertEqual(vus, ["deep-qemu-4"]) def test_a_child_without_an_address_is_refused(self): self._machine("", adresse=" x y N/A N/A") with contextlib.redirect_stdout(io.StringIO()) as sortie: identite, _a = self.d.creer_enfant( {"target": "p"}, 2, {"vcpu": 2, "ram": 2048, "disque": 20}, ("default",), ) self.assertIsNone(identite) self.assertIn("sans adresse", sortie.getvalue()) class TestNeDetruireQueLeSien(unittest.TestCase): """« virsh undefine --remove-all-storage » efface un disque pour de bon.""" def setUp(self): self.vrai = deep_qemu.pve.run self.addCleanup(setattr, deep_qemu.pve, "run", self.vrai) self.lances = [] def _parent_avec(self, noms): def faux(hote, remote, timeout=120): self.lances.append(remote) if "list --all --name" in remote: return 0, "\n".join(noms) + "\n" return 0, "" deep_qemu.pve.run = faux def test_a_name_that_merely_contains_ours_is_left_alone(self): self._parent_avec(["deep-qemu-lab", "autre"]) with contextlib.redirect_stdout(io.StringIO()): res = deep_qemu.detruire_une( "p", "deep-qemu-2", "deep-qemu-2", None ) self.assertTrue(res) # absente, donc rien à faire self.assertFalse([c for c in self.lances if "undefine" in c]) def test_our_own_machine_is_stopped_then_undefined(self): self._parent_avec(["deep-qemu-2"]) with contextlib.redirect_stdout(io.StringIO()): res = deep_qemu.detruire_une( "p", "deep-qemu-2", "deep-qemu-2", None ) self.assertTrue(res) ordre = [c for c in self.lances if "destroy" in c or "undefine" in c] self.assertEqual(len(ordre), 2) self.assertIn("destroy", ordre[0]) self.assertIn("--remove-all-storage", ordre[1]) def test_an_unreachable_parent_touches_nothing(self): deep_qemu.pve.run = lambda h, r, t=120: (255, "") with contextlib.redirect_stdout(io.StringIO()) as sortie: self.assertFalse( deep_qemu.detruire_une("p", "deep-qemu-2", "deep-qemu-2", None) ) self.assertIn("rien touché", sortie.getvalue()) class TestLesDeuxTestsLongsSeRessemblent(unittest.TestCase): """Ce qui doit être identique doit l'être, et ce qui doit différer aussi.""" def setUp(self): import deep_proxmox self.pve, self.qemu = deep_proxmox, deep_qemu def test_they_share_one_engine(self): import descente for module in (self.pve, self.qemu): self.assertTrue(issubclass(module.Descente, descente.Descente)) def test_every_hook_is_implemented_by_both(self): import descente crochets = ( "preparer_parent", "creer_enfant", "installer", "noyau_convient", "remettre_debout", "controler", ) for module in (self.pve, self.qemu): for crochet in crochets: self.assertIsNot( getattr(module.Descente, crochet), getattr(descente.Descente, crochet), f"{module.OUTIL} n'implémente pas {crochet}", ) def test_they_never_share_a_name_a_tool_or_a_report(self): self.assertNotEqual(self.pve.OUTIL, self.qemu.OUTIL) self.assertNotEqual(self.pve.NOM_BASE, self.qemu.NOM_BASE) self.assertNotEqual( self.pve.FAMILLE.detruire_une, self.qemu.FAMILLE.detruire_une ) def test_the_qemu_stack_asks_for_less(self): """libvirtd seul tient dans un gibioctet là où cinq démons PVE en demandent deux.""" from script.proxmox import nesting pve = nesting.nesting_plan(3, 28, 39000, 150) qemu = nesting.nesting_plan(3, 28, 39000, 150, nesting.COUTS_QEMU) self.assertLess(qemu["niveaux"][0]["ram"], pve["niveaux"][0]["ram"]) self.assertLess( qemu["niveaux"][0]["disque"], pve["niveaux"][0]["disque"] ) if __name__ == "__main__": unittest.main()