mathben/erplibre
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

[FIX] qemu menu : nommer l'URI libvirt, sinon la liste des VM est vide

Browse source 
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
This commit is contained in:
Mathieu Benoit 2026-09-03 00:57:56 -04:00
parent f417a9ea86
commit 3bd9f1edad
5 changed files with 151 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
script/todo/qemu_install_monitor.py
View file

@ -23,7 +23,11 @@ import socket
import subprocess import subprocess
import time import time
from script.todo.qemu_privilege import sudo_prefix from script.todo.qemu_privilege import (
LIBVIRT_URI as URI,
sudo_prefix,
virsh_argv,
)
from pathlib import Path from pathlib import Path
try: try:
@ -1172,7 +1176,7 @@ def virsh_domstates() -> dict:
pour tout le parc (à interroger à intervalle LENT).""" pour tout le parc (à interroger à intervalle LENT)."""
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "list", "--all"], virsh_argv("list", "--all"),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -1391,7 +1395,7 @@ def read_domstats() -> str:
processus à chaque tour.""" processus à chaque tour."""
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "domstats", "--balloon", "--block"], virsh_argv("domstats", "--balloon", "--block"),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -1760,15 +1764,7 @@ def arm_balloon(names) -> None:
for name in names or (): for name in names or ():
try: try:
subprocess.run( subprocess.run(
[ virsh_argv("dommemstat", name, "--period", "5", "--live"),
"sudo",
"virsh",
"dommemstat",
name,
"--period",
"5",
"--live",
],
capture_output=True, capture_output=True,
text=True, text=True,
timeout=10, timeout=10,
@ -1976,16 +1972,18 @@ def delete_vm_cmd(name: str, with_disks: bool, uuid: str = "") -> str:
cmd = "" cmd = ""
if uuid: if uuid:
cmd = ( cmd = (
f"vu=$({sudo_prefix()}virsh domuuid {q} 2>/dev/null" f"vu=$({sudo_prefix()}virsh --connect {URI} domuuid {q}"
" 2>/dev/null"
" | tr -d '[:space:]'); " " | tr -d '[:space:]'); "
f'if [ "$vu" != {shlex.quote(uuid)} ]; then ' f'if [ "$vu" != {shlex.quote(uuid)} ]; then '
f'echo "REFUS : {name} n\'est plus le même domaine"' f'echo "REFUS : {name} n\'est plus le même domaine"'
f' "($vu). Rien n\'a été effacé."; exit 1; fi; ' f' "($vu). Rien n\'a été effacé."; exit 1; fi; '
) )
cmd += ( cmd += (
f"{sudo_prefix()}virsh destroy {q} 2>/dev/null; " f"{sudo_prefix()}virsh --connect {URI} destroy {q} 2>/dev/null; "
f"{sudo_prefix()}virsh undefine {q} --nvram 2>/dev/null " f"{sudo_prefix()}virsh --connect {URI} "
f"|| {sudo_prefix()}virsh undefine {q}" f"undefine {q} --nvram 2>/dev/null "
f"|| {sudo_prefix()}virsh --connect {URI} undefine {q}"
) )
if with_disks: if with_disks:
disk = shlex.quote(f"/var/lib/libvirt/images/{name}.qcow2") disk = shlex.quote(f"/var/lib/libvirt/images/{name}.qcow2")
@ -2920,7 +2918,10 @@ def run_monitor(manifest_path: str, run_app: bool = True):
) )
sortie = "Ctrl+O" sortie = "Ctrl+O"
else: else:
cmd = f"{sudo_prefix()}virsh console {shlex.quote(vm['name'])}" cmd = (
f"{sudo_prefix()}virsh --connect {URI} "
f"console {shlex.quote(vm['name'])}"
)
titre = f"virsh console {vm['name']}" titre = f"virsh console {vm['name']}"
sortie = "Ctrl+]" sortie = "Ctrl+]"
with self.suspend(): with self.suspend():
@ -3107,7 +3108,7 @@ def run_monitor(manifest_path: str, run_app: bool = True):
) )
else: else:
subprocess.run( subprocess.run(
["sudo", "virsh", action, nom], virsh_argv(action, nom),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=30, timeout=30,

112
script/todo/qemu_manage.py
View file

@ -13,7 +13,11 @@ import subprocess
import time import time
from script.todo import todo_install from script.todo import todo_install
from script.todo.qemu_privilege import sudo_prefix from script.todo.qemu_privilege import (
LIBVIRT_URI as URI,
sudo_prefix,
virsh_argv,
)
from script.todo.todo_i18n import t from script.todo.todo_i18n import t
@ -128,7 +132,7 @@ class QemuManageMixin:
self.execute.exec_command_live(cmd, source_erplibre=False) self.execute.exec_command_live(cmd, source_erplibre=False)
def _qemu_list_vms(self, ask_advanced=False): def _qemu_list_vms(self, ask_advanced=False):
cmd = f"{sudo_prefix()}virsh list --all" cmd = f"{sudo_prefix()}virsh --connect {URI} list --all"
print(f"{t('Will execute:')} {cmd}") print(f"{t('Will execute:')} {cmd}")
self.execute.exec_command_live(cmd, source_erplibre=False) self.execute.exec_command_live(cmd, source_erplibre=False)
if not ask_advanced: if not ask_advanced:
@ -212,7 +216,10 @@ class QemuManageMixin:
print(t("Cancelled.")) print(t("Cancelled."))
return return
for real in resolved: for real in resolved:
cmd = f"{sudo_prefix()}virsh {action} {shlex.quote(real)}" cmd = (
f"{sudo_prefix()}virsh --connect {URI} "
f"{action} {shlex.quote(real)}"
)
print(f"\n{t('Will execute:')} {cmd}") print(f"\n{t('Will execute:')} {cmd}")
self.execute.exec_command_live(cmd, source_erplibre=False) self.execute.exec_command_live(cmd, source_erplibre=False)
@ -229,7 +236,7 @@ class QemuManageMixin:
EST OUVERT, c'est elle qui compte, un disque attaché à chaud n'existant EST OUVERT, c'est elle qui compte, un disque attaché à chaud n'existant
que là. que là.
""" """
argv = ["sudo", "virsh", "dumpxml"] argv = virsh_argv("dumpxml")
if inactive: if inactive:
argv.append("--inactive") argv.append("--inactive")
argv.append(name) argv.append(name)
@ -250,7 +257,7 @@ class QemuManageMixin:
"""Démarrage automatique activé ? (absent du XML : virsh seul le sait)""" """Démarrage automatique activé ? (absent du XML : virsh seul le sait)"""
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "dominfo", name], virsh_argv("dominfo", name),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -291,15 +298,11 @@ class QemuManageMixin:
un qui contourne la gestion du réseau par libvirt. un qui contourne la gestion du réseau par libvirt.
""" """
tokens = [] tokens = []
nets = self._qemu_cmd_lines( nets = self._qemu_cmd_lines(virsh_argv("net-list", "--all", "--name"))
["sudo", "virsh", "net-list", "--all", "--name"]
)
owned = set() owned = set()
for net in nets: for net in nets:
tokens.append(f"network:{net}") tokens.append(f"network:{net}")
for line in self._qemu_cmd_lines( for line in self._qemu_cmd_lines(virsh_argv("net-info", net)):
["sudo", "virsh", "net-info", net]
):
if line.startswith("Bridge:"): if line.startswith("Bridge:"):
owned.add(line.split(":", 1)[1].strip()) owned.add(line.split(":", 1)[1].strip())
for line in self._qemu_cmd_lines( for line in self._qemu_cmd_lines(
@ -477,7 +480,7 @@ class QemuManageMixin:
"""(vcpus, max_mem_kib) via « virsh dominfo », ou (0, 0).""" """(vcpus, max_mem_kib) via « virsh dominfo », ou (0, 0)."""
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "dominfo", name], virsh_argv("dominfo", name),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -562,22 +565,14 @@ class QemuManageMixin:
disparaît au prochain démarrage du domaine.""" disparaît au prochain démarrage du domaine."""
try: try:
subprocess.run( subprocess.run(
[ virsh_argv("dommemstat", name, "--period", "5", "--live"),
"sudo",
"virsh",
"dommemstat",
name,
"--period",
"5",
"--live",
],
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
env=QemuManageMixin._qemu_c_env(), env=QemuManageMixin._qemu_c_env(),
) )
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "dommemstat", name], virsh_argv("dommemstat", name),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -678,7 +673,8 @@ class QemuManageMixin:
targets = [name] targets = [name]
for tgt in targets: for tgt in targets:
cmd = ( cmd = (
f"{sudo_prefix()}virsh domifaddr {shlex.quote(tgt)}" f"{sudo_prefix()}virsh --connect {URI} "
f"domifaddr {shlex.quote(tgt)}"
" --source lease" " --source lease"
) )
print(f"\n{t('Will execute:')} {cmd}") print(f"\n{t('Will execute:')} {cmd}")
@ -697,7 +693,9 @@ class QemuManageMixin:
print( print(
f"👤 {t('Default login (if set at deploy): erplibre / erplibre')}" f"👤 {t('Default login (if set at deploy): erplibre / erplibre')}"
) )
cmd = f"{sudo_prefix()}virsh console {shlex.quote(name)}" cmd = (
f"{sudo_prefix()}virsh --connect {URI} console {shlex.quote(name)}"
)
print(f"{t('Will execute:')} {cmd}") print(f"{t('Will execute:')} {cmd}")
self.execute.exec_command_live(cmd, source_erplibre=False) self.execute.exec_command_live(cmd, source_erplibre=False)
@ -1001,7 +999,7 @@ class QemuManageMixin:
"""État libvirt de la VM (« running », « shut off », …) ou ''.""" """État libvirt de la VM (« running », « shut off », …) ou ''."""
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "domstate", name], virsh_argv("domstate", name),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -1020,7 +1018,7 @@ class QemuManageMixin:
return name return name
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "domname", str(name)], virsh_argv("domname", str(name)),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -1042,7 +1040,8 @@ class QemuManageMixin:
# --mode acpi,agent : envoie le SIGNAL d'extinction (bouton ACPI) puis # --mode acpi,agent : envoie le SIGNAL d'extinction (bouton ACPI) puis
# tente l'agent invité si présent — plus fiable qu'un arrêt brutal. # tente l'agent invité si présent — plus fiable qu'un arrêt brutal.
cmd = ( cmd = (
f"{sudo_prefix()}virsh shutdown {shlex.quote(name)}" f"{sudo_prefix()}virsh --connect {URI} "
f"shutdown {shlex.quote(name)}"
" --mode acpi,agent" " --mode acpi,agent"
) )
print(f"{t('Will execute:')} {cmd}") print(f"{t('Will execute:')} {cmd}")
@ -1075,7 +1074,10 @@ class QemuManageMixin:
) )
) )
): ):
cmd = f"{sudo_prefix()}virsh destroy {shlex.quote(name)}" cmd = (
f"{sudo_prefix()}virsh --connect {URI} "
f"destroy {shlex.quote(name)}"
)
print(f"{t('Will execute:')} {cmd}") print(f"{t('Will execute:')} {cmd}")
self.execute.exec_command_live(cmd, source_erplibre=False) self.execute.exec_command_live(cmd, source_erplibre=False)
time.sleep(2) time.sleep(2)
@ -1088,7 +1090,7 @@ class QemuManageMixin:
ignore le seed cloud-init (…-seed.iso, en lecture seule).""" ignore le seed cloud-init (…-seed.iso, en lecture seule)."""
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "domblklist", name, "--details"], virsh_argv("domblklist", name, "--details"),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -1269,7 +1271,8 @@ class QemuManageMixin:
# Agrandissement À CHAUD : le disque virtuel grossit, le FS invité # Agrandissement À CHAUD : le disque virtuel grossit, le FS invité
# devra être étendu ensuite. # devra être étendu ensuite.
cmd = ( cmd = (
f"{sudo_prefix()}virsh blockresize {shlex.quote(name)} " f"{sudo_prefix()}virsh --connect {URI} "
f"blockresize {shlex.quote(name)} "
f"{shlex.quote(disk)} {new_gb:g}G" f"{shlex.quote(disk)} {new_gb:g}G"
) )
else: else:
@ -1315,7 +1318,10 @@ class QemuManageMixin:
if self._is_yes(input(t("Start the VM now? (y/N): "))): if self._is_yes(input(t("Start the VM now? (y/N): "))):
# `name` est déjà le nom canonique : « virsh start <id> » # `name` est déjà le nom canonique : « virsh start <id> »
# échouerait car l'ID disparaît quand la VM est éteinte. # échouerait car l'ID disparaît quand la VM est éteinte.
cmd = f"{sudo_prefix()}virsh start {shlex.quote(name)}" cmd = (
f"{sudo_prefix()}virsh --connect {URI} "
f"start {shlex.quote(name)}"
)
print(f"{t('Will execute:')} {cmd}") print(f"{t('Will execute:')} {cmd}")
self.execute.exec_command_live(cmd, source_erplibre=False) self.execute.exec_command_live(cmd, source_erplibre=False)
@ -1808,13 +1814,9 @@ class QemuManageMixin:
def agent(payload): def agent(payload):
try: try:
res = subprocess.run( res = subprocess.run(
[ virsh_argv(
"sudo", "qemu-agent-command", name, json.dumps(payload)
"virsh", ),
"qemu-agent-command",
name,
json.dumps(payload),
],
capture_output=True, capture_output=True,
text=True, text=True,
timeout=30, timeout=30,
@ -1875,7 +1877,9 @@ class QemuManageMixin:
) )
if not self._is_yes(input(t("Open the serial console now? (y/N): "))): if not self._is_yes(input(t("Open the serial console now? (y/N): "))):
return return
cmd = f"{sudo_prefix()}virsh console {shlex.quote(name)}" cmd = (
f"{sudo_prefix()}virsh --connect {URI} console {shlex.quote(name)}"
)
print(f"{t('Will execute:')} {cmd}") print(f"{t('Will execute:')} {cmd}")
self.execute.exec_command_live(cmd, source_erplibre=False) self.execute.exec_command_live(cmd, source_erplibre=False)
@ -1883,7 +1887,7 @@ class QemuManageMixin:
"""Noms des VM libvirt définies (via virsh).""" """Noms des VM libvirt définies (via virsh)."""
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "list", "--all", "--name"], virsh_argv("list", "--all", "--name"),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -1936,9 +1940,11 @@ class QemuManageMixin:
# Éteindre si en cours, puis retirer la définition (+ nvram si # Éteindre si en cours, puis retirer la définition (+ nvram si
# UEFI ; repli sans l'option pour les vieilles versions de virsh). # UEFI ; repli sans l'option pour les vieilles versions de virsh).
cmd = ( cmd = (
f"{sudo_prefix()}virsh destroy {q} 2>/dev/null; " f"{sudo_prefix()}virsh --connect {URI} "
f"{sudo_prefix()}virsh undefine {q} --nvram 2>/dev/null " f"destroy {q} 2>/dev/null; "
f"|| {sudo_prefix()}virsh undefine {q}" f"{sudo_prefix()}virsh --connect {URI} "
f"undefine {q} --nvram 2>/dev/null "
f"|| {sudo_prefix()}virsh --connect {URI} undefine {q}"
) )
if del_disks and fichiers: if del_disks and fichiers:
cmd += "; sudo rm -f " + " ".join( cmd += "; sudo rm -f " + " ".join(
@ -2009,7 +2015,7 @@ class QemuManageMixin:
for name in self._qemu_list_domains(): for name in self._qemu_list_domains():
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "domiflist", name], virsh_argv("domiflist", name),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -2128,7 +2134,7 @@ class QemuManageMixin:
for name in self._qemu_list_domains(): for name in self._qemu_list_domains():
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "domblklist", name, "--details"], virsh_argv("domblklist", name, "--details"),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -2161,9 +2167,11 @@ class QemuManageMixin:
for name in ghosts: for name in ghosts:
q = shlex.quote(name) q = shlex.quote(name)
cmd = ( cmd = (
f"{sudo_prefix()}virsh destroy {q} 2>/dev/null; " f"{sudo_prefix()}virsh --connect {URI} "
f"{sudo_prefix()}virsh undefine {q} --nvram 2>/dev/null " f"destroy {q} 2>/dev/null; "
f"|| {sudo_prefix()}virsh undefine {q}" f"{sudo_prefix()}virsh --connect {URI} "
f"undefine {q} --nvram 2>/dev/null "
f"|| {sudo_prefix()}virsh --connect {URI} undefine {q}"
) )
print(f"{t('Will execute:')} {cmd}") print(f"{t('Will execute:')} {cmd}")
self.execute.exec_command_live(cmd, source_erplibre=False) self.execute.exec_command_live(cmd, source_erplibre=False)
@ -2354,7 +2362,7 @@ class QemuManageMixin:
"""Vrai si une VM libvirt de ce nom est déjà définie.""" """Vrai si une VM libvirt de ce nom est déjà définie."""
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "dominfo", name], virsh_argv("dominfo", name),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -2398,7 +2406,7 @@ class QemuManageMixin:
for source in ("lease", "agent", "arp"): for source in ("lease", "agent", "arp"):
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "domifaddr", name, "--source", source], virsh_argv("domifaddr", name, "--source", source),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -2430,7 +2438,7 @@ class QemuManageMixin:
for source in ("lease", "agent", "arp"): for source in ("lease", "agent", "arp"):
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "domifaddr", name, "--source", source], virsh_argv("domifaddr", name, "--source", source),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,
@ -2607,7 +2615,7 @@ class QemuManageMixin:
"""Architecture d'une VM (jeton amd64/arm64/s390x) via virsh dumpxml.""" """Architecture d'une VM (jeton amd64/arm64/s390x) via virsh dumpxml."""
try: try:
res = subprocess.run( res = subprocess.run(
["sudo", "virsh", "dumpxml", name], virsh_argv("dumpxml", name),
capture_output=True, capture_output=True,
text=True, text=True,
timeout=15, timeout=15,

21
script/todo/qemu_privilege.py
View file

@ -72,6 +72,27 @@ def sudo_prefix() -> str:
return "sudo " if needs_sudo() else "" return "sudo " if needs_sudo() else ""
# L'URI ne se laisse JAMAIS implicite. Pour un utilisateur non root, libvirt
# choisit « qemu:///session », un hyperviseur séparé où AUCUNE des VM du
# système n'existe : « virsh list --all » y rend une liste vide, sans erreur
# et sans avertissement. Appartenir au groupe libvirt donne le DROIT d'accéder
# à qemu:///system, mais ne change pas l'URI par défaut. Tant que les
# commandes passaient par sudo, l'URI de root masquait l'omission.
LIBVIRT_URI = "qemu:///system"
def virsh_argv(*args: str) -> list:
"""Argv d'un virsh local : sudo si besoin, URI toujours."""
prefixe = ["sudo"] if needs_sudo() else []
return prefixe + ["virsh", "--connect", LIBVIRT_URI, *args]
def virsh_cmd(args: str = "") -> str:
"""Même chose, en chaîne pour un shell. « args » est déjà échappé."""
base = f"{sudo_prefix()}virsh --connect {LIBVIRT_URI}"
return f"{base} {args}" if args else base
def group_state(user: str = "") -> tuple[bool, bool]: def group_state(user: str = "") -> tuple[bool, bool]:
"""(déclaré, actif) pour le groupe libvirt. """(déclaré, actif) pour le groupe libvirt.

8
test/test_qemu_monitor_pve.py
View file

@ -571,9 +571,9 @@ class TestEffacerDepuisUnSuiviRouvert(unittest.TestCase):
def test_a_local_delete_checks_the_uuid(self): def test_a_local_delete_checks_the_uuid(self):
cmd = mon.delete_vm_cmd("vm-a", True, "5d55d05a-1e77") cmd = mon.delete_vm_cmd("vm-a", True, "5d55d05a-1e77")
self.assertIn("virsh domuuid vm-a", cmd) self.assertIn("domuuid vm-a", cmd)
self.assertIn("5d55d05a-1e77", cmd) self.assertIn("5d55d05a-1e77", cmd)
self.assertLess(cmd.index("domuuid"), cmd.index("virsh destroy vm-a")) self.assertLess(cmd.index("domuuid"), cmd.index("destroy vm-a"))
def test_an_old_manifest_without_identity_still_deletes(self): def test_an_old_manifest_without_identity_still_deletes(self):
# Un manifeste écrit avant ce correctif n'a pas d'UUID. Refuser toute # Un manifeste écrit avant ce correctif n'a pas d'UUID. Refuser toute
@ -581,7 +581,7 @@ class TestEffacerDepuisUnSuiviRouvert(unittest.TestCase):
# d'avant, la confirmation à deux mains. # d'avant, la confirmation à deux mains.
cmd = mon.delete_vm_cmd("vm-a", True) cmd = mon.delete_vm_cmd("vm-a", True)
self.assertNotIn("domuuid", cmd) self.assertNotIn("domuuid", cmd)
self.assertIn("virsh undefine vm-a", cmd) self.assertIn("undefine vm-a", cmd)
sans_nom = mon.delete_vm_cmd_pve({"target": "pve9", "vmid": 101}) sans_nom = mon.delete_vm_cmd_pve({"target": "pve9", "vmid": 101})
self.assertNotIn("qm config", sans_nom) self.assertNotIn("qm config", sans_nom)
self.assertIn("qm destroy 101", sans_nom) self.assertIn("qm destroy 101", sans_nom)
@ -930,7 +930,7 @@ class TestLeWebEtLaSuppression(unittest.TestCase):
def test_deleting_a_local_vm_is_unchanged(self): def test_deleting_a_local_vm_is_unchanged(self):
cmd = mon.delete_vm_cmd("vm-a", True) cmd = mon.delete_vm_cmd("vm-a", True)
self.assertIn("virsh undefine", cmd) self.assertIn("undefine vm-a", cmd)
self.assertIn("/var/lib/libvirt/images/vm-a.qcow2", cmd) self.assertIn("/var/lib/libvirt/images/vm-a.qcow2", cmd)

47
test/test_qemu_privilege.py
View file

@ -20,6 +20,7 @@ import io
import sys import sys
import unittest import unittest
from contextlib import redirect_stdout from contextlib import redirect_stdout
from pathlib import Path
from unittest import mock from unittest import mock
sys.argv = ["todo.py"] sys.argv = ["todo.py"]
@ -135,5 +136,51 @@ class AvertissementAvantInstallation(unittest.TestCase):
self.assertEqual("", self._rendu(False, True, True)) self.assertEqual("", self._rendu(False, True, True))
class LUriEstToujoursExplicite(unittest.TestCase):
"""Sans « --connect », un virsh non root vise qemu:///session.
Cet hyperviseur-là est SÉPARÉ : aucune VM du système n'y existe, et
« list --all » y rend une liste vide, sans erreur ni avertissement. Tant
que les commandes passaient par sudo, l'URI de root masquait l'omission ;
la retirer l'a mise au jour.
"""
def test_the_builder_always_names_the_uri(self):
for besoin in (True, False):
with mock.patch.object(qp, "needs_sudo", return_value=besoin):
argv = qp.virsh_argv("list", "--all")
self.assertIn("--connect", argv)
self.assertEqual(
qp.LIBVIRT_URI, argv[argv.index("--connect") + 1]
)
self.assertIn(f"--connect {qp.LIBVIRT_URI}", qp.virsh_cmd("x"))
def test_sudo_only_when_the_probe_asks_for_it(self):
with mock.patch.object(qp, "needs_sudo", return_value=False):
self.assertNotIn("sudo", qp.virsh_argv("list"))
with mock.patch.object(qp, "needs_sudo", return_value=True):
self.assertEqual("sudo", qp.virsh_argv("list")[0])
def test_no_menu_call_builds_virsh_by_hand(self):
"""Un virsh écrit à la main échapperait au constructeur, donc à
l'URI : c'est exactement ce qui vidait la liste des VM."""
for chemin in (
"script/todo/qemu_manage.py",
"script/todo/qemu_install_monitor.py",
):
source = Path(chemin).read_text(encoding="utf-8")
for num, ligne in enumerate(source.splitlines(), 1):
if '"virsh"' not in ligne and "}virsh " not in ligne:
continue
voisin = "\n".join(
source.splitlines()[max(0, num - 4) : num + 4]
)
self.assertIn(
"--connect",
voisin,
f"{chemin}:{num} appelle virsh sans URI",
)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()