erplibre/test/test_proxmox_deploy.py

#!/usr/bin/env python3
# © 2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
"""Déployer sur un hôte Proxmox : l'hyperviseur est AILLEURS.

Toute la différence avec QEMU/KVM tient là. Rien ne s'exécute sur la machine
locale : il faut d'abord savoir OÙ, puis tout envoyer par SSH. Ces tests
gardent ce qui a été appris contre un hôte réel (Proxmox VE 9.2.11 dans une VM
libvirt), une panne après l'autre :

- « qm » exige root. La voie « VM QEMU locale » ne donne que l'accès
  d'erplibre : il faut sudo, et l'enrober AUTOUR de toute la commande — les
  commandes de ce module sont des suites et des redirections, et « sudo cmd »
  n'élèverait que le premier mot.
- Une Proxmox installée SUR Debian n'a AUCUN pont. On en propose un INTERNE :
  ajouter l'interface physique à un pont déplace l'adresse de l'hôte et coupe
  la session SSH — à distance, c'est sans retour.
- Sur un pont interne, aucun DHCP ne répond : l'adresse doit être fixe, et
  elle est alors connue AVANT le démarrage. La chercher ensuite était absurde.
- L'agent invité n'est pas dans l'image cloud Debian : le voisinage de l'hôte
  (« ip neigh ») est le seul repli, et il a trouvé l'adresse là où l'agent
  répondait « not running ».
"""

import shlex
import subprocess
import sys
import unittest
from unittest import mock

sys.argv = ["todo.py"]
from script.proxmox import proxmox_deploy as pve  # noqa: E402
from script.todo.todo import TODO  # noqa: E402
from script.todo.todo_i18n import t  # noqa: E402

# Sorties RÉELLES relevées sur l'hôte d'essai.
PVEVERSION = (
    "pve-manager/9.2.11/f6997e698c7933ea (running kernel: 7.0.14-12-pve)"
)
# Ce que ssh écrit sur stderr à chaque connexion d'un hôte en
# UserKnownHostsFile=/dev/null. Ce n'est pas un diagnostic.
AVERTISSEMENT = (
    "Warning: Permanently added '192.168.123.227' (ED25519) to the list "
    "of known hosts.\n"
)
SPEC_VM = {
    "name": "essai",
    "memory": 512,
    "vcpus": 1,
    "disk": "4G",
    "storage": "local",
    "bridge": "vmbr0",
    "image": "debian-13.qcow2",
    "user": "erplibre",
    "start": True,
}
QM_LIST = """      VMID NAME                 STATUS     MEM(MB)    BOOTDISK(GB) PID
       100 vm-essai             running    2048              16.00 2726
       101 avec un espace       stopped    4096              32.00 0
"""
PVESM = """Name             Type     Status           Total            Used       Available        %
local             dir     active        32815812         6873084        24559348   20.94%
sauvegarde        dir   inactive        99999999               0        99999999    0.00%
"""
NEIGH = """192.168.123.1 dev enp1s0 lladdr 52:54:00:cd:73:ef REACHABLE
10.10.10.150 dev vmbr0 lladdr bc:24:11:93:da:22 REACHABLE
"""
QM_CONFIG = """boot: order=scsi0
memory: 2048
net0: virtio=BC:24:11:93:DA:22,bridge=vmbr0
scsi0: local:100/vm-100-disk-0.raw,discard=on,size=16G,ssd=1
"""
INTERFACES = """auto lo
iface lo inet loopback

iface enp1s0 inet manual

auto vmbr0
iface vmbr0 inet static
    address 10.10.10.1/24
    bridge-ports none
    bridge-stp off

auto vmbr1
iface vmbr1 inet manual
    bridge-ports enp2s0
"""


