mathben/erplibre
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity
erplibre/test/test_qemu_monitor_pve.py
Mathieu Benoit 3bd9f1edad [FIX] qemu menu : nommer l'URI libvirt, sinon la liste des VM est vide 
Sans « --connect », un virsh non root vise qemu:///session : un
hyperviseur SÉPARÉ, où aucune VM du système n'existe. « list --all » y
rend une liste vide, sans erreur ni avertissement. L'URI par défaut de
root masquait l'omission tant que les commandes passaient par sudo ;
appartenir au groupe libvirt donne le droit d'atteindre qemu:///system
mais ne change pas l'URI. Les 40 appels locaux passent donc par un
constructeur unique — dont 19 en liste d'arguments, qui gardaient encore
sudo en dur.

Vérifié : une garde balaie les deux fichiers et échoue si un virsh est
écrit sans URI ; la réintroduire fait rougir.

--- EN ---

Without « --connect », a non-root virsh targets qemu:///session: a
SEPARATE hypervisor, where none of the system's VMs exist. « list --all »
returns an empty list there, with no error and no warning. Root's default
URI masked the omission as long as commands went through sudo; libvirt
group membership grants the right to reach qemu:///system but does not
change the URI. All 40 local calls therefore go through one builder —
19 of them argument lists that still hardcoded sudo.

Checked: a guard sweeps both files and fails if a virsh is written
without the URI; putting one back turns it red.

Assisted-by: Claude Opus 5
2026-09-03 05:00:55 -04:00

