Mathieu Benoit
612d943671
Le rapport disait ce que l'HÔTE sait faire, jamais ce que chaque VM recevra au prochain démarrage. Trois valeurs y répondent ensemble et aucune seule : le type de vidéo, « accel3d », et le device figé par libvirt, qui l'emporte sur les deux autres. La section les lit dans la définition persistante et écrit la commande qui défige, sans la lancer. Les sections Python du rapport sont désormais isolées. Le fichier s'écrit d'un bloc à la fin : une section qui lève emportait tout ce qui avait été relevé avant elle, laissant sans rapport au moment d'en avoir besoin. Le double de test rendait un objet sans « returncode », ce qui masquait le défaut. --- EN --- The report said what the HOST can do, never what each VM will get on its next boot. Three values answer that together and none alone: video type, « accel3d », and the device libvirt pinned, which outranks both. The section reads them from the persistent definition and writes out the unpinning command without running it. Python sections of the report are now isolated. The file is written in one block at the end: a section that raised took down everything gathered before it, leaving no report exactly when one is needed. The test double returned an object with no « returncode », hiding the fault. Assisted-by: Claude Opus 5
750 lines
28 KiB
Python
750 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
# © 2026 TechnoLibre (http://www.technolibre.ca)
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
|
"""Une 3D écrite dans la définition d'une VM se met à l'épreuve au démarrage.
|
|
|
|
Le nœud de rendu peut exister sans qu'EGL y démarre : QEMU refuse alors le
|
|
domaine, et la VM reste inutilisable tant que quelqu'un ne défait pas le
|
|
réglage. Rien ne permet de le savoir avant l'essai — la création a déjà son
|
|
repli, le réglage d'une VM existante avait besoin du sien.
|
|
|
|
Ce que ces tests gardent :
|
|
|
|
- l'échec EGL est reconnu au démarrage et le retrait est proposé ;
|
|
- le retrait ne touche QUE la 3D : passer « autostart » à None l'éteindrait,
|
|
et « heads » ferait naître une étape sans rapport ;
|
|
- un échec de démarrage qui n'est pas celui d'EGL n'est pas rattrapé.
|
|
"""
|
|
|
|
import io
|
|
import os
|
|
import re
|
|
import shlex
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.argv = ["todo.py"]
|
|
from script.todo import qemu_hardware as hw # noqa: E402
|
|
from script.todo import qemu_manage as qm # noqa: E402
|
|
from script.todo.todo import TODO # noqa: E402
|
|
|
|
SORTIE_EGL = (
|
|
"error: erreur interne : le processus s'est arrêté pendant la connexion"
|
|
" au moniteur: qemu-system-x86_64: egl: eglInitialize failed:"
|
|
" EGL_NOT_INITIALIZED\n"
|
|
"qemu-system-x86_64: egl: render node init failed"
|
|
)
|
|
SORTIE_AUTRE = (
|
|
"error: Failed to start domain: internal error:"
|
|
" qemu unexpectedly closed"
|
|
)
|
|
|
|
ETAT = {
|
|
"name": "vm-a",
|
|
"vcpus": 16,
|
|
"mem_mib": 32768,
|
|
"video": "virtio",
|
|
"accel3d": True,
|
|
"egl": True,
|
|
"render": "/dev/dri/renderD128",
|
|
"screen": True,
|
|
"heads": 1,
|
|
"cpu": "",
|
|
"net": "default",
|
|
"autostart": True,
|
|
}
|
|
|
|
|
|
class LaSignature(unittest.TestCase):
|
|
def setUp(self):
|
|
self.todo = TODO.__new__(TODO)
|
|
|
|
def test_it_recognises_the_egl_failure(self):
|
|
self.assertTrue(self.todo._qemu_egl_failed(SORTIE_EGL))
|
|
|
|
def test_it_ignores_another_failure(self):
|
|
self.assertFalse(self.todo._qemu_egl_failed(SORTIE_AUTRE))
|
|
|
|
def test_an_unloadable_module_does_not_raise(self):
|
|
"""Le menu ne doit pas tomber parce que la source de vérité manque."""
|
|
self.todo._qemu_import_module = mock.Mock(side_effect=OSError("x"))
|
|
self.assertFalse(self.todo._qemu_egl_failed(SORTIE_EGL))
|
|
|
|
|
|
class LeRetrait(unittest.TestCase):
|
|
def setUp(self):
|
|
self.todo = TODO.__new__(TODO)
|
|
self.lances = []
|
|
self.todo.execute = mock.MagicMock()
|
|
self.todo.execute.exec_command_live.side_effect = (
|
|
lambda cmd, **k: self.lances.append(cmd) or 0
|
|
)
|
|
self.todo._qemu_dumpxml = lambda n, **k: "<domain/>"
|
|
self.todo._qemu_autostart = lambda n: True
|
|
self.todo._qemu_import_module = lambda: _FauxModule()
|
|
|
|
def _lancer(self, sortie, reponse="o"):
|
|
with mock.patch.object(hw, "hw_state", return_value=dict(ETAT)):
|
|
with mock.patch("builtins.input", return_value=reponse):
|
|
with mock.patch("builtins.print"):
|
|
self.todo._qemu_start_failed(
|
|
"vm-a", "virsh start vm-a", sortie
|
|
)
|
|
return self.lances
|
|
|
|
def test_another_failure_is_left_alone(self):
|
|
self.assertEqual([], self._lancer(SORTIE_AUTRE))
|
|
|
|
def test_a_refusal_changes_nothing(self):
|
|
self.assertEqual([], self._lancer(SORTIE_EGL, reponse="n"))
|
|
|
|
def test_it_removes_the_3d_then_starts_again(self):
|
|
lances = self._lancer(SORTIE_EGL)
|
|
joint = " ".join(lances)
|
|
self.assertIn("accel3d=off", joint)
|
|
self.assertIn("--remove-device --graphics type=egl-headless", joint)
|
|
# Le démarrage revient EN DERNIER : le retrait doit être écrit avant.
|
|
self.assertEqual("virsh start vm-a", lances[-1])
|
|
|
|
def test_it_touches_only_the_3d(self):
|
|
"""Le test précédent passerait même si le retrait éteignait aussi le
|
|
démarrage automatique : c'est ce que fait « autostart » à None, et
|
|
seul un contrôle sur les commandes RÉELLEMENT lancées l'attrape."""
|
|
lances = self._lancer(SORTIE_EGL)
|
|
# TOUT ce qui précède le démarrage, sans filtrer sur l'outil : la
|
|
# commande d'autostart passe par virsh et non par virt-xml, donc un
|
|
# filtre sur « virt-xml » laisserait justement passer la fautive.
|
|
retraits = lances[:-1]
|
|
self.assertEqual(2, len(retraits), retraits)
|
|
joint = " ".join(retraits)
|
|
self.assertNotIn("--disable", joint)
|
|
self.assertNotIn("autostart", joint)
|
|
self.assertNotIn("model.heads", joint)
|
|
|
|
|
|
class LeRapport3D(unittest.TestCase):
|
|
"""Ce que l'hôte doit fournir pour qu'une 3D de VM démarre.
|
|
|
|
Deux briques distinctes : « egl-headless » ouvre le nœud et crée le
|
|
contexte EGL par GBM — c'est Mesa qui répond — et virglrenderer ne sert
|
|
qu'ENSUITE. Les chercher par fichier plutôt que par paquet : les noms de
|
|
paquets changent d'une distribution à l'autre, pas les emplacements.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.todo = TODO.__new__(TODO)
|
|
self.todo._qemu_host_gpu_node = lambda: "/dev/dri/renderD128"
|
|
|
|
def _rendu(self, presents):
|
|
self.todo._qemu_lib_present = lambda motif: (
|
|
f"/usr/lib/{motif}.0" if motif in presents else ""
|
|
)
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
self.todo._qemu_gpu_3d_report()
|
|
return buf.getvalue()
|
|
|
|
def test_it_names_the_two_bricks(self):
|
|
"""virglrenderer sans EGL ne sert à rien : les deux se lisent."""
|
|
rendu = self._rendu(set())
|
|
self.assertIn("virglrenderer", rendu)
|
|
self.assertIn("libEGL.so", rendu)
|
|
self.assertIn("ui-egl-headless.so", rendu)
|
|
|
|
def test_a_missing_piece_is_marked(self):
|
|
rendu = self._rendu({"libgbm.so"})
|
|
lignes = [l for l in rendu.splitlines() if "libgbm" in l]
|
|
self.assertTrue(lignes and lignes[0].strip().startswith("✅"))
|
|
manquant = [l for l in rendu.splitlines() if "virglrenderer" in l]
|
|
self.assertTrue(manquant[0].strip().startswith("❌"))
|
|
|
|
def test_it_says_the_node_alone_proves_nothing(self):
|
|
"""Le nœud existe dans le cas qui échoue : le rapport doit donc
|
|
proposer d'éprouver EGL, pas se contenter de le lister."""
|
|
rendu = self._rendu(set())
|
|
self.assertIn("/dev/dri/renderD128", rendu)
|
|
self.assertIn("eglinfo", rendu)
|
|
|
|
def test_it_changes_nothing(self):
|
|
"""Un rapport qui installerait quoi que ce soit ne serait plus un
|
|
rapport."""
|
|
self.todo.execute = mock.MagicMock()
|
|
self._rendu(set())
|
|
self.todo.execute.exec_command_live.assert_not_called()
|
|
|
|
|
|
class LeDiagnostic(unittest.TestCase):
|
|
"""Un relevé destiné à quelqu'un qui n'a pas accès à la machine.
|
|
|
|
Il doit donc être complet, ne RIEN modifier, et dire à celui qui l'envoie
|
|
ce qu'il contient — un rapport de machine porte son nom, ses chemins de
|
|
compte et ses adresses.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.todo = TODO.__new__(TODO)
|
|
self.todo._qemu_host_gpu_node = lambda: "/dev/dri/renderD128"
|
|
self.todo._qemu_lib_present = lambda motif: ""
|
|
# La proposition d'outils a ses propres tests : ici elle ne ferait
|
|
# qu'attendre une réponse que personne ne donne.
|
|
self.todo._qemu_diag_offer_tools = lambda: False
|
|
self.lances = []
|
|
|
|
def _lancer(self, tmp):
|
|
def faux_run(cmd, **k):
|
|
self.lances.append(cmd)
|
|
|
|
class R:
|
|
stdout = f"sortie de {cmd}\n"
|
|
stderr = ""
|
|
# Le vrai objet de subprocess en porte un, et les helpers
|
|
# qui appellent virsh s'en servent pour distinguer une
|
|
# sortie d'un échec. L'omettre faisait mentir le double.
|
|
returncode = 0
|
|
|
|
return R()
|
|
|
|
with mock.patch.object(
|
|
__import__("script.todo.qemu_manage", fromlist=["x"]).subprocess,
|
|
"run",
|
|
side_effect=faux_run,
|
|
), mock.patch.object(
|
|
os.path, "expanduser", return_value=tmp
|
|
), mock.patch(
|
|
"builtins.print"
|
|
):
|
|
self.todo._qemu_diagnostics()
|
|
return sorted(os.listdir(tmp))
|
|
|
|
def test_it_writes_one_readable_report(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
fichiers = self._lancer(tmp)
|
|
self.assertEqual(1, len(fichiers), fichiers)
|
|
self.assertTrue(fichiers[0].startswith("qemu-diagnostic-"))
|
|
contenu = Path(tmp, fichiers[0]).read_text(encoding="utf-8")
|
|
# Les quatre familles qui décident d'un problème QEMU : la machine,
|
|
# l'hyperviseur, le GPU, et l'interpréteur qui porte virt-xml.
|
|
# « 3D » couvre l'hôte, « 3D par VM » ce que chaque définition
|
|
# livrera : deux questions distinctes, deux sections.
|
|
for attendu in (
|
|
"uname",
|
|
"virsh",
|
|
"dri",
|
|
"virt-xml",
|
|
"===== 3D =====",
|
|
"===== 3D par VM =====",
|
|
):
|
|
self.assertIn(attendu, contenu)
|
|
|
|
# Programmes qui ne peuvent que LIRE. « command -v » cherche un outil
|
|
# sans le lancer : la présence du mot « virt-install » dans une sonde ne
|
|
# dit donc rien, seul le programme en tête de commande compte.
|
|
LECTURE_SEULE = {
|
|
"uname",
|
|
"cat",
|
|
"systemd-detect-virt",
|
|
"ls",
|
|
"lspci",
|
|
"eglinfo",
|
|
"command",
|
|
"head",
|
|
"df",
|
|
"id",
|
|
"true",
|
|
"grep",
|
|
"ps",
|
|
"sort",
|
|
"echo",
|
|
}
|
|
VIRSH_LECTURE = {"version", "list", "net-list", "dumpxml", "dominfo"}
|
|
|
|
def test_every_probe_is_read_only(self):
|
|
"""Un rapport qui modifie l'hôte n'est plus un rapport."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
self._lancer(tmp)
|
|
operateurs = {";", "|", "||", "&&"}
|
|
for cmd in self.lances:
|
|
# Le relevé lance deux formes : les sondes, chaînes passées au
|
|
# shell, et les helpers qui appellent virsh en argv-liste. Une
|
|
# liste est DÉJÀ découpée — la passer à shlex lèverait.
|
|
if isinstance(cmd, (list, tuple)):
|
|
self._juger(list(cmd), cmd)
|
|
continue
|
|
# shlex plutôt qu'un découpage sur « | » : le motif de grep en
|
|
# contient un, et le couper au milieu ferait juger « 3d » comme
|
|
# s'il était un programme.
|
|
morceau = []
|
|
for jeton in shlex.split(cmd) + [";"]:
|
|
if jeton not in operateurs:
|
|
morceau.append(jeton)
|
|
continue
|
|
if morceau:
|
|
self._juger(morceau, cmd)
|
|
morceau = []
|
|
|
|
def _juger(self, jetons, cmd):
|
|
"""Un morceau de commande ne doit que lire.
|
|
|
|
« sudo » n'est qu'un préfixe : l'admettre en tête ouvrirait la porte
|
|
à tout. On le retire et on juge le programme qu'il porte.
|
|
"""
|
|
while jetons and (
|
|
os.path.basename(jetons[0]) == "sudo" or jetons[0].startswith("-")
|
|
):
|
|
jetons = jetons[1:]
|
|
if not jetons:
|
|
return
|
|
tete = os.path.basename(jetons[0])
|
|
if tete == "python3":
|
|
return
|
|
if tete == "virsh":
|
|
sous = [
|
|
j
|
|
for j in jetons[1:]
|
|
if not j.startswith("-") and "://" not in j
|
|
]
|
|
self.assertIn(sous[0], self.VIRSH_LECTURE, cmd)
|
|
return
|
|
self.assertIn(tete, self.LECTURE_SEULE, cmd)
|
|
|
|
def test_a_hanging_probe_does_not_hold_the_report(self):
|
|
"""Une commande qui pend ne doit pas retenir le rapport : son absence
|
|
de réponse est elle-même une information."""
|
|
import subprocess as sp
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
with mock.patch.object(
|
|
__import__(
|
|
"script.todo.qemu_manage", fromlist=["x"]
|
|
).subprocess,
|
|
"run",
|
|
side_effect=sp.TimeoutExpired("x", 30),
|
|
), mock.patch.object(
|
|
os.path, "expanduser", return_value=tmp
|
|
), mock.patch(
|
|
"builtins.print"
|
|
):
|
|
self.todo._qemu_diagnostics()
|
|
contenu = Path(tmp, os.listdir(tmp)[0]).read_text(encoding="utf-8")
|
|
self.assertIn("(timeout)", contenu)
|
|
|
|
|
|
class LesOutilsDuRapport(unittest.TestCase):
|
|
"""eglinfo est le seul qui ÉPROUVE EGL : sans lui le rapport dit ce qui
|
|
est installé, jamais si ça démarre — et c'est la question posée."""
|
|
|
|
def setUp(self):
|
|
self.todo = TODO.__new__(TODO)
|
|
self.lances = []
|
|
self.todo.execute = mock.MagicMock()
|
|
self.todo.execute.exec_command_live.side_effect = (
|
|
lambda cmd, **k: self.lances.append(cmd) or 0
|
|
)
|
|
|
|
def _proposer(self, presents, reponse="o"):
|
|
"""`presents` : les outils du rapport déjà installés.
|
|
|
|
UN SEUL patch de shutil.which : « qm.shutil » et le shutil de
|
|
qemu_privilege sont le MÊME objet, et deux patchs concurrents dessus
|
|
se recouvrent — l'hôte est donc simulé en une fois, gestionnaire de
|
|
paquets compris.
|
|
"""
|
|
connus = set(presents) | {"pacman"}
|
|
|
|
def faux_which(binaire):
|
|
return "/usr/bin/x" if binaire in connus else None
|
|
|
|
with mock.patch.object(
|
|
qm.shutil, "which", side_effect=faux_which
|
|
), mock.patch("builtins.input", return_value=reponse), mock.patch(
|
|
"builtins.print"
|
|
):
|
|
return self.todo._qemu_diag_offer_tools()
|
|
|
|
def test_nothing_is_offered_when_all_are_there(self):
|
|
self.assertFalse(self._proposer({"eglinfo", "lspci"}))
|
|
self.assertEqual([], self.lances)
|
|
|
|
def test_the_package_follows_the_distribution(self):
|
|
"""« mesa-utils » n'existe pas sur Fedora, qui livre « mesa-demos »."""
|
|
self._proposer(set())
|
|
joint = " ".join(self.lances)
|
|
self.assertIn("pacman", joint)
|
|
self.assertIn("mesa-utils", joint)
|
|
self.assertIn("pciutils", joint)
|
|
|
|
def test_the_command_is_announced_before_the_question(self):
|
|
"""« Les installer ? » ne dit ni ce qui sera lancé ni avec quels
|
|
droits : la commande entière, sudo compris, précède la question."""
|
|
connus = {"pacman"}
|
|
vus = []
|
|
with mock.patch.object(
|
|
qm.shutil,
|
|
"which",
|
|
side_effect=lambda b: "/usr/bin/x" if b in connus else None,
|
|
), mock.patch(
|
|
"builtins.input",
|
|
side_effect=lambda p="": (vus.append(("?", p)), "n")[1],
|
|
), mock.patch(
|
|
"builtins.print",
|
|
side_effect=lambda *a, **k: vus.append(
|
|
("!", " ".join(str(x) for x in a))
|
|
),
|
|
):
|
|
self.todo._qemu_diag_offer_tools()
|
|
avant = [txt for genre, txt in vus[: [g for g, _ in vus].index("?")]]
|
|
joint = "\n".join(avant)
|
|
self.assertIn("sudo pacman", joint)
|
|
self.assertIn("mesa-utils", joint)
|
|
|
|
def test_a_refusal_installs_nothing(self):
|
|
self.assertFalse(self._proposer(set(), reponse="n"))
|
|
self.assertEqual([], self.lances)
|
|
|
|
def test_without_a_terminal_it_installs_nothing(self):
|
|
"""Un lancement scripté n'a personne pour répondre : l'invite y lève
|
|
EOFError, et le rapport — déjà écrit — ne doit pas tomber avec."""
|
|
connus = {"pacman"}
|
|
with mock.patch.object(
|
|
qm.shutil,
|
|
"which",
|
|
side_effect=lambda b: "/usr/bin/x" if b in connus else None,
|
|
), mock.patch("builtins.input", side_effect=EOFError), mock.patch(
|
|
"builtins.print"
|
|
):
|
|
self.assertFalse(self.todo._qemu_diag_offer_tools())
|
|
self.assertEqual([], self.lances)
|
|
|
|
def test_only_the_missing_ones_are_proposed(self):
|
|
self._proposer({"lspci"})
|
|
self.assertEqual(1, len(self.lances), self.lances)
|
|
self.assertIn("mesa-utils", self.lances[0])
|
|
|
|
|
|
class LeConseilAcl(unittest.TestCase):
|
|
"""La liste de périphériques n'est proposée que si elle explique le
|
|
blocage : hôte à carte NVIDIA propriétaire, dont la liste ne nomme pas
|
|
ses nœuds. libvirt y ajoute le nœud de rendu quand le domaine le
|
|
déclare, jamais ceux de la carte, et la pile propriétaire ouvre les deux.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.todo = TODO.__new__(TODO)
|
|
self.todo._qemu_host_gpu_node = lambda: "/dev/dri/renderD128"
|
|
|
|
def _conseil(self, nodes, acl):
|
|
vus = []
|
|
with mock.patch.object(
|
|
self.todo, "_qemu_nvidia_nodes", return_value=nodes
|
|
), mock.patch.object(
|
|
self.todo, "_qemu_acl_active", return_value=acl
|
|
), mock.patch(
|
|
"builtins.print",
|
|
side_effect=lambda *a, **k: vus.append(
|
|
" ".join(str(x) for x in a)
|
|
),
|
|
):
|
|
parle = self.todo._qemu_nvidia_acl_advice()
|
|
return parle, "\n".join(vus)
|
|
|
|
def test_a_host_without_nvidia_hears_nothing(self):
|
|
parle, rendu = self._conseil([], "")
|
|
self.assertFalse(parle)
|
|
self.assertEqual("", rendu)
|
|
|
|
def test_a_list_that_already_names_them_is_left_alone(self):
|
|
"""Répéter un conseil déjà suivi le rend invisible quand il compte."""
|
|
nodes = ["/dev/nvidia0", "/dev/nvidiactl"]
|
|
parle, _ = self._conseil(
|
|
nodes, 'cgroup_device_acl = ["/dev/nvidia0", "/dev/nvidiactl"]'
|
|
)
|
|
self.assertFalse(parle)
|
|
|
|
def test_a_missing_node_brings_the_whole_list_back(self):
|
|
"""La clé REMPLACE le défaut au lieu de s'y ajouter : proposer
|
|
seulement les nœuds manquants ferait perdre /dev/kvm."""
|
|
parle, rendu = self._conseil(
|
|
["/dev/nvidia0", "/dev/nvidiactl"],
|
|
'cgroup_device_acl = ["/dev/nvidia0"]',
|
|
)
|
|
self.assertTrue(parle)
|
|
for attendu in ("/dev/kvm", "/dev/null", "/dev/dri/renderD128"):
|
|
self.assertIn(attendu, rendu)
|
|
self.assertIn("AJOUTER", rendu)
|
|
|
|
def test_an_unreadable_file_says_so_instead_of_concluding(self):
|
|
parle, rendu = self._conseil(["/dev/nvidia0"], None)
|
|
self.assertTrue(parle)
|
|
self.assertIn("grep -n cgroup_device_acl", rendu)
|
|
|
|
def test_unreadable_and_absent_are_not_the_same_answer(self):
|
|
"""Ne pas pouvoir lire n'est pas savoir qu'il n'y a rien : le premier
|
|
cas doit rendre None, pour que le conseil le DISE au lieu de conclure
|
|
qu'aucune liste n'existe."""
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
vide = Path(tmp, "vide.conf")
|
|
vide.write_text("# tout en commentaire\n", encoding="utf-8")
|
|
with mock.patch.object(type(self.todo), "_QEMU_CONF", str(vide)):
|
|
self.assertEqual("", self.todo._qemu_acl_active())
|
|
with mock.patch.object(
|
|
type(self.todo), "_QEMU_CONF", str(Path(tmp, "absent.conf"))
|
|
):
|
|
self.assertIsNone(self.todo._qemu_acl_active())
|
|
regle = Path(tmp, "regle.conf")
|
|
regle.write_text(
|
|
'#cgroup_device_acl = ["/dev/vieux"]\n'
|
|
'cgroup_device_acl = ["/dev/nvidia0"]\n',
|
|
encoding="utf-8",
|
|
)
|
|
with mock.patch.object(type(self.todo), "_QEMU_CONF", str(regle)):
|
|
actif = self.todo._qemu_acl_active()
|
|
# La ligne en COMMENTAIRE ne compte pas : la lire ferait croire à un
|
|
# réglage que libvirt ignore.
|
|
self.assertIn("/dev/nvidia0", actif)
|
|
self.assertNotIn("/dev/vieux", actif)
|
|
|
|
def test_it_names_the_restart_and_the_vm_recreation(self):
|
|
"""Un redémarrage de l'invité ne suffit pas : la liste s'applique au
|
|
lancement de QEMU, donc il faut éteindre et rallumer."""
|
|
_, rendu = self._conseil(["/dev/nvidia0"], "")
|
|
self.assertIn("systemctl restart", rendu)
|
|
self.assertIn("is-active", rendu)
|
|
|
|
|
|
class _FauxModule:
|
|
"""deploy_qemu réduit à ce que le menu lui demande."""
|
|
|
|
@staticmethod
|
|
def egl_failed(sortie):
|
|
return "eglInitialize failed" in (sortie or "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class LaSondeVideo(unittest.TestCase):
|
|
"""La 3D d'une VM se lit sur la ligne de commande de son QEMU.
|
|
|
|
La définition dit ce qui est DEMANDÉ ; la ligne de commande dit ce qui
|
|
a été REÇU. Entre les deux, libvirt peut retirer l'accélération sans le
|
|
signaler, et « egl-headless » s'affiche dans les deux cas — chercher ce
|
|
seul mot fait conclure à tort que la 3D est en place. Le suffixe
|
|
« -gl » du device est la pièce qui tranche.
|
|
"""
|
|
|
|
ARGV = (
|
|
"/usr/bin/qemu-system-x86_64",
|
|
"-name",
|
|
"guest=une-vm,debug-threads=on",
|
|
"-device",
|
|
"%s,id=video0,max_outputs=1,bus=pcie.0",
|
|
"-display",
|
|
"egl-headless,rendernode=/dev/dri/renderD128",
|
|
"-audiodev",
|
|
'{"id":"audio1","driver":"none"}',
|
|
)
|
|
|
|
def _sonder(self, device, argv0=None, extra=()):
|
|
"""Sortie de la sonde devant un /proc bâti pour l'occasion."""
|
|
from script.todo.qemu_manage import _DIAG_VIDEO_PY
|
|
|
|
import subprocess
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
proc = Path(tmp) / "proc" / "4242"
|
|
proc.mkdir(parents=True)
|
|
argv = list(self.ARGV)
|
|
argv[0] = argv0 or argv[0]
|
|
argv[4] = argv[4] % device
|
|
argv.extend(extra)
|
|
(proc / "cmdline").write_bytes("\0".join(argv).encode() + b"\0")
|
|
programme = _DIAG_VIDEO_PY.replace(
|
|
"/proc/", str(Path(tmp) / "proc") + "/"
|
|
).replace('chemin.split("/")[2]', "chemin")
|
|
fini = subprocess.run(
|
|
[sys.executable, "-c", programme],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertEqual(fini.returncode, 0, fini.stderr)
|
|
return fini.stdout
|
|
|
|
def test_the_accelerated_device_is_reported(self):
|
|
sortie = self._sonder("virtio-vga-gl")
|
|
self.assertIn("une-vm", sortie)
|
|
self.assertIn("virtio-vga-gl", sortie)
|
|
|
|
def test_software_rendering_is_told_apart(self):
|
|
"""L'épreuve du rapport : les deux cas ne doivent PAS se lire pareil.
|
|
|
|
« egl-headless » est présent des deux côtés ; un rapport qui ne
|
|
montrerait que lui laisserait croire la 3D acquise.
|
|
"""
|
|
accelere = self._sonder("virtio-vga-gl")
|
|
logiciel = self._sonder("virtio-vga")
|
|
self.assertIn("egl-headless", accelere)
|
|
self.assertIn("egl-headless", logiciel)
|
|
self.assertNotEqual(accelere, logiciel)
|
|
self.assertNotIn("virtio-vga-gl", logiciel)
|
|
|
|
def test_a_process_merely_naming_qemu_is_ignored(self):
|
|
"""Le tri se fait sur argv[0], et non sur la ligne entière.
|
|
|
|
Un processus quelconque peut porter « qemu-system » dans ses
|
|
arguments — un pager ouvert sur un journal, et la sonde elle-même,
|
|
dont le programme contient le mot. Les compter ferait naître des
|
|
VM qui n'existent pas. Le contre-exemple porte donc le motif
|
|
AILLEURS qu'en tête, sans quoi il ne départage rien.
|
|
"""
|
|
argv = ["/var/log/qemu-system-x86_64.log"]
|
|
self.assertTrue(any("qemu-system" in a for a in argv))
|
|
self.assertEqual(
|
|
self._sonder("virtio-vga-gl", argv0="/usr/bin/less", extra=argv),
|
|
"",
|
|
)
|
|
|
|
def test_the_diagnostic_carries_the_probe(self):
|
|
from script.todo.qemu_manage import QemuManageMixin
|
|
|
|
sondes = dict(QemuManageMixin._DIAG_PROBES)
|
|
self.assertIn("video du qemu en cours", sondes)
|
|
|
|
|
|
class LesPieces3D(unittest.TestCase):
|
|
"""Le rapport 3D cherche les DEUX devices accélérés, non un seul.
|
|
|
|
« virtio-vga-gl » et « virtio-gpu-gl » sont des modules distincts, et
|
|
les distributions qui découpent QEMU les empaquettent séparément.
|
|
Celui qui sert dépend du type de vidéo demandé ; n'en chercher qu'un
|
|
rend un rapport tout vert à côté de la pièce absente.
|
|
"""
|
|
|
|
def test_both_accelerated_devices_are_looked_for(self):
|
|
from script.todo.qemu_manage import QemuManageMixin
|
|
|
|
motifs = [m for _, m in QemuManageMixin._GPU_3D_PIECES]
|
|
for module in (
|
|
"qemu/hw-display-virtio-vga-gl.so",
|
|
"qemu/hw-display-virtio-gpu-gl.so",
|
|
):
|
|
self.assertIn(module, motifs)
|
|
|
|
|
|
class La3DParVM(unittest.TestCase):
|
|
"""Le rapport dit, VM par VM, ce que le prochain démarrage livrera.
|
|
|
|
Trois valeurs y répondent ENSEMBLE et aucune seule : le type de vidéo,
|
|
« accel3d », et le device figé par libvirt. Une VM dont l'ABI est
|
|
figée sur un device sans GL tourne sans 3D quoi que demande sa
|
|
définition, et c'est le cas qu'un rapport doit nommer.
|
|
"""
|
|
|
|
XML = (
|
|
"<domain><name>{n}</name><devices><video>"
|
|
"<model type='virtio' heads='1' {attr}>"
|
|
"<acceleration accel3d='{a}'/></model></video>"
|
|
"<graphics type='egl-headless'>"
|
|
"<gl rendernode='/dev/dri/renderD128'/></graphics>"
|
|
"</devices></domain>"
|
|
)
|
|
|
|
def _rendu(self, vms):
|
|
"""Sortie du relevé pour {nom: (device figé, accel3d)}."""
|
|
faux = {
|
|
nom: self.XML.format(
|
|
n=nom,
|
|
attr=f"device='{dev}'" if dev else "",
|
|
a=accel,
|
|
)
|
|
for nom, (dev, accel) in vms.items()
|
|
}
|
|
todo = TODO.__new__(TODO)
|
|
vus = []
|
|
with mock.patch.object(
|
|
TODO, "_qemu_list_domains", lambda s: list(faux)
|
|
), mock.patch.object(
|
|
TODO,
|
|
"_qemu_dumpxml",
|
|
staticmethod(lambda n, inactive=True: faux[n]),
|
|
), mock.patch(
|
|
"builtins.print",
|
|
side_effect=lambda *a, **k: vus.append(
|
|
" ".join(str(x) for x in a)
|
|
),
|
|
):
|
|
todo._qemu_vm_3d_report()
|
|
return "\n".join(vus)
|
|
|
|
def test_a_non_gl_pin_is_named_and_explained(self):
|
|
rendu = self._rendu({"vm": ("virtio-vga", "yes")})
|
|
self.assertIn("❌", rendu)
|
|
self.assertIn("device=virtio-vga", rendu)
|
|
self.assertIn("accel3d=on", rendu)
|
|
|
|
def test_the_repair_is_written_not_run(self):
|
|
"""Un rapport ne modifie rien : la commande est ÉCRITE.
|
|
|
|
Elle passe par « define » et non par virt-xml, dont le vocabulaire
|
|
ne connaît pas partout cet attribut.
|
|
"""
|
|
rendu = self._rendu({"vm": ("virtio-vga", "yes")})
|
|
self.assertIn("virtio-vga-gl", rendu)
|
|
self.assertIn("define", rendu)
|
|
|
|
def test_a_gl_pin_is_left_alone(self):
|
|
"""Le suffixe « -gl » distingue les deux devices : sans cette
|
|
lecture, une VM correctement accélérée serait accusée à tort."""
|
|
rendu = self._rendu({"vm": ("virtio-vga-gl", "yes")})
|
|
self.assertIn("✅", rendu)
|
|
self.assertNotIn("❌", rendu)
|
|
self.assertNotIn("define", rendu)
|
|
|
|
def test_a_vm_that_never_asked_for_3d_is_not_accused(self):
|
|
rendu = self._rendu({"vm": ("virtio-vga", "no")})
|
|
self.assertNotIn("❌", rendu)
|
|
self.assertNotIn("define", rendu)
|
|
|
|
def test_only_the_frozen_one_is_counted(self):
|
|
"""Un lot mêlé : le bloc de réparation ne doit paraître qu'une
|
|
fois, et nommer une VM réellement figée."""
|
|
rendu = self._rendu(
|
|
{
|
|
"saine": ("virtio-vga-gl", "yes"),
|
|
"figee": ("virtio-vga", "yes"),
|
|
"sans": ("virtio-vga", "no"),
|
|
}
|
|
)
|
|
self.assertEqual(1, rendu.count("❌"))
|
|
self.assertEqual(1, rendu.count("define"))
|
|
self.assertIn("vm=figee", rendu)
|
|
|
|
|
|
class LesSectionsDuRapport(unittest.TestCase):
|
|
"""Le rapport s'écrit d'un bloc à la FIN.
|
|
|
|
Une section qui lève emporterait donc tout ce qui a été relevé avant
|
|
elle, et l'utilisateur se retrouverait sans fichier au moment précis
|
|
où il en a besoin.
|
|
"""
|
|
|
|
def test_a_failing_section_becomes_a_line_not_a_loss(self):
|
|
def casse():
|
|
raise RuntimeError("sonde cassée")
|
|
|
|
texte = TODO._diag_section("essai", casse)
|
|
self.assertIn("===== essai =====", texte)
|
|
self.assertIn("RuntimeError", texte)
|
|
self.assertIn("sonde cassée", texte)
|
|
|
|
def test_a_healthy_section_keeps_its_output(self):
|
|
texte = TODO._diag_section("essai", lambda: print("relevé"))
|
|
self.assertIn("relevé", texte)
|
|
self.assertNotIn("Error", texte)
|