class TestLectureDesSorties(unittest.TestCase):
    def test_the_version_proves_it_is_a_proxmox(self):
        """Une adresse saisie à la main peut être n'importe quelle machine :
        sans cette preuve, la première commande « qm » échouerait sur un
        « command not found » qui n'explique rien."""
        self.assertEqual("9.2.11", pve.parse_pveversion(PVEVERSION))
        self.assertEqual(
            "", pve.parse_pveversion("bash: pveversion: not found")
        )

    def test_a_vm_name_with_spaces_is_read_whole(self):
        """« qm list » sépare par des espaces, et un nom peut en contenir : on
        découpe par les deux bouts, le milieu est le nom."""
        vms = pve.parse_qm_list(QM_LIST)
        self.assertEqual([100, 101], [v["vmid"] for v in vms])
        self.assertEqual("avec un espace", vms[1]["name"])
        self.assertEqual("running", vms[0]["status"])

    def test_the_header_is_not_a_vm(self):
        self.assertEqual([], pve.parse_qm_list("      VMID NAME  STATUS\n"))
        self.assertEqual([], pve.parse_qm_list(""))

    def test_only_active_storages_count_and_kib_become_bytes(self):
        st = pve.parse_storages(PVESM)
        self.assertEqual(["local", "sauvegarde"], [s["name"] for s in st])
        self.assertTrue(st[0]["actif"])
        self.assertFalse(st[1]["actif"])
        self.assertEqual(24559348 * 1024, st[0]["avail"])

    def test_bridges_are_read_without_their_at_suffix(self):
        texte = "3: vmbr0: <BROADCAST,UP>\n4: vmbr1@if2: <BROADCAST>\n"
        self.assertEqual(["vmbr0", "vmbr1"], pve.parse_bridges(texte))

    def test_the_guest_agent_answer_drops_loopback_and_ipv6(self):
        json_txt = (
            '[{"name":"lo","ip-addresses":[{"ip-address-type":"ipv4",'
            '"ip-address":"127.0.0.1"}]},{"name":"eth0","ip-addresses":['
            '{"ip-address-type":"ipv4","ip-address":"10.10.10.150"},'
            '{"ip-address-type":"ipv6","ip-address":"fe80::1"}]}]'
        )
        self.assertEqual(["10.10.10.150"], pve.parse_guest_ips(json_txt))

    def test_a_missing_agent_is_not_a_crash(self):
        """Sa réponse n'est pas du JSON : « QEMU guest agent is not running »."""
        self.assertEqual(
            [], pve.parse_guest_ips("QEMU guest agent is not running")
        )

    def test_the_mac_links_a_vm_to_its_address(self):
        """Le seul lien quand l'agent manque, et l'image cloud Debian ne
        l'embarque pas."""
        mac = pve.mac_from_config(QM_CONFIG)
        self.assertEqual("bc:24:11:93:da:22", mac)
        self.assertEqual("10.10.10.150", pve.ip_from_neigh(NEIGH, mac))

    def test_an_unknown_mac_finds_nothing(self):
        self.assertEqual("", pve.ip_from_neigh(NEIGH, "de:ad:be:ef:00:00"))
        self.assertEqual("", pve.ip_from_neigh(NEIGH, ""))

    def test_a_lan_bridge_and_an_internal_one_are_told_apart(self):
        ponts = pve.parse_bridge_config(INTERFACES)
        self.assertEqual("", ponts["vmbr0"]["ports"])
        self.assertEqual("10.10.10.1/24", ponts["vmbr0"]["address"])
        self.assertEqual("enp2s0", ponts["vmbr1"]["ports"])

    def test_orphans_are_the_volumes_no_vm_claims(self):
        liste = (
            "Volid                     Format  Type      Size VMID\n"
            "local:100/vm-100-disk-0.raw raw   images 17179869184 100\n"
            "local:999/vm-999-disk-0.raw raw   images  8589934592 999\n"
        )
        orph = pve.parse_orphans(liste, [100])
        self.assertEqual(1, len(orph))
        self.assertIn("999", orph[0][0])