951 lines
38 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# © 2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
"""Les colonnes vivantes d'une VM posée sur un hôte Proxmox distant.
Elles venaient toutes de virsh : état, durée, écrit/s, RAM, disque. Or virsh
ne connaît pas les VM d'un hôte Proxmox — elles restaient donc VIDES, et le
relevé d'état les déclarait même « effacées », ce qui éteignait le reste.
L'hôte sait tout cela : « pvesh get /cluster/resources --type vm » rend
l'état, la mémoire, le disque et le cumul écrit de TOUTES ses VM en un appel.
Le relevé prend la forme exacte de celui de virsh, pour que le calcul du
débit, de la RAM et des colonnes ne sache pas d'où vient la mesure.
"""
import sys
import unittest
from unittest import mock
sys.argv = ["todo.py"]
from script.todo import qemu_install_monitor as mon # noqa: E402
# Sortie RÉELLE relevée sur l'hôte d'essai (une VM, stockage en fichiers :
# Proxmox y rapporte « disk: 0 », d'où le « du » qui suit).
REPONSE = (
'[{"cpu":0.01,"disk":0,"diskread":32633856,"diskwrite":328233472,'
'"id":"qemu/100","maxcpu":1,"maxdisk":4294967296,"maxmem":536870912,'
'"mem":385351680,"name":"pve-suivi","netin":190,"netout":0,'
'"node":"erplibre-proxmox-9","status":"running","template":0,'
'"type":"qemu","uptime":127,"vmid":100}]\n'
"---ERPLIBRE-DU---\n"
"4294971392\t/var/lib/vz/images/100/\n"
)
class TestLaLecture(unittest.TestCase):
def setUp(self):
self.releves = mon.parse_pvestats(REPONSE)
def test_the_vm_is_keyed_by_its_vmid(self):
"""Par VMID, et c'est tout le sujet.
« /cluster/resources » est bâti par pvestatd ; celui-ci arrêté, l'hôte
rend une entrée SQUELETTIQUE — ni nom, ni mémoire, ni disque, et
« status: unknown ». Indexée par nom, elle disparaissait : la VM
passait pour absente du relevé alors que l'hôte venait de la nommer.
Trois tours plus tard, 🗑 — état TERMINAL — et le suivi annonçait
« 1/1 terminées » au bout de neuf secondes. Vécu sur une VM Arch dans
un Proxmox imbriqué."""
self.assertEqual(list(self.releves), [100])
self.assertEqual(self.releves[100]["name"], "pve-suivi")
def test_the_shape_matches_the_virsh_one(self):
# Même forme exprès : `ram_pair`, `WriteWindow` et les colonnes
# fonctionnent alors sans distinguer la source.
attendus = {
"ram_used",
"ram_total",
"ram_at",
"wr_bytes",
"disk_used",
"disk_total",
}
self.assertTrue(attendus <= set(self.releves[100]))
def test_the_measures_are_the_ones_the_host_gave(self):
rec = self.releves[100]
self.assertEqual(rec["ram_used"], 385351680)
self.assertEqual(rec["ram_total"], 536870912)
self.assertEqual(rec["wr_bytes"], 328233472)
self.assertEqual(rec["state"], "running")
self.assertEqual(rec["uptime"], 127)
def test_a_zero_disk_falls_back_to_the_real_size(self):
# Sur un stockage en fichiers, Proxmox NE CALCULE PAS la taille
# occupée et rapporte 0 : la colonne aurait affiché « 0/4G ».
self.assertEqual(self.releves[100]["disk_used"], 4294971392)
self.assertEqual(self.releves[100]["disk_total"], 4294967296)
def test_the_reading_is_fresh_so_the_ram_is_shown(self):
# `ram_pair` refuse un relevé périmé : sans horodatage, la RAM d'une
# VM distante ne s'afficherait jamais.
rec = self.releves[100]
self.assertNotEqual(mon.ram_pair(rec, rec["ram_at"]), "-")
def test_garbage_yields_nothing_rather_than_raising(self):
for brut in ("", "pas du json", "{}", "---ERPLIBRE-DU---"):
self.assertEqual(mon.parse_pvestats(brut), {})
class TestLAppel(unittest.TestCase):
"""Un appel par HÔTE, mis en cache : chaque tour coûte une poignée de
main ssh (1 s mesuré) quand virsh coûte 0,03 s pour tout le parc."""
def setUp(self):
mon._PVE_CACHE.update({"at": 0.0, "stats": {}})
def _vms(self, n=2):
return [
{
"name": f"vm-{i}",
"pve": {
"target": "erplibre@pve1",
"sudo": "sudo ",
"vmid": 100 + i,
},
}
for i in range(n)
]
def test_two_vms_of_the_same_host_cost_one_call(self):
with mock.patch(
"script.proxmox.proxmox_deploy.run", return_value=(0, REPONSE)
) as appel:
mon.read_pvestats(self._vms(2), now=1000.0)
self.assertEqual(appel.call_count, 1)
def test_a_local_only_plan_calls_nothing(self):
with mock.patch("script.proxmox.proxmox_deploy.run") as appel:
self.assertEqual(mon.read_pvestats([{"name": "locale"}]), {})
appel.assert_not_called()
def test_the_cache_spares_the_next_tick(self):
with mock.patch(
"script.proxmox.proxmox_deploy.run", return_value=(0, REPONSE)
) as appel:
mon.read_pvestats(self._vms(1), now=1000.0)
mon.read_pvestats(self._vms(1), now=1000.0 + 1)
self.assertEqual(appel.call_count, 1)
# Passé l'intervalle, on redemande.
mon.read_pvestats(
self._vms(1), now=1000.0 + mon.PVE_STATS_INTERVAL + 0.1
)
self.assertEqual(appel.call_count, 2)
def test_a_failing_host_yields_nothing_rather_than_wrong_numbers(self):
with mock.patch(
"script.proxmox.proxmox_deploy.run", return_value=(255, "timeout")
):
self.assertEqual(mon.read_pvestats(self._vms(1), now=1.0), {})
class TestPasDePoubelleTropTot(unittest.TestCase):
"""« Effacée » est un état TERMINAL : la ligne gèle sur 🗑 pour de bon.
Rapporté sur une VM Arch déployée sur Proxmox : poubelle dès le premier
tour, alors que la VM venait de naître. Un relevé manquant ne prouve
rien — l'hôte peut être occupé, la VM en train de démarrer, le relevé en
cache d'avant sa création.
"""
def setUp(self):
mon._PVE_CACHE.update({"at": 0.0, "stats": {}, "ok": False})
def _vm(self):
return {
"name": "vm-a",
"pve": {
"target": "hote",
"sudo": "sudo ",
"vmid": 105,
"addr": "10.10.10.155",
},
}
def test_a_silent_host_is_not_a_deletion(self):
# (relevés, succès) : la nuance est tout le correctif.
with mock.patch(
"script.proxmox.proxmox_deploy.run", return_value=(255, "timeout")
):
stats, ok = mon.read_pvestats_detail([self._vm()], now=1.0)
self.assertEqual(stats, {})
self.assertFalse(ok)
def test_a_host_that_answers_says_so(self):
with mock.patch(
"script.proxmox.proxmox_deploy.run", return_value=(0, "[]")
):
stats, ok = mon.read_pvestats_detail([self._vm()], now=1.0)
self.assertEqual(stats, {})
self.assertTrue(ok, "l'hôte a répondu : la VM est vraiment absente")
def test_the_cache_keeps_the_verdict_too(self):
with mock.patch(
"script.proxmox.proxmox_deploy.run", return_value=(0, "[]")
) as appel:
mon.read_pvestats_detail([self._vm()], now=100.0)
_stats, ok = mon.read_pvestats_detail([self._vm()], now=100.5)
self.assertEqual(appel.call_count, 1)
self.assertTrue(ok, "le cache rendait « échec » à chaque tour suivant")
def test_it_takes_several_absences_to_conclude(self):
# Trois, pas une : le nombre est explicite, pas enfoui.
self.assertGreaterEqual(mon.PVE_ABSENCES_AVANT_EFFACEE, 2)
def test_the_short_reader_still_returns_only_stats(self):
# `read_pvestats` reste la forme courte pour la boucle des colonnes.
with mock.patch(
"script.proxmox.proxmox_deploy.run", return_value=(0, "[]")
):
self.assertEqual(mon.read_pvestats([self._vm()], now=7.0), {})
class TestLesAutresCheminsVersLaPoubelle(unittest.TestCase):
"""Trouvés par un audit, pas à l'usage : trois autres façons d'arriver au
🗑 sur un seul incident. « Effacée » gèle la ligne pour de bon, donc
chacune valait un correctif."""
def test_a_broken_pvesh_is_not_an_answer(self):
# La commande est une SUITE : son code de sortie est celui du DERNIER
# maillon. Un pvesh en panne rendait « l'hôte a répondu, la VM n'y est
# plus » — et trois tours plus tard, la poubelle.
self.assertFalse(
mon._resources_parsable("permission denied\n---ERPLIBRE-DU---\n")
)
self.assertTrue(mon._resources_parsable("[]\n---ERPLIBRE-DU---\n"))
def test_a_failing_host_never_counts_as_an_answer(self):
mon._PVE_CACHE.update({"at": 0.0, "stats": {}, "ok": False})
vm = {"name": "vm-a", "pve": {"target": "h", "sudo": "", "vmid": 1}}
with mock.patch(
"script.proxmox.proxmox_deploy.run",
return_value=(0, "sudo: a password is required\n"),
):
_stats, ok = mon.read_pvestats_detail([vm], now=10.0)
self.assertFalse(ok, "code 0 ne prouve pas que pvesh a parlé")
def test_an_unknown_proxmox_status_is_not_a_deletion(self):
# Proxmox en a d'autres que les trois attendus : « prelaunch »,
# « suspended », « internal-error », « hibernated ».
for etat in ("prelaunch", "suspended", "internal-error", "hibernated"):
self.assertIsNone(
mon.PVE_ETATS.get(etat),
"l'état n'est pas dans la table : le repli doit être « la VM"
" existe », pas « gone »",
)
def test_the_absence_counter_is_explicit(self):
self.assertGreaterEqual(mon.PVE_ABSENCES_AVANT_EFFACEE, 2)
class TestLeRedemarrageQuiFaitPartieDeLInstallation(unittest.TestCase):
"""Proxmox VE n'existe qu'après un redémarrage, et le script ne peut pas
survivre au sien.
install_proxmox.sh pose le noyau puis s'arrête, à raison — lancé par ssh,
un reboot couperait la session et ferait passer l'installation pour un
échec. La VM restait donc sur le noyau cloud de Debian, dépouillé de tout
netfilter, et on le découvrait des jours plus tard en créant un pont.
L'enveloppe, elle, tourne sur NOTRE machine : elle survit au redémarrage
de la VM. Elle redémarre, attend, vérifie le noyau, et ne conclut
qu'ensuite — le ✅ veut donc dire « hyperviseur utilisable »."""
def _joue(self, rc, faux_ssh, tours=4):
"""Exécute le shell RÉEL, avec ssh bouchonné par une fonction."""
import os
import subprocess
import tempfile
log = tempfile.NamedTemporaryFile(suffix=".log", delete=False)
log.close()
cpt = tempfile.NamedTemporaryFile("w+", delete=False)
cpt.write("0")
cpt.close()
shell = (
f"CPT={cpt.name}\n{faux_ssh}\n"
f"ip=10.0.0.1; rc={rc}; "
+ mon._reboot_steps(log.name, "-pve", tours=tours)
+ f'echo "{mon.EXIT_MARKER} $rc" >> {log.name}'
)
subprocess.run(
["bash", "-c", shell],
env=dict(os.environ, ERPLIBRE_REBOOT_SLEEP="0"),
timeout=60,
)
with open(log.name, encoding="utf-8") as fh:
texte = fh.read()
etat = mon.read_status(log.name)
os.unlink(log.name)
os.unlink(cpt.name)
return etat, texte
# Le compteur vit dans un FICHIER : « k=$(ssh …) » tourne en sous-shell.
REVIENT = """ssh() {
case "$*" in
*"uname -r"*)
n=$(cat "$CPT"); n=$((n+1)); echo "$n" > "$CPT"
if [ "$n" -ge 3 ]; then echo "7.0.14-14-pve";
else echo "6.12.101+deb13-cloud-amd64"; fi;;
*reboot*) return 255;;
esac
}"""
RESTE = """ssh() {
case "$*" in
*"uname -r"*) echo "6.12.101+deb13-cloud-amd64";;
*reboot*) return 255;;
esac
}"""
def test_only_the_pve_kernel_ends_the_wait(self):
# Le piège : sshd répond encore une seconde ou deux après l'ordre de
# redémarrage. Lire « uname -r » et s'arrêter là donnerait l'ANCIEN
# noyau en croyant avoir la réponse. Ici les deux premières lectures
# rendent le noyau cloud et doivent être REJETÉES.
(etat, code), texte = self._joue(0, self.REVIENT)
self.assertEqual((etat, code), ("done", 0))
self.assertIn("7.0.14-14-pve", texte)
def test_a_machine_that_stays_on_the_old_kernel_fails(self):
# Sans le noyau attendu, l'hyperviseur n'a ni table NAT ni module
# bridge. Le dire ✅ serait le mensonge qui a coûté deux jours.
(etat, code), texte = self._joue(0, self.RESTE)
self.assertEqual(etat, "failed")
self.assertEqual(code, 1)
self.assertIn("-pve", texte)
def test_a_machine_that_never_answers_fails(self):
(etat, _c), _texte = self._joue(0, "ssh() { return 255; }")
self.assertEqual(etat, "failed")
def test_a_failed_install_is_never_rebooted(self):
# Redémarrer après un échec effacerait la seule machine sur laquelle
# on pouvait chercher.
(etat, code), texte = self._joue(
2, 'ssh() { echo "NE DEVRAIT PAS ETRE APPELE"; }'
)
self.assertEqual((etat, code), ("failed", 2))
self.assertNotIn("NE DEVRAIT PAS", texte)
self.assertNotIn("Redémarrage", texte)
def test_which_installs_ask_for_it(self):
self.assertEqual(
mon.reboot_expected("./script/proxmox/install_proxmox.sh"), "-pve"
)
self.assertEqual(
mon.reboot_expected("make install_os && make install_odoo_18"), ""
)
self.assertEqual(mon.reboot_expected(""), "")
def test_the_wrapper_only_reboots_when_asked(self):
import inspect
src = inspect.getsource(mon._launch_one)
self.assertIn("_reboot_steps(log_q, reboot) if reboot else", src)
class TestUnHoteQuiNeNommePasSesVm(unittest.TestCase):
"""pvestatd arrêté, l'hôte rend une entrée SQUELETTIQUE par VM.
Vécu sur une VM Arch dans un Proxmox imbriqué :
{"id":"qemu/100","node":"…","status":"unknown","type":"qemu",
"vmid":100}
Le nom manque, donc la VM passait pour absente du relevé — alors que
l'hôte venait de la nommer. Trois tours plus tard : 🗑, état TERMINAL, et
le suivi annonçait « 1/1 terminées » au bout de neuf secondes sur une
installation qui tournait. Une cause, deux symptômes."""
SQUELETTE = (
'[{"id":"qemu/100","node":"n","status":"unknown","type":"qemu",'
'"vmid":100}]\n'
"---ERPLIBRE-DU---\n"
"2248339456\t/var/lib/vz/images/100/\n"
"---ERPLIBRE-ODOO---\n"
)
def _lit(self, sortie, nom="vm-arch", vmid=100):
mon._PVE_CACHE.update({"at": 0.0, "stats": {}, "ok": False})
vm = {
"name": nom,
"pve": {
"target": "h",
"sudo": "",
"vmid": vmid,
"addr": "1.2.3.4",
},
}
with mock.patch(
"script.proxmox.proxmox_deploy.run", return_value=(1, sortie)
):
return mon.read_pvestats_detail([vm], now=50.0)
def test_the_vm_is_still_found(self):
stats, ok = self._lit(self.SQUELETTE)
self.assertTrue(ok)
self.assertIn("vm-arch", stats, "trouvée par son VMID, pas par un nom")
def test_an_unknown_status_is_not_a_deletion(self):
stats, _ok = self._lit(self.SQUELETTE)
etat = stats["vm-arch"]["state"]
self.assertEqual(mon.PVE_ETATS.get(etat, "running"), "running")
def test_what_the_host_does_know_is_kept(self):
# Le « du » est indexé par VMID : la taille occupée survit même quand
# tout le reste manque.
stats, _ok = self._lit(self.SQUELETTE)
self.assertEqual(stats["vm-arch"]["disk_used"], 2248339456)
def test_a_vmid_of_another_host_is_not_borrowed(self):
# Deux hôtes peuvent avoir un VMID 100. La correspondance ne vaut que
# pour les VM de CET hôte.
mon._PVE_CACHE.update({"at": 0.0, "stats": {}, "ok": False})
vms = [
{
"name": "ici",
"pve": {"target": "h", "sudo": "", "vmid": 100},
},
{
"name": "ailleurs",
"pve": {"target": "autre", "sudo": "", "vmid": 100},
},
]
# Le bouchon répond PAR HÔTE : sans cela, la même sortie servirait
# aux deux et le test ne prouverait rien.
def par_hote(info, _cmd, _timeout=40):
if info.get("target") == "h":
return 1, self.SQUELETTE
return 1, "[]\n---ERPLIBRE-DU---\n---ERPLIBRE-ODOO---\n"
with mock.patch("script.proxmox.proxmox_deploy.run", par_hote):
stats, _ok = mon.read_pvestats_detail(vms, now=60.0)
self.assertIn("ici", stats)
self.assertNotIn("ailleurs", stats)
class TestTroisVmSurUnProxmox(unittest.TestCase):
"""Rapporté à l'usage : sur trois VM d'un même Proxmox, une seule avait
ses colonnes vides — et les deux autres montraient les chiffres d'une
AUTRE machine.
Deux fautes, dont une était le miroir d'un correctif précédent."""
def test_a_reading_stands_even_when_odoo_is_not_up_yet(self):
# Le code de sortie de la suite distante est celui de son DERNIER
# maillon : la sonde Odoo. Tant qu'Odoo n'écoute pas — c'est-à-dire
# pendant TOUTE l'installation, précisément quand on regarde — la
# boucle finit en échec et le relevé, parfait, était jeté.
#
# On avait corrigé l'erreur inverse (code 0 pris pour une réponse) ;
# exiger 0 était la même faute, retournée.
sortie = (
'[{"vmid":101,"name":"vm-a","status":"running","maxmem":1024,'
'"mem":512,"maxdisk":2048,"diskwrite":10}]\n'
"---ERPLIBRE-DU---\n---ERPLIBRE-ODOO---\n"
)
mon._PVE_CACHE.update({"at": 0.0, "stats": {}, "ok": False})
vm = {"name": "vm-a", "pve": {"target": "h", "sudo": "", "vmid": 101}}
with mock.patch(
"script.proxmox.proxmox_deploy.run", return_value=(1, sortie)
):
stats, ok = mon.read_pvestats_detail([vm], now=10.0)
self.assertTrue(
ok, "un code non nul ne réfute pas une réponse lisible"
)
self.assertIn("vm-a", stats)
def test_the_odoo_flag_survives_a_partly_closed_fleet(self):
"""Le cas rapporté, et il est le cas NORMAL.
La sonde est le dernier maillon : elle boucle sur toutes les adresses
et son code est celui de la DERNIÈRE. Un parc où une seule VM n'a pas
d'Odoo — un hyperviseur imbriqué, par exemple — finit donc en échec,
et le relevé entier partait, drapeaux Odoo compris. Le navigateur, lui,
répondait 303."""
sortie = (
'[{"vmid":100,"name":"vm-a","status":"running","maxmem":1024,'
'"mem":512,"maxdisk":2048,"diskwrite":10},'
'{"vmid":102,"name":"pve-imbrique","status":"running",'
'"maxmem":1024,"mem":512,"maxdisk":2048,"diskwrite":10}]\n'
"---ERPLIBRE-DU---\n"
"---ERPLIBRE-ODOO---\nODOO 10.10.10.150\n"
)
mon._PVE_CACHE.update({"at": 0.0, "stats": {}, "ok": False})
vms = [
{
"name": "vm-a",
"pve": {
"target": "h",
"sudo": "",
"vmid": 100,
"addr": "10.10.10.150",
},
},
{
"name": "pve-imbrique",
"pve": {
"target": "h",
"sudo": "",
"vmid": 102,
"addr": "10.10.10.152",
},
},
]
with mock.patch(
"script.proxmox.proxmox_deploy.run", return_value=(1, sortie)
):
stats, ok = mon.read_pvestats_detail(vms, now=30.0)
self.assertTrue(ok)
self.assertTrue(stats["vm-a"]["odoo"], "la VM qui répond doit être 🟢")
self.assertFalse(
stats["pve-imbrique"]["odoo"], "un hyperviseur n'a pas d'Odoo"
)
def test_a_broken_pvesh_is_still_refuted(self):
# L'autre sens tient toujours : sans liste analysable, pas de réponse.
mon._PVE_CACHE.update({"at": 0.0, "stats": {}, "ok": False})
vm = {"name": "vm-a", "pve": {"target": "h", "sudo": "", "vmid": 101}}
with mock.patch(
"script.proxmox.proxmox_deploy.run",
return_value=(0, "permission denied\n---ERPLIBRE-DU---\n"),
):
_stats, ok = mon.read_pvestats_detail([vm], now=20.0)
self.assertFalse(ok)
def test_a_remote_vm_never_borrows_a_local_namesake(self):
# « virsh domstats » indexe par NOM, et un nom se partage. Mesuré :
# « erplibre-ubuntu-2604 » sur Proxmox affichait 1,5 Gio sur 12 et
# 58 Gio de disque — ceux de la machine locale du même nom — quand la
# vraie tournait avec 3 Gio et 25.
locaux = {
"erplibre-ubuntu-2604": {
"ram_used": 1 << 30,
"ram_total": 12 << 30,
},
"erplibre-arch-latest": {"ram_used": 5, "ram_total": 9},
"vm-locale": {"ram_used": 7, "ram_total": 8},
}
vms = [
{"name": "erplibre-ubuntu-2604", "pve": {"vmid": 100}},
{"name": "vm-locale"},
]
reste = mon.drop_local_twins(dict(locaux), vms)
self.assertNotIn("erplibre-ubuntu-2604", reste)
# Une VM locale garde les siens, et une VM étrangère au manifeste
# n'est pas touchée.
self.assertIn("vm-locale", reste)
self.assertIn("erplibre-arch-latest", reste)
def test_a_silent_host_leaves_the_column_empty(self):
# Vide, c'est vrai. Une colonne vide se remarque ; une colonne juste
# et fausse, non — c'est ce qui a fait remonter le défaut.
stats = {"vm-a": {"ram_used": 1, "ram_total": 2}}
mon.drop_local_twins(stats, [{"name": "vm-a", "pve": {"vmid": 1}}])
self.assertEqual(stats, {})
class TestEffacerDepuisUnSuiviRouvert(unittest.TestCase):
"""Le suivi se ROUVRE sur un manifeste passé — c'est fait pour.
Mais un nom de domaine se réemploie, et un VMID libéré est RÉATTRIBUÉ.
Effacer « le 101 » d'un run de mars, c'est effacer ce qui porte le 101
aujourd'hui, et « erplibre-ubuntu-2604 » de mars n'est pas celui
d'aujourd'hui. Même famille que tout le reste : on jugeait sur le nom.
La commande porte donc son garde. Dans la commande et non dans l'écran :
elle protège ainsi tous ses appelants, et la vérification se fait SUR la
machine, à l'instant d'effacer."""
def test_a_proxmox_delete_checks_the_vmid_still_bears_the_name(self):
cmd = mon.delete_vm_cmd_pve(
{"target": "pve9", "vmid": 101}, name="vm-a"
)
self.assertIn("qm config 101", cmd)
self.assertIn("exit 1", cmd)
self.assertIn("qm destroy 101", cmd)
# Le garde vient AVANT la destruction, sinon il ne garde rien.
self.assertLess(cmd.index("qm config 101"), cmd.index("qm destroy"))
def test_a_local_delete_checks_the_uuid(self):
cmd = mon.delete_vm_cmd("vm-a", True, "5d55d05a-1e77")
self.assertIn("domuuid vm-a", cmd)
self.assertIn("5d55d05a-1e77", cmd)
self.assertLess(cmd.index("domuuid"), cmd.index("destroy vm-a"))
def test_an_old_manifest_without_identity_still_deletes(self):
# Un manifeste écrit avant ce correctif n'a pas d'UUID. Refuser toute
# suppression y serait une régression : on retombe sur la protection
# d'avant, la confirmation à deux mains.
cmd = mon.delete_vm_cmd("vm-a", True)
self.assertNotIn("domuuid", cmd)
self.assertIn("undefine vm-a", cmd)
sans_nom = mon.delete_vm_cmd_pve({"target": "pve9", "vmid": 101})
self.assertNotIn("qm config", sans_nom)
self.assertIn("qm destroy 101", sans_nom)
def test_the_guard_is_shell_correct(self):
"""Le garde est EXÉCUTÉ, « qm » bouchonné par une fonction shell.
Il traverse ensuite deux « shlex.quote » avant d'atteindre un dash :
chaque niveau est une occasion de le casser, et un garde cassé
s'OUVRE au lieu de se fermer. Éprouvé aussi sur le vrai hôte, où il a
refusé un nom périmé et laissé passer le bon."""
import subprocess
garde = mon.pve_identity_guard(101, "vm-a")
for vu, attendu in (("vm-a", 0), ("autre-vm", 1)):
res = subprocess.run(
[
"sh",
"-c",
f"qm() {{ echo 'name: {vu}'; }}; {garde} exit 0",
],
capture_output=True,
text=True,
)
self.assertEqual(res.returncode, attendu, f"{vu} : {res.stdout}")
self.assertIn("vm-a", garde)
class TestCeQueLaConfirmationPromet(unittest.TestCase):
"""La confirmation de suppression annonçait un fichier qcow2 local à
TOUTE VM, Proxmox comprise.
Sur une VM Proxmox ce fichier n'existe pas : son disque vit dans un
stockage que seul l'hôte connaît. La ligne désignait donc un chemin local
— au mieux inexistant, au pire celui d'une autre VM du même nom. C'est
exactement la peur qui avait fait remonter le nettoyage : « le nettoyage
risque d'effacer des VM en production »."""
def test_a_proxmox_vm_never_shows_a_local_path(self):
lignes = mon.delete_lines(
{"name": "vm-a", "pve": {"target": "pve9", "vmid": 101}}
)
texte = " ".join(lignes)
self.assertNotIn("/var/lib/libvirt", texte)
self.assertNotIn("qcow2", texte)
def test_it_names_the_host_and_the_vmid(self):
# Le nom ne suffit pas : deux VM peuvent le porter, seul le VMID est
# unique — et il faut savoir SUR QUELLE machine ça se passe.
texte = " ".join(
mon.delete_lines(
{"name": "vm-a", "pve": {"target": "pve9", "vmid": 101}}
)
)
self.assertIn("pve9", texte)
self.assertIn("101", texte)
self.assertIn("qm destroy 101", texte)
def test_a_local_vm_still_names_its_file(self):
# La voie libvirt ne régresse pas : là, le fichier EST ce qu'on efface.
texte = " ".join(mon.delete_lines({"name": "vm-a"}))
self.assertIn("/var/lib/libvirt/images/vm-a.qcow2", texte)
class TestQuandLeVertRedescend(unittest.TestCase):
"""Le 🟢 était acquis pour toujours.
« Odoo ne redescend pas en cours d'install » : c'était faux. Le service
redémarre au moins une fois — systemd l'active à la fin du make — et il
lui arrive de mourir. La VM restait verte en servant plus rien, et c'est
précisément le moment où on veut le savoir."""
def _sonde(self, reponse):
return lambda _ip, _port: reponse
def test_a_local_vm_that_stopped_answering_goes_back(self):
etat, sonde = mon.odoo_reading(
{"ip": "10.0.0.1"},
{},
deja_vert=True,
dernier=0.0,
maintenant=mon.ODOO_RECHECK + 1,
sonde=self._sonde(False),
)
self.assertIs(etat, False)
self.assertTrue(sonde)
def test_a_green_local_vm_is_not_probed_every_tick(self):
# La sonde est un connect() TCP par VM : à chaque tour (2 s) c'est
# cher pour une réponse qui ne bouge presque jamais.
etat, sonde = mon.odoo_reading(
{"ip": "10.0.0.1"},
{},
deja_vert=True,
dernier=100.0,
maintenant=101.0,
sonde=self._sonde(False),
)
self.assertIsNone(etat, "rien de neuf ne doit être affirmé")
self.assertFalse(sonde)
def test_a_red_local_vm_is_probed_every_tick(self):
_etat, sonde = mon.odoo_reading(
{"ip": "10.0.0.1"},
{},
deja_vert=False,
dernier=100.0,
maintenant=101.0,
sonde=self._sonde(True),
)
self.assertTrue(sonde)
def test_a_proxmox_vm_is_read_every_tick_for_free(self):
# Le port est testé DEPUIS L'HÔTE, avec les statistiques : d'ici, une
# adresse de pont interne ne répond jamais.
def lu(odoo):
return mon.odoo_reading(
{"pve": {"vmid": 1}},
{"odoo": odoo},
deja_vert=True,
dernier=0.0,
maintenant=1.0,
sonde=self._sonde(True),
)[0]
self.assertIs(lu(True), True)
self.assertIs(lu(False), False)
def test_a_silent_host_is_not_a_dead_odoo(self):
# Sans relevé, on ne sait RIEN : l'appelant garde le dernier état.
etat, _s = mon.odoo_reading(
{"pve": {"vmid": 1}},
{},
deja_vert=True,
dernier=0.0,
maintenant=1.0,
sonde=self._sonde(True),
)
self.assertIsNone(etat)
class TestLaBonneMachine(unittest.TestCase):
"""Le pire défaut de la série : l'installation partie AILLEURS.
Vécu le 24 août 2026. Une VM déployée sur Proxmox sous le nom
« erplibre-ubuntu-2604 » — nom déjà porté par un domaine LOCAL. Le
lanceur détaché ré-résout l'adresse de la VM à chaque tour par virsh, qui
a répondu avec le domaine local : ERPLibre + Odoo se sont installés sur la
MAUVAISE machine, et le journal l'affichait sans que rien n'alerte
(« → 192.168.123.118 »).
Pour une VM distante, l'alias ~/.ssh/config est la seule vérité : il
porte le rebond par l'hôte Proxmox.
"""
def _wrapper(self, **kw):
"""Le script du lanceur, capturé sans rien exécuter."""
vus = {}
vrai = mon.subprocess.Popen
class FauxPopen:
def __init__(self, argv, *a, **k):
vus["argv"] = argv
mon.subprocess.Popen = FauxPopen
try:
mon._launch_one("cible", "echo bonjour", "/dev/null", "vm-a", **kw)
finally:
mon.subprocess.Popen = vrai
return vus["argv"][-1]
def test_a_local_vm_still_gets_its_address_refreshed(self):
# Le bail change en cours de route (cloud-init renomme l'hôte) : la
# ré-résolution est indispensable pour une VM LOCALE.
script = self._wrapper(pve=False)
self.assertIn("virsh", script)
def test_a_proxmox_vm_is_never_re_resolved(self):
# C'est le correctif : aucun appel à virsh, donc aucun risque de
# tomber sur un domaine local homonyme.
script = self._wrapper(pve=True)
self.assertNotIn("virsh", script)
def test_the_target_stays_the_alias(self):
script = self._wrapper(pve=True)
self.assertIn("ip=cible", script)
class TestLeDisque(unittest.TestCase):
"""La colonne Disque annonçait un disque PLEIN qui ne l'était pas.
Rapporté : « 6.0G/6.0G » sur une VM dont l'invité disait « 845M utilisés
sur 5.8G ». La mesure venait de « du -sb », qui rend la taille APPARENTE :
un disque raw creux la donne entière. « du -sB1 » compte les blocs
réellement occupés — 1,2 Go, ce qui correspond.
"""
def test_the_command_counts_real_blocks(self):
self.assertIn("du -sB1", mon.PVE_STATS_CMD)
self.assertNotIn("du -sb", mon.PVE_STATS_CMD)
def test_the_measure_is_read_per_vmid(self):
texte = (
'[{"vmid":101,"name":"vm-a","disk":0,"maxdisk":6442450944,'
'"mem":1,"maxmem":2,"diskwrite":0,"status":"running","uptime":1}]\n'
"---ERPLIBRE-DU---\n"
"1268518912\t/var/lib/vz/images/101/\n"
"4294967296\t/var/lib/vz/images/999/\n"
)
rec = mon.parse_pvestats(texte)[101]
self.assertEqual(rec["disk_used"], 1268518912)
self.assertEqual(rec["disk_total"], 6442450944)
self.assertEqual(
mon.fmt_pair(rec["disk_used"], rec["disk_total"]), "1.2G/6.0G"
)
class TestOuVaLaCommande(unittest.TestCase):
"""Chaque action doit viser la BONNE machine.
Une VM distante et un domaine local peuvent porter le même nom : « s »
ouvrait la locale, la console ouvrait la console de la locale, et la pause
suspendait la locale. Le VMID et le rebond sont les seules désignations
qui ne trompent pas.
"""
LOCALE = {"name": "vm-a", "ip": "192.168.123.118"}
DISTANTE = {
"name": "vm-a",
"ip": "pve1+vm-a",
"pve": {
"target": "erplibre@pve1",
"sudo": "sudo ",
"jump": "",
"vmid": 101,
"addr": "10.10.10.151",
},
}
def test_ssh_to_a_local_vm_uses_its_address(self):
self.assertIn(
"erplibre@192.168.123.118", mon.vm_ssh_prefix(self.LOCALE)
)
self.assertNotIn("-J", mon.vm_ssh_prefix(self.LOCALE))
def test_ssh_to_a_remote_vm_goes_through_the_jump(self):
cmd = mon.vm_ssh_prefix(self.DISTANTE)
self.assertIn("-J", cmd)
self.assertIn("erplibre@pve1", cmd)
self.assertIn("erplibre@10.10.10.151", cmd)
def test_without_an_address_it_falls_back_to_the_alias(self):
vm = {"name": "vm-a", "ip": "pve1+vm-a", "pve": {"target": "pve1"}}
self.assertIn("erplibre@pve1+vm-a", mon.vm_ssh_prefix(vm))
def test_the_console_of_a_remote_vm_is_qm_terminal(self):
cmd = mon.pve_host_cmd(
self.DISTANTE["pve"], "qm terminal 101", tty=True
)
self.assertIn("ssh -t", cmd)
self.assertIn("qm terminal 101", cmd)
self.assertIn("sudo", cmd)
# Et surtout PAS virsh, qui viserait le domaine local homonyme.
self.assertNotIn("virsh", cmd)
def test_a_host_without_sudo_is_not_wrapped(self):
cmd = mon.pve_host_cmd({"target": "root@pve1", "sudo": ""}, "qm list")
self.assertNotIn("sh -c", cmd)
class TestLaColonneOdoo(unittest.TestCase):
"""Le port 8069 se teste DEPUIS L'HÔTE, dans l'appel déjà payé.
Sondé depuis le poste, il ne répond jamais pour une VM sur pont interne :
la colonne restait « — » quel que soit l'état d'Odoo. Un aller-retour ssh
de plus par tour aurait coûté une seconde ; celui des statistiques est
déjà là.
"""
def test_the_probe_rides_along_the_stats_call(self):
cmd = mon.pve_stats_cmd(["10.10.10.150", "10.10.10.151"])
self.assertIn("pvesh get /cluster/resources", cmd)
self.assertIn("/dev/tcp/$a/8069", cmd)
self.assertIn("10.10.10.151", cmd)
def test_without_addresses_nothing_is_added(self):
self.assertEqual(mon.pve_stats_cmd([]), mon.PVE_STATS_CMD)
def test_the_answer_is_read_per_address(self):
sortie = (
"[]\n---ERPLIBRE-DU---\n---ERPLIBRE-ODOO---\n"
"ODOO 10.10.10.151\n"
)
self.assertEqual(mon.parse_odoo_probe(sortie), {"10.10.10.151"})
def test_a_silent_port_yields_nothing(self):
self.assertEqual(
mon.parse_odoo_probe("[]\n---ERPLIBRE-DU---\n"), set()
)
self.assertEqual(mon.parse_odoo_probe(""), set())
def test_the_du_block_is_not_mistaken_for_a_probe(self):
# Les deux blocs se suivent : le lecteur doit prendre le bon.
sortie = (
"[]\n---ERPLIBRE-DU---\n1268518912\t/var/lib/vz/images/101/\n"
"---ERPLIBRE-ODOO---\nODOO 10.10.10.151\n"
)
self.assertEqual(mon.parse_odoo_probe(sortie), {"10.10.10.151"})
class TestLeWebEtLaSuppression(unittest.TestCase):
"""Les deux dernières actions qui visaient la mauvaise machine."""
INFO = {
"target": "erplibre-proxmox-9",
"sudo": "sudo ",
"jump": "",
"vmid": 101,
"addr": "10.10.10.151",
}
def test_the_web_view_tunnels_through_the_host(self):
argv = mon.web_tunnel_argv(self.INFO)
self.assertEqual(argv[-1], "erplibre-proxmox-9")
self.assertIn("-L", argv)
self.assertIn("18069:10.10.10.151:8069", argv)
# Pas de « -f » : le tunnel se referme par son PID, et « pkill -f »
# tuait le shell qui l'avait lancé.
self.assertNotIn("-f", argv)
def test_a_local_vm_needs_no_tunnel(self):
self.assertIsNone(mon.web_tunnel_argv(None))
self.assertIsNone(mon.web_tunnel_argv({"target": "pve1"}))
def test_the_jump_of_the_host_is_chained(self):
argv = mon.web_tunnel_argv(dict(self.INFO, jump="rebond"))
self.assertIn("-J", argv)
self.assertIn("rebond", argv)
def test_deleting_a_remote_vm_uses_its_vmid(self):
cmd = mon.delete_vm_cmd_pve(self.INFO)
self.assertIn("qm destroy 101", cmd)
self.assertIn("--purge", cmd)
# « virsh undefine <nom> » aurait effacé le domaine LOCAL homonyme.
self.assertNotIn("virsh", cmd)
def test_deleting_a_local_vm_is_unchanged(self):
cmd = mon.delete_vm_cmd("vm-a", True)
self.assertIn("undefine vm-a", cmd)
self.assertIn("/var/lib/libvirt/images/vm-a.qcow2", cmd)
class TestLEtat(unittest.TestCase):
"""Une VM absente de « virsh list » passait pour EFFACÉE."""
def test_proxmox_states_map_to_libvirt_ones(self):
self.assertEqual(mon.PVE_ETATS.get("running"), "running")
self.assertEqual(mon.PVE_ETATS.get("stopped"), "shut off")
def test_an_unknown_state_is_really_gone(self):
# Absente de la réponse de l'hôte : la VM a vraiment disparu.
self.assertIsNone(mon.PVE_ETATS.get(None))
self.assertIsNone(mon.PVE_ETATS.get("n'importe quoi"))
if __name__ == "__main__":
unittest.main(verbosity=2)
Reference in a new issue View git blame Copy permalink