erplibre/test/test_qemu_start_egl.py

#!/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 "")


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)


if __name__ == "__main__":
    unittest.main()