class TestLeBruitDeSsh(unittest.TestCase):
    """Ce que ssh ajoute n'est pas la réponse de l'hôte.

    Le cas vécu, du début à la fin : « ip -o link show type bridge » ne rend
    RIEN sur un hôte sans pont, la sortie ne contient donc que
    l'avertissement de ssh sur la clé — que `parse_bridges` a pris pour un nom
    de pont. « (ED25519) » s'est retrouvé dans « --net0 virtio,bridge=… »,
    enrobé de « sudo sh -c », et dash a répondu :

        sh: 1: Syntax error: "(" unexpected

    Trois lignes de code entre la cause et un message incompréhensible.
    """

    def test_the_warning_never_becomes_a_bridge(self):
        self.assertEqual(pve.parse_bridges(AVERTISSEMENT), [])

    def test_real_bridges_are_still_read(self):
        vrai = (
            "2: vmbr0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc "
            "noqueue state UP mode DEFAULT group default qlen 1000\\    "
            "link/ether bc:24:11:00:00:01\n"
            "3: vmbr1: <BROADCAST,MULTICAST> mtu 1500 qdisc noop state DOWN\n"
        )
        self.assertEqual(
            pve.parse_bridges(AVERTISSEMENT + vrai), ["vmbr0", "vmbr1"]
        )

    def test_a_veth_pair_keeps_only_its_own_name(self):
        ligne = "7: fwln100i0@fwpr100p0: <BROADCAST,MULTICAST,UP> mtu 1500\n"
        self.assertEqual(pve.parse_bridges(ligne), ["fwln100i0"])

    def test_the_noise_is_stripped_at_the_source(self):
        for bruit in (
            AVERTISSEMENT,
            "Pseudo-terminal will not be allocated because stdin is not a terminal.\n",
            "Connection to 10.0.0.5 closed.\n",
            "Shared connection to 10.0.0.5 closed.\n",
            "mesg: ttyname failed: Inappropriate ioctl for device\n",
        ):
            self.assertEqual(pve.strip_ssh_noise(bruit), "")
        self.assertEqual(
            pve.strip_ssh_noise(AVERTISSEMENT + "vmbr0\n"), "vmbr0\n"
        )

    def test_the_answer_survives_the_filter(self):
        # Un filtre qui mange la réponse serait pire que le bruit.
        self.assertIn(
            "pve-manager", pve.strip_ssh_noise(AVERTISSEMENT + PVEVERSION)
        )


class TestLeNoyau(unittest.TestCase):
    """Tant que l'hôte tourne le noyau de la distribution, il n'a ni module
    bridge ni table NAT : ifupdown2 répond « Operation not supported », et
    quand /run/network manque il répond même « Another instance of this
    program is already running » — un mensonge. Vécu sur l'hôte d'essai."""

    def test_the_running_kernel_is_read_from_pveversion(self):
        self.assertEqual(
            pve.parse_kernel(
                "pve-manager/9.2.11/abc (running kernel: 6.12.95+deb13-cloud-amd64)"
            ),
            "6.12.95+deb13-cloud-amd64",
        )
        self.assertEqual(
            pve.parse_kernel(
                "pve-manager/9.2.11/abc (running kernel: 7.0.14-12-pve)"
            ),
            "7.0.14-12-pve",
        )
        self.assertEqual(pve.parse_kernel("n'importe quoi"), "")

    def test_the_bridge_creates_the_lock_directory_first(self):
        # Sans /run/network, ifupdown2 accuse une autre instance et le pont
        # ne monte jamais.
        montee = pve.bridge_setup_cmds("vmbr0", "10.10.10.1/24", "enp1s0")[-1]
        self.assertIn("mkdir -p /run/network", montee)
        # Et l'erreur d'ifup n'est plus masquée : c'est elle qui explique.
        self.assertNotIn("2>/dev/null", montee)


class TestLeDns(unittest.TestCase):
    """« --ipconfig0 » ne porte pas le DNS : une VM en adresse fixe se
    retrouvait sans résolveur. Mesuré sur la VM d'essai — le NAT routait, mais
    « getent hosts deb.debian.org » ne rendait rien."""

    def test_the_resolved_stub_is_useless_to_a_guest(self):
        self.assertEqual(pve.parse_nameservers("nameserver 127.0.0.53"), [])

    def test_real_resolvers_are_kept_in_order(self):
        self.assertEqual(
            pve.parse_nameservers(
                "nameserver 192.168.123.1\nnameserver 1.1.1.1\n"
                "nameserver 192.168.123.1\n"
            ),
            ["192.168.123.1", "1.1.1.1"],
        )

    def test_a_static_address_gets_the_resolvers(self):
        spec = dict(
            SPEC_VM,
            ipconfig="ip=10.10.10.150/24,gw=10.10.10.1",
            nameservers=["192.168.123.1"],
        )
        ci = [c for c in pve.create_cmds(100, spec) if "--ciuser" in c][0]
        self.assertIn("--nameserver 192.168.123.1", ci)

    def test_dhcp_needs_none(self):
        # Le bail DHCP porte déjà le DNS.
        spec = dict(SPEC_VM, ipconfig="ip=dhcp", nameservers=["192.168.123.1"])
        ci = [c for c in pve.create_cmds(100, spec) if "--ciuser" in c][0]
        self.assertNotIn("--nameserver", ci)


class TestLAvancement(unittest.TestCase):
    """Cent lignes « transferred … » enterraient l'erreur utile : le journal du
    premier essai réel faisait 136 lignes pour 34 utiles."""

    def test_a_burst_collapses_to_one_line(self):
        texte = (
            "Formatting 'disk.raw'\n"
            + "".join(
                f"transferred {i}.0 MiB of 3.0 GiB ({i}%)\n" for i in range(50)
            )
            + "scsi0: successfully created disk\n"
        )
        propre = pve.collapse_progress(texte)
        self.assertIn("50 lignes d'avancement", propre)
        self.assertIn("successfully created disk", propre)
        self.assertLess(len(propre.splitlines()), 6)

    def test_what_is_not_progress_is_untouched(self):
        texte = "400 Parameter verification failed.\nnet0: invalid format\n"
        self.assertEqual(pve.collapse_progress(texte).strip(), texte.strip())


class TestLesChoix(unittest.TestCase):
    def test_the_vmid_skips_the_taken_ones(self):
        """Proxmox refuse un VMID pris, et le dit APRÈS le téléchargement de
        l'image : on choisit donc avant, d'après ce que l'hôte déclare."""
        self.assertEqual(
            102, pve.next_vmid([{"vmid": 100}, {"vmid": 101}, {"vmid": 103}])
        )
        self.assertEqual(100, pve.next_vmid([]))

    def test_the_storage_is_the_freest_active_one(self):
        st = pve.parse_storages(PVESM)
        self.assertEqual("local", pve.pick_storage(st))

    def test_an_unknown_storage_is_refused_not_guessed(self):
        """« local-lvm » n'existe pas partout : un repli deviné ferait échouer
        « qm set » après le téléchargement de l'image."""
        self.assertEqual(
            "", pve.pick_storage(pve.parse_storages(PVESM), "nas")
        )

    def test_vmbr0_wins_when_it_exists(self):
        self.assertEqual("vmbr0", pve.pick_bridge(["vmbr9", "vmbr0"]))
        self.assertEqual("br-lan", pve.pick_bridge(["br-lan"]))
        self.assertEqual("", pve.pick_bridge([]))


class TestLesCommandes(unittest.TestCase):
    def _spec(self, **extra):
        base = {
            "name": "vm-essai",
            "memory": 2048,
            "vcpus": 2,
            "disk": "12G",
            "storage": "local",
            "bridge": "vmbr0",
            "image": "debian-13-genericcloud-amd64.qcow2",
            "sshkey_path": "/root/.ssh/erplibre-deploy.pub",
            "ipconfig": "ip=10.10.10.150/24,gw=10.10.10.1",
        }
        base.update(extra)
        return base

    def test_the_sequence_is_in_the_order_proxmox_needs(self):
        cmds = pve.create_cmds(100, self._spec())
        joint = "\n".join(cmds)
        self.assertTrue(cmds[0].startswith("qm create 100"))
        self.assertIn("import-from=", joint)
        self.assertIn(":cloudinit", joint)
        self.assertIn("--boot order=scsi0", joint)
        self.assertIn("qm resize 100 scsi0 12G", joint)
        self.assertTrue(cmds[-1].endswith("qm start 100"))

    def test_the_agent_and_the_serial_console_are_asked_for(self):
        """Sans agent, aucune adresse ; sans serial0, « qm terminal » est
        inutilisable et il ne reste que l'interface web."""
        cmd = pve.create_cmds(100, self._spec())[0]
        self.assertIn("--agent enabled=1", cmd)
        self.assertIn("--serial0 socket", cmd)

    def test_a_name_with_a_space_cannot_break_the_command(self):
        """Le nom vient d'une saisie : découpée par le shell, elle doit rester
        UN argument. « rm » ne doit jamais devenir une commande."""
        mechant = "vm essai; rm -rf /"
        cmds = pve.create_cmds(100, self._spec(name=mechant))
        args = shlex.split(cmds[0])
        self.assertIn(mechant, args)
        self.assertNotIn("rm", args)

    def test_destroy_stops_first_and_purges(self):
        cmds = pve.destroy_cmds(100)
        self.assertIn("qm stop 100", cmds[0])
        self.assertIn("--purge 1", cmds[1])

    def test_the_image_is_fetched_once_on_the_host(self):
        cmd = pve.image_fetch_cmd("https://x/deb.qcow2", "deb.qcow2")
        self.assertIn("if [ -s", cmd)
        self.assertIn("wget", cmd)

    def test_the_internal_bridge_never_touches_a_physical_nic(self):
        """Le point le plus important de ce module : ajouter l'interface au
        pont déplace l'adresse de l'hôte et coupe la session SSH — à distance,
        sans retour."""
        cmds = pve.bridge_setup_cmds(uplink="enp1s0")
        joint = "\n".join(cmds)
        self.assertIn("bridge-ports none", joint)
        self.assertNotIn("bridge-ports enp1s0", joint)
        self.assertIn("MASQUERADE", joint)
        self.assertIn("ip_forward", joint)

    def test_the_bridge_stanza_is_added_only_once(self):
        cmds = pve.bridge_setup_cmds()
        self.assertIn("grep -qE", cmds[0])
        self.assertIn("||", cmds[0])

    def test_an_internal_bridge_gets_a_static_address(self):
        """Aucun DHCP n'y répondrait : la VM resterait muette."""
        ponts = pve.parse_bridge_config(INTERFACES)
        self.assertEqual(
            "ip=10.10.10.150/24,gw=10.10.10.1",
            pve.ipconfig_for(ponts["vmbr0"], 100),
        )

    def test_a_lan_bridge_gets_dhcp(self):
        ponts = pve.parse_bridge_config(INTERFACES)
        self.assertEqual("ip=dhcp", pve.ipconfig_for(ponts["vmbr1"], 100))

    def test_two_vms_do_not_share_an_address(self):
        ponts = pve.parse_bridge_config(INTERFACES)
        a = pve.ipconfig_for(ponts["vmbr0"], 100)
        b = pve.ipconfig_for(ponts["vmbr0"], 101)
        self.assertNotEqual(a, b)

    def test_a_static_address_is_known_before_boot(self):
        self.assertEqual(
            "10.10.10.150",
            pve.ip_from_ipconfig("ip=10.10.10.150/24,gw=10.10.10.1"),
        )
        self.assertEqual("", pve.ip_from_ipconfig("ip=dhcp"))


class TestLePrivilege(unittest.TestCase):
    def test_the_whole_command_is_wrapped_not_just_its_first_word(self):
        """« sudo mkdir && if … fi » n'élèverait que le mkdir, et la
        redirection resterait celle du shell non privilégié : « permission
        denied » sur /root ou /boot/efi."""
        compose = "mkdir -p /root/.ssh && printf x > /root/.ssh/k"
        enrobe = pve.wrap_privilege(compose, "sudo ")
        self.assertTrue(enrobe.startswith("sudo sh -c "))
        self.assertIn("printf x > /root/.ssh/k", enrobe)

    def test_without_sudo_the_command_is_untouched(self):
        self.assertEqual("qm list", pve.wrap_privilege("qm list", ""))

    def test_ssh_never_asks_a_question_it_cannot_show(self):
        """BatchMode : une invite de mot de passe dans un menu bloquerait sans
        rien afficher."""
        argv = pve.ssh_argv({"target": "root@h"}, "qm list")
        self.assertIn("BatchMode=yes", argv)
        self.assertIn("ConnectTimeout=10", " ".join(argv))

    def test_the_jump_and_the_port_travel(self):
        argv = pve.ssh_argv(
            {"target": "root@h", "jump": "rebond", "port": "2222"}, "x"
        )
        self.assertIn("-J", argv)
        self.assertIn("rebond", argv)
        self.assertIn("-p", argv)
        self.assertIn("2222", argv)

    def test_a_console_gets_a_tty_and_no_batchmode(self):
        argv = pve.ssh_argv({"target": "root@h"}, "qm terminal 100", tty=True)
        self.assertIn("-t", argv)
        self.assertNotIn("BatchMode=yes", argv)


class TestChoixDeLHote(unittest.TestCase):
    """La question propre à Proxmox : sur QUELLE machine ?"""

    def _todo(self):
        todo = TODO.__new__(TODO)
        todo._pve_remember_host = lambda h: None
        return todo

    def test_an_unknown_host_key_is_recognised(self):
        for texte in (
            "Host key verification failed.",
            "The authenticity of host '10.0.0.1' can't be established.",
            "No ED25519 host key is known for 10.0.0.1",
        ):
            self.assertTrue(TODO._pve_hostkey_missing(texte), texte)
        self.assertFalse(TODO._pve_hostkey_missing("Permission denied"))

    def _confirm(self, reponses):
        """reponses : [(code, sortie)] pour chaque appel à pve.run."""
        it = iter(reponses)
        with mock.patch.object(
            pve, "run", side_effect=lambda *a, **k: next(it)
        ):
            import contextlib
            import io

            out = io.StringIO()
            with contextlib.redirect_stdout(out):
                host = self._todo()._pve_confirm_host(
                    {"target": "erplibre@10.0.0.5", "jump": ""}
                )
            return host, out.getvalue()

    def test_a_non_proxmox_host_is_refused_with_what_was_seen(self):
        # Deux appels : « pveversion », puis la sonde qui demande à ssh s'il
        # passe — c'est elle qui distingue les deux pannes.
        host, sortie = self._confirm(
            [(127, "bash: pveversion: not found"), (0, "")]
        )
        self.assertIsNone(host)
        self.assertIn("pveversion", sortie)

    def test_a_reachable_machine_without_proxmox_says_exactly_that(self):
        """Le cas rapporté : « je n'arrive pas à me connecter, pourtant il est
        accessible ». La machine répondait ; c'est Proxmox qui manquait, et le
        message parlait d'injoignabilité."""
        host, sortie = self._confirm(
            [
                (127, AVERTISSEMENT + "bash: pveversion: command not found"),
                (0, AVERTISSEMENT),
            ]
        )
        self.assertIsNone(host)
        # Comparé à la TRADUCTION, pas à un mot français : la langue de
        # l'interface se change (EL_LANG), et un test qui la suppose échoue
        # pour une raison qui n'a rien à voir avec ce qu'il vérifie.
        self.assertIn(t("Reachable, but Proxmox VE is not there:"), sortie)
        self.assertIn("install_proxmox.sh", sortie)
        # Et surtout : ne plus envoyer chercher un problème de réseau.
        self.assertNotIn(t("SSH does not get through:"), sortie)

    def test_an_unreachable_machine_says_ssh_does_not_get_through(self):
        panne = "ssh: connect to host 10.0.0.9 port 22: No route to host"
        host, sortie = self._confirm([(255, panne), (255, panne)])
        self.assertIsNone(host)
        self.assertIn("No route to host", sortie)
        self.assertNotIn("install_proxmox.sh", sortie)

    def test_the_ssh_key_warning_is_never_shown_as_the_error(self):
        # Affichée comme preuve, elle envoyait chercher un problème de clé
        # d'hôte qui n'existait pas — c'est ce qu'on voyait dans le rapport.
        host, sortie = self._confirm(
            [(127, AVERTISSEMENT), (0, AVERTISSEMENT)]
        )
        self.assertIsNone(host)
        self.assertNotIn("Permanently added", sortie)

    def test_only_the_lines_that_teach_something_are_kept(self):
        self.assertEqual(
            TODO._pve_clean_output(
                AVERTISSEMENT + "\nbash: pveversion: command not found\n"
            ),
            ["bash: pveversion: command not found"],
        )
        self.assertEqual(TODO._pve_clean_output(AVERTISSEMENT), [])
        self.assertEqual(TODO._pve_clean_output(""), [])

    def test_the_install_hint_pipes_the_repo_script(self):
        # Le script est autonome : « bash -s » suffit, rien à copier d'abord.
        indice = self._todo()._pve_install_hint({"target": "pve1"})
        self.assertIn(TODO.PVE_INSTALL_SCRIPT, indice)
        self.assertIn("ssh pve1 sudo bash -s", indice)

    def test_a_non_root_access_gets_sudo(self):
        """C'est le cas de la voie « VM QEMU locale » : cloud-init crée
        erplibre, pas root."""
        host, _s = self._confirm([(0, PVEVERSION), (0, "1000\n"), (0, "")])
        self.assertEqual("sudo ", host["sudo"])
        self.assertEqual("9.2.11", host["version"])

    def test_root_needs_no_sudo(self):
        host, _s = self._confirm([(0, PVEVERSION), (0, "0\n")])
        self.assertEqual("", host["sudo"])

    def test_without_root_nor_passwordless_sudo_it_stops(self):
        """Un sudo qui réclame un mot de passe bloquerait chaque commande du
        menu sur une invite que personne ne voit."""
        host, sortie = self._confirm(
            [
                (0, PVEVERSION),
                (0, "1000\n"),
                (1, "sudo: a password is required"),
            ]
        )
        self.assertIsNone(host)
        self.assertIn("root", sortie)


class TestLeMenu(unittest.TestCase):
    def test_proxmox_sits_right_under_qemu_in_the_deploy_menu(self):
        src = open("script/todo/todo.py", encoding="utf-8").read()
        i_qemu = src.index('"QEMU/KVM - Deploy an Ubuntu VM (libvirt)"')
        i_pve = src.index('"Proxmox VE - Deploy a VM on a remote host"')
        i_ntfy = src.index('"Deploy - Install NTFY notification server"')
        self.assertLess(i_qemu, i_pve)
        self.assertLess(i_pve, i_ntfy)

    def test_the_dispatch_follows_the_list(self):
        src = open("script/todo/todo.py", encoding="utf-8").read()
        self.assertIn(
            'elif status == "6":\n                self.prompt_execute_proxmox()',
            src,
        )
        self.assertIn(
            'elif status == "7":\n                self._deploy_ntfy_server()',
            src,
        )

    def test_every_qemu_entry_has_its_proxmox_counterpart(self):
        """L'équivalent des dix-sept commandes, plus le choix de l'hôte.

        Le menu vit dans son propre fichier depuis le refactor : la cohérence
        numéro/dispatch, elle, est vérifiée par le socle commun de
        test_todo_menu.py, qui sert les deux menus.
        """
        src = open("script/todo/proxmox_menu.py", encoding="utf-8").read()
        debut = src.index("    def prompt_execute_proxmox(self):")
        bloc = src[debut : src.index("    def _pve_fetch_image(self):")]
        for n in range(1, 19):
            self.assertIn(f'elif status == "{n}":', bloc, f"entrée {n}")

    def test_the_script_is_valid_python(self):
        res = subprocess.run(
            [
                sys.executable,
                "-c",
                "import ast;ast.parse(open('script/proxmox/proxmox_deploy.py',encoding='utf-8').read())",
            ],
            capture_output=True,
            text=True,
        )
        self.assertEqual(0, res.returncode, res.stderr)


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