Mathieu Benoit
e0624d9aaa
Rien n'invalidait une entrée dont la somme ne correspond pas : le magasin continuait de la servir, et retélécharger ne changeait rien puisque c'est lui qui répond. Les deux purges existantes ne l'atteignent pas — l'une efface tout, l'autre saute ce qui est récent alors que chaque service remet cette date, si bien qu'un objet empoisonné qui sert ne vieillit jamais. Symétrique de « --detient » : mêmes lignes, mêmes fonctions de clé, mêmes refus, et un objet présent qui résiste est dit refusé plutôt qu'oublié. Mesuré : garde, puis oublié 53080, puis absent. --- EN --- Nothing invalidated an entry whose checksum does not match: the store kept serving it, and re-downloading changed nothing since the store is what answers. Neither existing purge reaches it — one erases everything, the other skips what is recent while every service resets that date, so a poisoned object that serves never ages. Symmetric to "--detient": same lines, same key functions, same refusals, and a present object that resists is reported refused rather than forgotten. Measured: held, then forgotten 53080, then absent. Assisted-by: Claude Opus 5
1501 lines
62 KiB
Python
1501 lines
62 KiB
Python
#!/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 pas masquée : c'est elle qui explique.
|
|
# Porté sur l'appel lui-même, et non sur toute la ligne : le repli qui
|
|
# suit sonde légitimement (« ip link show », « iptables -C »), et
|
|
# interdire « 2>/dev/null » partout lui interdisait d'exister.
|
|
ifup = montee[montee.index("ifup ") :].split("||")[0]
|
|
self.assertNotIn("2>", ifup)
|
|
|
|
|
|
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_without_3d_the_screen_stays_the_serial_console(self):
|
|
"""Le défaut d'une machine de serveur : « qm terminal » en dépend, et
|
|
poser un périphérique vidéo d'office changerait son matériel."""
|
|
creation = pve.create_cmds(100, self._spec())[0]
|
|
self.assertIn("--vga serial0", creation)
|
|
self.assertIn("--serial0 socket", creation)
|
|
|
|
def test_3d_asks_for_an_accelerated_screen(self):
|
|
"""« virtio-gl » donne à l'invité un virtio-gpu que le VIRGL de
|
|
l'hôte accélère."""
|
|
creation = pve.create_cmds(100, self._spec(gpu3d=True))[0]
|
|
self.assertIn("--vga virtio-gl", creation)
|
|
self.assertNotIn("--vga serial0", creation)
|
|
|
|
def test_the_serial_port_survives_the_accelerated_screen(self):
|
|
"""Le port série reste posé : sans lui, « qm terminal » n'a plus rien
|
|
où s'attacher, et une VM qui ne démarre pas devient muette."""
|
|
self.assertIn(
|
|
"--serial0 socket", pve.create_cmds(100, self._spec(gpu3d=True))[0]
|
|
)
|
|
|
|
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_a_substituted_image_names_what_to_do_first(self):
|
|
"""Cet hôte peut être une VM derrière le cache de téléchargement :
|
|
effacer l'image la fera resservir à l'identique depuis le magasin.
|
|
L'entrée s'en retire d'abord, et l'échec le dit — c'est là que
|
|
l'opérateur se tient quand la somme lâche."""
|
|
cmd = pve.image_fetch_cmd(
|
|
"https://x/deb.qcow2", "deb.qcow2", sha256="ab" * 32
|
|
)
|
|
self.assertIn("--oublie", cmd)
|
|
self.assertIn("rm -f", cmd)
|
|
# Vers l'erreur standard : la sortie normale est lue par la machine.
|
|
self.assertIn(">&2", cmd)
|
|
|
|
def test_the_failure_is_still_a_failure(self):
|
|
"""Un message n'absout pas : « false » garde le code de retour, sans
|
|
quoi le déploiement continuerait sur une image substituée."""
|
|
cmd = pve.image_fetch_cmd(
|
|
"https://x/deb.qcow2", "deb.qcow2", sha256="ab" * 32
|
|
)
|
|
self.assertIn("false;", 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)
|
|
|
|
|
|
class TestLaTableNat(unittest.TestCase):
|
|
"""« Table does not exist » : six lignes d'iptables et « code de retour 1 »,
|
|
après avoir déjà écrit la strophe dans /etc/network/interfaces.
|
|
|
|
Rien dans ce bruit ne dit qu'il faut redémarrer. Et le cas n'a rien
|
|
d'exotique : notre propre install_proxmox.sh pose le noyau Proxmox sans
|
|
redémarrer — lancé par ssh, un reboot couperait la session. Une Proxmox
|
|
imbriquée fraîchement installée est donc TOUJOURS sur le noyau cloud de
|
|
Debian, qui est dépouillé de tout netfilter.
|
|
|
|
On demande donc à la table NAT elle-même, et non au NOM du noyau : « -pve »
|
|
est un indice, pas une preuve."""
|
|
|
|
def _sortie(self, kernel, nat, pve_kernel=""):
|
|
return (
|
|
f"{kernel}\n---ERPLIBRE-NAT---\n"
|
|
f"{'NAT-OK' if nat else 'NAT-KO'}\n"
|
|
f"---ERPLIBRE-PVE-KERNEL---\n{pve_kernel}\n"
|
|
)
|
|
|
|
def test_a_working_host(self):
|
|
lu = pve.parse_nat_check(
|
|
self._sortie("7.0.14-14-pve", True, "7.0.14-14-pve")
|
|
)
|
|
self.assertTrue(lu["nat"])
|
|
self.assertEqual(lu["kernel"], "7.0.14-14-pve")
|
|
|
|
def test_the_cloud_kernel_waiting_for_a_reboot(self):
|
|
# L'état exact rapporté : le noyau Proxmox est POSÉ, pas amorcé.
|
|
lu = pve.parse_nat_check(
|
|
self._sortie("6.12.101+deb13-cloud-amd64", False, "7.0.14-14-pve")
|
|
)
|
|
self.assertFalse(lu["nat"])
|
|
self.assertEqual(lu["pve_kernel"], "7.0.14-14-pve")
|
|
|
|
def test_an_unfinished_install_has_no_pve_kernel(self):
|
|
lu = pve.parse_nat_check(
|
|
self._sortie("6.12.101+deb13-cloud-amd64", False)
|
|
)
|
|
self.assertFalse(lu["nat"])
|
|
self.assertEqual(lu["pve_kernel"], "")
|
|
|
|
def test_ssh_noise_does_not_become_a_kernel(self):
|
|
brut = (
|
|
"Warning: Permanently added 'x' (ED25519) to the list of known"
|
|
" hosts.\n" + self._sortie("7.0.14-14-pve", True, "7.0.14-14-pve")
|
|
)
|
|
self.assertEqual(pve.parse_nat_check(brut)["kernel"], "7.0.14-14-pve")
|
|
|
|
def test_the_probe_asks_the_table_not_the_name(self):
|
|
self.assertIn("iptables -t nat", pve.NAT_CHECK_CMD)
|
|
self.assertIn("uname -r", pve.NAT_CHECK_CMD)
|
|
|
|
|
|
class TestLeReseauDuPontInterne(unittest.TestCase):
|
|
"""Le pont interne avait une adresse CODÉE EN DUR, 10.10.10.1/24.
|
|
|
|
Un Proxmox dans un Proxmox hérite du réseau interne de son parent : la VM
|
|
vivait en 10.10.10.152 avec 10.10.10.1 pour PASSERELLE. Lui demander de
|
|
poser 10.10.10.1/24 sur son propre pont, c'est prendre l'adresse de sa
|
|
passerelle et rendre tout le /24 local — la machine s'isole au milieu de
|
|
la commande qui la configure. Vécu : « ifup » n'a jamais rendu la main, et
|
|
la VM ne répondait plus ni en ssh ni en ping."""
|
|
|
|
IMBRIQUE = (
|
|
"2: eth0 inet 10.10.10.152/24 brd 10.10.10.255 scope global eth0\n"
|
|
"default via 10.10.10.1 dev eth0 onlink\n"
|
|
"10.10.10.0/24 dev eth0 proto kernel scope link src 10.10.10.152\n"
|
|
)
|
|
|
|
def test_a_nested_host_gets_another_subnet(self):
|
|
self.assertNotEqual(
|
|
pve.pick_internal_cidr(self.IMBRIQUE), "10.10.10.1/24"
|
|
)
|
|
self.assertEqual(
|
|
pve.pick_internal_cidr(self.IMBRIQUE), "10.10.20.1/24"
|
|
)
|
|
|
|
def test_a_fresh_host_keeps_the_usual_one(self):
|
|
vierge = "1: lo inet 127.0.0.1/8 scope host lo\n"
|
|
self.assertEqual(pve.pick_internal_cidr(vierge), "10.10.10.1/24")
|
|
|
|
def test_a_route_alone_is_enough_to_collide(self):
|
|
# Une route sans adresse locale suffit : c'est le cas exact de la
|
|
# route par défaut « via 10.10.10.1 ».
|
|
seule = "default via 10.10.10.1 dev eth0\n"
|
|
self.assertNotEqual(pve.pick_internal_cidr(seule), "10.10.10.1/24")
|
|
|
|
def test_a_supernet_rules_out_everything_under_it(self):
|
|
# « 10.0.0.0/8 » couvre tous les candidats en 10.x. Un test sur les
|
|
# trois premiers octets l'aurait raté.
|
|
choisi = pve.pick_internal_cidr("10.0.0.0/8 dev x\n")
|
|
self.assertFalse(choisi.startswith("10."), choisi)
|
|
|
|
def test_when_nothing_is_free_it_says_so(self):
|
|
tout = "\n".join(
|
|
c.replace("1/24", "0/24") for c in pve.INTERNAL_CANDIDATES
|
|
)
|
|
self.assertEqual(pve.pick_internal_cidr(tout), "")
|
|
|
|
def test_the_chosen_subnet_reaches_every_command(self):
|
|
cmds = pve.bridge_setup_cmds(cidr="10.10.20.1/24", uplink="eth0")
|
|
texte = "\n".join(cmds)
|
|
self.assertIn("address 10.10.20.1/24", texte)
|
|
self.assertIn("10.10.20.0/24", texte)
|
|
self.assertNotIn("10.10.10.", texte)
|
|
|
|
|
|
class TestLeRepliQuiNeCoupePasLaLigne(unittest.TestCase):
|
|
"""« ifreload -a » en repli rechargeait TOUTES les interfaces.
|
|
|
|
Y compris celle qui porte la session ssh — et sur une image cloud
|
|
l'interface principale est décrite ailleurs (interfaces.d, netplan), donc
|
|
ifupdown2 la descend sans la remonter. Le repli monte donc le pont à la
|
|
main, sans toucher à rien d'autre."""
|
|
|
|
def test_ifreload_is_gone(self):
|
|
texte = "\n".join(pve.bridge_setup_cmds(uplink="eth0"))
|
|
self.assertNotIn("ifreload", texte)
|
|
|
|
def test_the_fallback_builds_the_bridge_itself(self):
|
|
derniere = pve.bridge_setup_cmds(cidr="10.10.20.1/24", uplink="eth0")[
|
|
-1
|
|
]
|
|
self.assertIn("ifup vmbr0 ||", derniere)
|
|
self.assertIn("ip link add vmbr0 type bridge", derniere)
|
|
self.assertIn("ip addr add 10.10.20.1/24 dev vmbr0", derniere)
|
|
self.assertIn("ip link set vmbr0 up", derniere)
|
|
|
|
def test_the_masquerade_rule_is_idempotent(self):
|
|
# « -C » avant « -A » : rejouée, la commande n'empile pas les règles.
|
|
derniere = pve.bridge_setup_cmds(uplink="eth0")[-1]
|
|
self.assertIn("iptables -t nat -C POSTROUTING", derniere)
|
|
self.assertLess(
|
|
derniere.index("-t nat -C"), derniere.index("-t nat -A")
|
|
)
|
|
|
|
def test_the_fallback_is_valid_shell(self):
|
|
"""Exécuté pour de vrai, ip/iptables/ifup bouchonnés.
|
|
|
|
Un repli qu'on ne sait pas exécuter s'ouvre le jour où il casse — et
|
|
celui-là tourne sur une machine qu'on ne peut plus joindre s'il rate.
|
|
"""
|
|
import subprocess
|
|
|
|
derniere = pve.bridge_setup_cmds(cidr="10.10.20.1/24", uplink="eth0")[
|
|
-1
|
|
]
|
|
bouchons = (
|
|
'ip() { [ "$1 $2" = "link show" ] && return 1; return 0; }\n'
|
|
"iptables() { return 1; }\n"
|
|
"ifup() { return 1; }\n"
|
|
"mkdir() { :; }\n"
|
|
)
|
|
res = subprocess.run(
|
|
["bash", "-c", bouchons + derniere],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
self.assertEqual(res.stderr, "", res.stderr)
|
|
|
|
|
|
class TestPourquoiAucunStockage(unittest.TestCase):
|
|
"""« Il manque le stockage » est un symptôme, pas une cause.
|
|
|
|
« pvesm » ne parle qu'à travers /etc/pve, monté par pmxcfs. pmxcfs à
|
|
terre, la commande répond « Connection refused », la liste est vide, et
|
|
l'écran s'arrête sur le symptôme — le défaut est trois étages plus bas.
|
|
|
|
Vécu sur un Proxmox imbriqué : le nom d'hôte ne résolvait que vers
|
|
127.0.1.1, parce que cloud-init réécrit /etc/hosts à CHAQUE démarrage. Le
|
|
redémarrage désormais automatique défaisait donc la correction que
|
|
l'installation venait de poser."""
|
|
|
|
def _sortie(self, actif, monte, adresses):
|
|
return (
|
|
f"{'active' if actif else 'inactive'}\n"
|
|
"---ERPLIBRE-PVE-FS---\n"
|
|
f"{'MONTE' if monte else 'ABSENT'}\n"
|
|
"---ERPLIBRE-HOSTNAME-IP---\n"
|
|
f"{' '.join(adresses)}\n"
|
|
)
|
|
|
|
def test_a_healthy_host(self):
|
|
lu = pve.parse_cluster_check(
|
|
self._sortie(True, True, ["10.10.10.152"])
|
|
)
|
|
self.assertTrue(lu["monte"])
|
|
self.assertEqual(lu["routables"], ["10.10.10.152"])
|
|
|
|
def test_a_probe_that_did_not_answer_says_so(self):
|
|
"""« La sonde n'a pas répondu » n'est PAS « rien n'est monté ».
|
|
|
|
Un dépassement de délai — hostname bloqué sur un DNS injoignable —
|
|
rend les mêmes vides. On affirmait alors « le nom ne résout que vers
|
|
? » sans avoir rien mesuré, ce qui envoyait réécrire /etc/hosts sur
|
|
une machine peut-être saine."""
|
|
self.assertFalse(pve.parse_cluster_check("timeout")["lu"])
|
|
self.assertFalse(pve.parse_cluster_check("")["lu"])
|
|
self.assertTrue(
|
|
pve.parse_cluster_check(self._sortie(True, True, ["10.0.0.1"]))[
|
|
"lu"
|
|
]
|
|
)
|
|
|
|
def test_a_link_local_address_is_not_routable(self):
|
|
"""Mesuré : « hostname --ip-address » peut ne rendre QUE des fe80::.
|
|
|
|
Le seul test « ne commence pas par 127. » les prenait pour routables,
|
|
et une APIPA en 169.254 aussi. pmxcfs n'a alors rien d'utilisable,
|
|
mais le diagnostic concluait l'inverse — et renvoyait vers journalctl
|
|
au lieu de /etc/hosts."""
|
|
for adresses in (
|
|
["fe80::5054:ff:fecf:bba9", "fe80::fc54:ff:fe79:78a4"],
|
|
["169.254.3.4"],
|
|
["127.0.1.1"],
|
|
):
|
|
with self.subTest(adresses=adresses):
|
|
lu = pve.parse_cluster_check(
|
|
self._sortie(False, False, adresses)
|
|
)
|
|
self.assertEqual(lu["routables"], [])
|
|
self.assertEqual(lu["adresses"], adresses)
|
|
|
|
def test_a_real_address_among_link_locals_still_counts(self):
|
|
lu = pve.parse_cluster_check(
|
|
self._sortie(True, True, ["10.10.10.152", "fe80::1"])
|
|
)
|
|
self.assertEqual(lu["routables"], ["10.10.10.152"])
|
|
|
|
def test_the_loopback_only_case(self):
|
|
lu = pve.parse_cluster_check(self._sortie(False, False, ["127.0.1.1"]))
|
|
self.assertFalse(lu["monte"])
|
|
self.assertEqual(lu["routables"], [])
|
|
self.assertEqual(lu["adresses"], ["127.0.1.1"])
|
|
|
|
def test_the_probe_does_not_ask_for_storage_cfg(self):
|
|
"""storage.cfg N'EXISTE PAS sur une installation neuve.
|
|
|
|
Proxmox se contente alors de ses stockages par défaut, et « local »
|
|
répond parfaitement — mesuré sur l'hôte imbriqué, où /etc/pve était
|
|
monté sans ce fichier. Le tester revenait à déclarer /etc/pve absent
|
|
sur un hôte sain."""
|
|
self.assertNotIn("storage.cfg", pve.CLUSTER_CHECK_CMD)
|
|
self.assertIn("/etc/pve/.version", pve.CLUSTER_CHECK_CMD)
|
|
|
|
def test_inactive_is_not_read_as_active(self):
|
|
# « inactive » contient « active » : la naïveté coûterait un
|
|
# diagnostic inversé.
|
|
lu = pve.parse_cluster_check(self._sortie(False, False, []))
|
|
self.assertFalse(lu["actif"])
|
|
|
|
|
|
class TestLInstalleurRendPmxcfsAuMonde(unittest.TestCase):
|
|
"""Deux gestes que l'installation ne faisait pas, et sans lesquels elle
|
|
laissait un hôte inutilisable."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
from pathlib import Path as P
|
|
|
|
cls.src = P("script/proxmox/install_proxmox.sh").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
def test_cloud_init_stops_rewriting_etc_hosts(self):
|
|
# Sans ce gel, tout ce que fait fix_hosts est ANNULÉ au prochain
|
|
# démarrage — celui que nous déclenchons nous-mêmes désormais.
|
|
self.assertIn("manage_etc_hosts: false", self.src)
|
|
self.assertIn("/etc/cloud/cloud.cfg.d", self.src)
|
|
self.assertIn("freeze_cloud_hosts", self.src)
|
|
|
|
def test_every_failed_pve_service_is_revived(self):
|
|
"""systemd marque l'unité « failed » après cinq essais rapprochés et
|
|
n'y revient jamais seul : corriger /etc/hosts ne suffit pas.
|
|
|
|
Et ils ont TOUS échoué pendant que le fichier était faux — le journal
|
|
de pvestatd le dit mot pour mot : « ipcc_send_rec failed: Connection
|
|
refused », c'est-à-dire pve-cluster absent. Relancer le seul
|
|
pve-cluster laissait pvestatd mort, donc un hôte qui ne nomme même pas
|
|
ses VM."""
|
|
self.assertIn("reset-failed", self.src)
|
|
for unite in ("pve-cluster", "pvestatd", "pvedaemon", "pveproxy"):
|
|
self.assertIn(unite, self.src, unite)
|
|
|
|
def test_pve_cluster_comes_first(self):
|
|
# Il monte /etc/pve, dont les autres dépendent.
|
|
import re
|
|
|
|
m = re.search(r'PVE_SERVICES="([^"]+)"', self.src)
|
|
self.assertIsNotNone(m)
|
|
self.assertEqual(m.group(1).split()[0], "pve-cluster")
|
|
|
|
def test_the_firewall_is_never_started_from_outside(self):
|
|
"""Le seul constat que trois lentilles ont trouvé indépendamment.
|
|
|
|
La configuration de pve-firewall vit dans
|
|
/var/lib/pve-cluster/config.db : elle est donc INVISIBLE tant que
|
|
/etc/pve n'est pas monté — c'est-à-dire exactement dans l'état qu'on
|
|
répare. Le démarrer, c'est appliquer des règles qu'on ne peut pas lire
|
|
sur la seule voie d'accès à la machine ; ce script tourne au bout d'un
|
|
ssh, et une VM imbriquée n'a pas d'autre porte.
|
|
|
|
Il n'est pas nécessaire au but : le stockage et le suivi demandent
|
|
pve-cluster et pvestatd, l'interface web pveproxy."""
|
|
import re
|
|
|
|
m = re.search(r'PVE_SERVICES="([^"]+)"', self.src)
|
|
self.assertIsNotNone(m)
|
|
self.assertNotIn("pve-firewall", m.group(1).split())
|
|
|
|
def test_the_freeze_is_guarded_on_content(self):
|
|
"""« printf … > fichier » TRONQUE avant d'écrire.
|
|
|
|
Une coupure au mauvais moment laisse zéro octet, et une garde à
|
|
l'EXISTENCE annonce « déjà gelé » pour toujours : cloud-init continue
|
|
de remettre 127.0.1.1 à chaque démarrage et le défaut redevient
|
|
invisible."""
|
|
bloc = self.src[self.src.index("freeze_cloud_hosts() {") :]
|
|
bloc = bloc[: bloc.index("\nfix_hosts()")]
|
|
self.assertIn("manage_etc_hosts:[[:space:]]*false", bloc)
|
|
self.assertNotIn('[ -f "${fichier}" ]', bloc)
|
|
|
|
def test_the_first_apt_survives_the_boot_time_lock(self):
|
|
"""Mesuré une seconde après le premier ssh d'une image cloud :
|
|
|
|
E: Could not get lock /var/lib/apt/lists/lock.
|
|
It is held by process 1026 (apt-get)
|
|
|
|
Ce n'est pas cloud-init — « status --wait » avait rendu la main. C'est
|
|
apt-daily, qui se déclenche au démarrage. Et le verrou des LISTES
|
|
n'est pas couvert par « DPkg::Lock::Timeout », qui ne vaut que pour
|
|
celui de dpkg."""
|
|
self.assertIn("prepare_apt", self.src)
|
|
bloc = self.src[self.src.index("prepare_apt() {") :]
|
|
bloc = bloc[: bloc.index("\ninstall_pve()")]
|
|
self.assertIn("apt-daily", bloc)
|
|
# Arrêter le minuteur n'interrompt pas l'apt-get déjà en vol : il faut
|
|
# RÉESSAYER, pas seulement stopper.
|
|
self.assertIn("for i in", bloc)
|
|
self.assertIn("nouvel essai", bloc)
|
|
|
|
def test_the_retry_loop_really_retries(self):
|
|
"""Exécutée, apt_get bouchonné : elle doit insister puis rendre 0."""
|
|
import re
|
|
import subprocess
|
|
|
|
fonction = re.search(
|
|
r"^prepare_apt\(\) \{.*?^\}", self.src, re.M | re.S
|
|
)
|
|
self.assertIsNotNone(fonction)
|
|
shell = (
|
|
"say() { :; }; die() { exit 9; }; run() { :; }; sudo() { :; }; "
|
|
"sleep() { :; }; N=0; "
|
|
"apt_get() { N=$((N+1)); [ $N -ge 3 ] && return 0 || return 100; };"
|
|
+ fonction.group(0)
|
|
+ '\nprepare_apt && echo "ESSAIS $N"'
|
|
)
|
|
res = subprocess.run(
|
|
["bash", "-c", shell], capture_output=True, text=True, timeout=60
|
|
)
|
|
self.assertEqual(res.returncode, 0, res.stderr)
|
|
self.assertIn("ESSAIS 3", res.stdout)
|
|
|
|
def test_the_retry_loop_gives_up_loudly(self):
|
|
# Une boucle qui abandonne en silence laisserait « apt update » échoué
|
|
# passer pour un succès.
|
|
import re
|
|
import subprocess
|
|
|
|
fonction = re.search(
|
|
r"^prepare_apt\(\) \{.*?^\}", self.src, re.M | re.S
|
|
)
|
|
shell = (
|
|
"say() { :; }; die() { echo ABANDON; exit 9; }; run() { :; }; "
|
|
"sudo() { :; }; sleep() { :; }; apt_get() { return 100; };"
|
|
+ fonction.group(0)
|
|
+ "\nprepare_apt"
|
|
)
|
|
res = subprocess.run(
|
|
["bash", "-c", shell], capture_output=True, text=True, timeout=60
|
|
)
|
|
self.assertEqual(res.returncode, 9)
|
|
self.assertIn("ABANDON", res.stdout)
|
|
|
|
def test_the_mount_is_verified_not_assumed(self):
|
|
self.assertIn("/etc/pve/.version", self.src)
|
|
|
|
def test_the_script_is_valid_shell(self):
|
|
import subprocess
|
|
|
|
res = subprocess.run(
|
|
["bash", "-n", "script/proxmox/install_proxmox.sh"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertEqual(res.returncode, 0, res.stderr)
|
|
|
|
|
|
class TestReparerEtcHosts(unittest.TestCase):
|
|
"""La réécriture de /etc/hosts, EXÉCUTÉE sur de faux fichiers.
|
|
|
|
Trois hôtes de suite sont tombés sur la même panne, et le conseil
|
|
« rejouer install_proxmox.sh » ne pouvait pas la corriger : la VM clone le
|
|
dépôt distant, donc sa copie du script est celle qui ne corrige rien.
|
|
L'outil répare donc lui-même — et une réécriture de /etc/hosts sur une
|
|
machine qu'on ne joint que par ssh doit être ÉPROUVÉE, pas relue.
|
|
|
|
Aucun bouchon de vérification ici : la commande relit elle-même ce qu'elle
|
|
a écrit. La première version s'en remettait à « getent hosts $short », qui
|
|
réussit via mDNS même quand rien n'a été écrit — et les tests bouchonnaient
|
|
getent à « return 0 », donc ils mesuraient le bouchon."""
|
|
|
|
def _joue(self, contenu, court="pve", passages=3, ecrivable=True):
|
|
"""Rejoue la commande RÉELLE `passages` fois sur un faux /etc/hosts."""
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
|
|
d = tempfile.mkdtemp()
|
|
hosts = os.path.join(d, "hosts")
|
|
with open(hosts, "w", encoding="utf-8") as fh:
|
|
fh.write(contenu)
|
|
cmd = pve.hosts_repair_cmd("10.10.10.150").replace("/etc/hosts", hosts)
|
|
if not ecrivable:
|
|
os.chmod(d, 0o500)
|
|
verdicts = []
|
|
try:
|
|
for _ in range(passages):
|
|
res = subprocess.run(
|
|
["sh", "-c", f"hostname() {{ echo {court}; }}; " + cmd],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
verdicts.append(res.stdout.strip())
|
|
finally:
|
|
os.chmod(d, 0o700)
|
|
with open(hosts, encoding="utf-8") as fh:
|
|
brut = fh.read()
|
|
restes = [f for f in os.listdir(d) if f != "hosts"]
|
|
return {
|
|
"lignes": [ligne for ligne in brut.splitlines() if ligne.strip()],
|
|
"verdicts": verdicts,
|
|
"brut": brut,
|
|
"restes": restes,
|
|
}
|
|
|
|
def test_the_cloud_init_line_is_replaced(self):
|
|
vu = self._joue("127.0.1.1 pve pve\n127.0.0.1 localhost\n")
|
|
self.assertEqual(vu["verdicts"], ["HOSTS-OK"] * 3)
|
|
self.assertIn("10.10.10.150\tpve pve\t# erplibre-hosts", vu["lignes"])
|
|
self.assertFalse(
|
|
[ligne for ligne in vu["lignes"] if ligne.startswith("127.0.1.1")]
|
|
)
|
|
|
|
def test_a_refused_write_leaves_the_file_ALONE(self):
|
|
"""Le constat le plus grave de l'attaque, mesuré sur trois états
|
|
réels : /etc en lecture seule, fichier immuable, quota atteint.
|
|
|
|
« sed -i » puis « printf >> » étaient DEUX écritures. Sed refusé et
|
|
ajout réussi, la ligne 127.0.1.1 survivait EN PREMIER et notre ligne
|
|
s'ajoutait une fois par tentative. Sed réussi et ajout refusé, l'hôte
|
|
perdait l'entrée de son nom — et sur une machine qu'on ne joint que
|
|
par ssh, chaque sudo attend ensuite le résolveur.
|
|
|
|
Une seule écriture, la dernière, et elle est vérifiée avant."""
|
|
vu = self._joue(
|
|
"127.0.1.1 pve.lan pve\n127.0.0.1 localhost\n", ecrivable=False
|
|
)
|
|
self.assertEqual(vu["verdicts"], ["HOSTS-KO"] * 3)
|
|
self.assertEqual(
|
|
vu["lignes"], ["127.0.1.1 pve.lan pve", "127.0.0.1 localhost"]
|
|
)
|
|
self.assertEqual(vu["restes"], [], "aucun temporaire ne doit rester")
|
|
|
|
def test_a_file_without_a_final_newline(self):
|
|
"""cloud-init « write_files » n'en met pas.
|
|
|
|
sed PRÉSERVE l'absence — vérifié — et notre ligne se collait à la
|
|
précédente : « 192.168.1.9 autre-machine10.10.10.150 pve », donc le
|
|
nom du nœud résolvait vers l'adresse d'une AUTRE machine. awk émet un
|
|
saut de ligne par enregistrement, donc il normalise."""
|
|
vu = self._joue(
|
|
"127.0.0.1 localhost\n127.0.1.1 pve\n192.168.1.9 autre-machine"
|
|
)
|
|
self.assertEqual(vu["verdicts"], ["HOSTS-OK"] * 3)
|
|
self.assertIn("192.168.1.9 autre-machine", vu["lignes"])
|
|
self.assertIn("10.10.10.150\tpve\t# erplibre-hosts", vu["lignes"])
|
|
self.assertTrue(vu["brut"].endswith("\n"))
|
|
|
|
def test_a_real_fqdn_survives_every_pass(self):
|
|
"""Le défaut que le TROISIÈME passage a révélé.
|
|
|
|
Rejouée, la commande ne trouve plus de ligne 127.0.1.1 — c'est elle
|
|
qui l'a retirée — et retombait sur « <court>.local ». Un vrai FQDN
|
|
était donc remplacé par un nom réservé au mDNS, au deuxième passage,
|
|
par la réparation elle-même."""
|
|
vu = self._joue(
|
|
"127.0.1.1\tpve.lan.example.com pve\n127.0.0.1 localhost\n"
|
|
)
|
|
self.assertIn(
|
|
"10.10.10.150\tpve.lan.example.com pve\t# erplibre-hosts",
|
|
vu["lignes"],
|
|
)
|
|
self.assertNotIn("pve.local", " ".join(vu["lignes"]))
|
|
|
|
def test_nothing_accumulates(self):
|
|
# En DHCP l'adresse change : sans marqueur, une ligne s'ajoutait à
|
|
# chaque passage sans que la précédente soit retirée.
|
|
for contenu in (
|
|
"127.0.1.1 pve pve\n",
|
|
"10.0.0.9\tpve.lan.example.com pve\t# erplibre-hosts\n",
|
|
):
|
|
with self.subTest(depart=contenu.strip()):
|
|
vu = self._joue(contenu, passages=4)
|
|
marquees = [
|
|
ligne
|
|
for ligne in vu["lignes"]
|
|
if "erplibre-hosts" in ligne
|
|
]
|
|
self.assertEqual(len(marquees), 1, vu["lignes"])
|
|
|
|
def test_tabs_everywhere_do_not_duplicate_the_short_name(self):
|
|
"""L'installeur Debian écrit /etc/hosts avec des TABULATIONS.
|
|
|
|
Le test du nom court cherchait des ESPACES : « pve.example.com\tpve »
|
|
ne contenait pas « pve » entouré d'espaces, et le rejeu écrivait
|
|
« pve.example.com pve pve »."""
|
|
vu = self._joue(
|
|
"127.0.0.1\tlocalhost\n127.0.1.1\tpve.example.com\tpve\n"
|
|
)
|
|
self.assertIn(
|
|
"10.10.10.150\tpve.example.com pve\t# erplibre-hosts",
|
|
vu["lignes"],
|
|
)
|
|
|
|
def test_a_trailing_comment_is_stripped(self):
|
|
vu = self._joue("127.0.1.1 pve # posé à la main\n")
|
|
self.assertEqual(vu["lignes"], ["10.10.10.150\tpve\t# erplibre-hosts"])
|
|
|
|
def test_the_short_name_is_always_there(self):
|
|
# C'est lui que pmxcfs résout : une ligne sans lui ne sert à rien.
|
|
vu = self._joue("127.0.1.1 autre-nom\n", court="pve")
|
|
self.assertIn("pve", vu["lignes"][0].split())
|
|
|
|
def test_an_unusable_address_produces_no_command(self):
|
|
for mauvaise in (
|
|
"",
|
|
"127.0.0.1",
|
|
"fe80::1",
|
|
"169.254.3.4",
|
|
"pas-une-ip",
|
|
):
|
|
with self.subTest(ip=mauvaise):
|
|
self.assertEqual(pve.hosts_repair_cmd(mauvaise), "")
|
|
|
|
def test_no_sudo_in_the_body(self):
|
|
"""wrap_privilege porte le privilège, pas le corps.
|
|
|
|
Sur un hôte root@ il n'enrobe rien — et un Proxmox installé par l'ISO
|
|
n'a pas forcément le paquet sudo : « sh: 1: sudo: not found », code
|
|
127, au milieu d'une réécriture de /etc/hosts."""
|
|
for cmd in (
|
|
pve.hosts_repair_cmd("10.0.0.1"),
|
|
pve.cloud_hosts_freeze_cmd(),
|
|
pve.mount_wait_cmd(),
|
|
) + tuple(pve.pve_unit_cmd(u) for u in pve.PVE_UNITS):
|
|
with self.subTest(cmd=cmd[:40]):
|
|
self.assertNotIn("sudo", cmd)
|
|
|
|
|
|
class TestGelerCloudInit(unittest.TestCase):
|
|
"""Le gel EXÉCUTÉ, y compris sur le fichier tronqué à zéro octet."""
|
|
|
|
def _joue(self, etat):
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
|
|
racine = tempfile.mkdtemp()
|
|
dossier = os.path.join(racine, "cloud.cfg.d")
|
|
fichier = os.path.join(dossier, "99-erplibre-hosts.cfg")
|
|
if etat != "sans-cloud":
|
|
os.makedirs(dossier)
|
|
if etat == "vide":
|
|
open(fichier, "w").close()
|
|
elif etat == "gele":
|
|
with open(fichier, "w", encoding="utf-8") as fh:
|
|
fh.write("manage_etc_hosts: false\n")
|
|
cmd = (
|
|
pve.cloud_hosts_freeze_cmd()
|
|
.replace("/etc/cloud/cloud.cfg.d", dossier)
|
|
.replace(
|
|
"/etc/cloud", racine if etat != "sans-cloud" else "/nexistepas"
|
|
)
|
|
)
|
|
res = subprocess.run(["sh", "-c", cmd], capture_output=True, text=True)
|
|
contenu = ""
|
|
if os.path.exists(fichier):
|
|
with open(fichier, encoding="utf-8") as fh:
|
|
contenu = fh.read()
|
|
return res.stdout.strip(), contenu
|
|
|
|
def test_a_fresh_host_gets_frozen(self):
|
|
verdict, contenu = self._joue("neuf")
|
|
self.assertEqual(verdict, "FREEZE-OK")
|
|
self.assertIn("manage_etc_hosts: false", contenu)
|
|
|
|
def test_an_empty_file_is_rewritten(self):
|
|
"""Le défaut que la garde à l'EXISTENCE laissait passer.
|
|
|
|
« printf … > » TRONQUE avant d'écrire : une coupure laisse zéro octet,
|
|
et la garde annonçait « déjà gelé » pour toujours. cloud-init
|
|
continuait de remettre 127.0.1.1 à chaque démarrage."""
|
|
verdict, contenu = self._joue("vide")
|
|
self.assertEqual(verdict, "FREEZE-OK")
|
|
self.assertIn("manage_etc_hosts: false", contenu)
|
|
|
|
def test_an_already_frozen_host_is_left_alone(self):
|
|
verdict, _c = self._joue("gele")
|
|
self.assertEqual(verdict, "FREEZE-DEJA")
|
|
|
|
def test_a_host_without_cloud_init_says_so(self):
|
|
verdict, _c = self._joue("sans-cloud")
|
|
self.assertEqual(verdict, "FREEZE-SANS-OBJET")
|
|
|
|
|
|
class TestQuelleAdressePourLeNoeud(unittest.TestCase):
|
|
"""L'adresse écrite doit être celle par laquelle on JOINT l'hôte.
|
|
|
|
Mesuré sur une Proxmox imbriquée : « hostname -I » rend
|
|
« 10.10.10.150 10.10.20.1 », et la seconde est le pont interne que notre
|
|
propre code vient de créer. La poser ferait s'identifier le nœud par une
|
|
adresse que personne ne joint."""
|
|
|
|
def test_the_server_field_of_ssh_connection(self):
|
|
self.assertEqual(
|
|
pve.ssh_server_ip("10.10.10.1 33580 10.10.10.150 22"),
|
|
"10.10.10.150",
|
|
)
|
|
|
|
def test_ssh_noise_does_not_shift_the_fields(self):
|
|
brut = (
|
|
"Warning: Permanently added 'x' (ED25519) to the list of known"
|
|
" hosts.\n10.10.10.1 33580 10.10.10.150 22"
|
|
)
|
|
self.assertEqual(pve.ssh_server_ip(brut), "10.10.10.150")
|
|
|
|
def test_an_empty_or_short_value_gives_nothing(self):
|
|
for brut in ("", "10.0.0.1 22", "n'importe quoi"):
|
|
with self.subTest(brut=brut):
|
|
self.assertEqual(pve.ssh_server_ip(brut), "")
|
|
|
|
def test_a_loopback_server_field_is_refused(self):
|
|
# Un tunnel local peut faire de l'hôte « 127.0.0.1 » : l'écrire dans
|
|
# /etc/hosts ne réglerait rien.
|
|
self.assertEqual(pve.ssh_server_ip("127.0.0.1 5555 127.0.0.1 22"), "")
|
|
|
|
|
|
class TestRelancerLesUnites(unittest.TestCase):
|
|
"""Chaque unité à part, jamais fatale, et le journal quand ça échoue.
|
|
|
|
Les bouchons ÉCHOUENT ici. La première version ne faisait jamais rater un
|
|
« start » : le journalctl bouchonné n'était donc jamais atteint, et
|
|
retirer complètement « reset-failed » de la commande laissait tous les
|
|
tests verts."""
|
|
|
|
def _joue(self, unite, etat, monte, existe=True, start_ok=True, **kw):
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
|
|
temoin = os.path.join(tempfile.mkdtemp(), "version")
|
|
if monte:
|
|
open(temoin, "w").close()
|
|
bouchons = (
|
|
"systemctl() { "
|
|
' case "$1" in '
|
|
f" list-unit-files) return {0 if existe else 1};; "
|
|
f" is-active) echo {etat};; "
|
|
' reset-failed) echo "RESET $2";; '
|
|
f' start|restart) echo "STARTED $1 $2"; '
|
|
f" return {0 if start_ok else 1};; "
|
|
" esac; }; "
|
|
"journalctl() { echo LIGNE-DE-JOURNAL; }; "
|
|
)
|
|
cmd = pve.pve_unit_cmd(unite, **kw).replace(
|
|
"/etc/pve/.version", temoin
|
|
)
|
|
res = subprocess.run(
|
|
["sh", "-c", bouchons + cmd], capture_output=True, text=True
|
|
)
|
|
return res.returncode, res.stdout
|
|
|
|
def test_an_absent_unit_is_skipped_not_fatal(self):
|
|
code, out = self._joue("pveproxy", "failed", False, existe=False)
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("SKIP pveproxy", out)
|
|
|
|
def test_a_failed_unit_is_RESET_then_started(self):
|
|
# Le reset débloque la limite de démarrage : sans lui, systemd refuse
|
|
# le start sans même le tenter. Son absence doit faire ROUGIR le test.
|
|
code, out = self._joue("pvestatd", "failed", False)
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("RESET pvestatd", out)
|
|
self.assertIn("STARTED start", out)
|
|
self.assertLess(out.index("RESET"), out.index("STARTED"))
|
|
|
|
def test_a_start_that_fails_names_the_unit_and_shows_the_journal(self):
|
|
"""La seule façon de dire la cause à quelqu'un dont l'unique accès à
|
|
l'hôte est cet outil."""
|
|
code, out = self._joue("pve-cluster", "failed", False, start_ok=False)
|
|
self.assertEqual(code, 0, "jamais fatale")
|
|
self.assertIn("KO pve-cluster", out)
|
|
self.assertIn("LIGNE-DE-JOURNAL", out)
|
|
|
|
def test_a_stale_mount_gets_a_restart_not_a_start(self):
|
|
"""« start » sur une unité ACTIVE est un no-op qui rend 0.
|
|
|
|
pmxcfs tué par l'OOM killer laisse /etc/pve monté mais mort, l'unité
|
|
pouvant rester « active » : la réparation ne convergeait jamais et ne
|
|
nommait rien."""
|
|
code, out = self._joue("pve-cluster", "active", False)
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("STARTED restart", out)
|
|
|
|
def test_an_active_unit_with_the_mount_is_left_alone(self):
|
|
code, out = self._joue("pve-cluster", "active", True)
|
|
self.assertEqual(code, 0)
|
|
self.assertIn("DEJA pve-cluster", out)
|
|
|
|
def test_the_dependents_are_restarted_when_the_mount_was_absent(self):
|
|
"""Leur état actif ne prouve rien sur leur lien à pmxcfs.
|
|
|
|
pvestatd, pvedaemon et pveproxy tournaient pendant toute la panne, en
|
|
échouant sur ipcc_send_rec. Les laisser après avoir remonté /etc/pve
|
|
donnait une GUI qui répond « communication failure » juste après le ✓
|
|
de la réparation."""
|
|
for unite in ("pvestatd", "pvedaemon", "pveproxy"):
|
|
with self.subTest(unite=unite):
|
|
_c, out = self._joue(unite, "active", True, remonte=True)
|
|
self.assertIn("STARTED restart", out)
|
|
_c, sans = self._joue(unite, "active", True)
|
|
self.assertIn(f"DEJA {unite}", sans)
|
|
|
|
def test_the_firewall_is_not_in_the_list(self):
|
|
# Sa configuration vit dans config.db, invisible tant que /etc/pve
|
|
# n'est pas monté : on appliquerait des règles illisibles sur la seule
|
|
# voie d'accès à la machine.
|
|
self.assertNotIn("pve-firewall", pve.PVE_UNITS)
|
|
|
|
def test_rrdcached_comes_before_pve_cluster(self):
|
|
# pve-cluster le requiert : une limite atteinte sur rrdcached fait
|
|
# échouer pve-cluster sur « dependency », et reset-failed sur
|
|
# pve-cluster n'y change rien.
|
|
units = list(pve.PVE_UNITS)
|
|
self.assertLess(units.index("rrdcached"), units.index("pve-cluster"))
|
|
|
|
|
|
class TestConstaterLeMontage(unittest.TestCase):
|
|
"""Une seule observation ne prouve rien, et un silence n'est pas une
|
|
absence."""
|
|
|
|
def test_a_mount_that_holds(self):
|
|
lu = pve.parse_mount_wait("MONTE\nNRESTARTS 0\n")
|
|
self.assertEqual((lu["verdict"], lu["relances"]), ("MONTE", 0))
|
|
|
|
def test_a_mount_that_flaps_is_not_a_success(self):
|
|
# reset-failed vient d'effacer la limite de relance : un pmxcfs qui
|
|
# battait repart pour une salve entière, et le ✓ serait suivi d'un
|
|
# « pvesm ne répond plus » dix secondes après.
|
|
lu = pve.parse_mount_wait("BATTEMENT\nNRESTARTS 4\n")
|
|
self.assertEqual((lu["verdict"], lu["relances"]), ("BATTEMENT", 4))
|
|
|
|
def test_silence_is_not_absence(self):
|
|
# Conclure « /etc/pve n'est pas monté » d'une perte de contact envoie
|
|
# chercher dans journalctl une panne qui n'existe pas.
|
|
for brut in ("", "timeout", "ssh: connect to host … port 22"):
|
|
with self.subTest(brut=brut):
|
|
self.assertEqual(
|
|
pve.parse_mount_wait(brut)["verdict"], "INCONNU"
|
|
)
|
|
|
|
def test_the_wait_is_a_single_round_trip(self):
|
|
cmd = pve.mount_wait_cmd()
|
|
self.assertIn("while", cmd)
|
|
self.assertIn("sleep", cmd)
|
|
self.assertEqual(cmd.count("NRESTARTS"), 1)
|
|
|
|
def _attends(self, present, disparait=False):
|
|
"""Exécute la commande RÉELLE, sentinelle créée puis retirée."""
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
|
|
temoin = os.path.join(tempfile.mkdtemp(), "version")
|
|
if present:
|
|
open(temoin, "w").close()
|
|
cmd = pve.mount_wait_cmd(tours=2, repos=1).replace(
|
|
"/etc/pve/.version", temoin
|
|
)
|
|
if disparait:
|
|
# Retiré PENDANT la pause de reconfirmation — la SECONDE, celle
|
|
# qui suit « then ». La première est dans la boucle d'attente.
|
|
cmd = cmd.replace(
|
|
"then sleep 1;", f"then rm -f {temoin}; sleep 1;", 1
|
|
)
|
|
res = subprocess.run(
|
|
["sh", "-c", "systemctl() { echo 3; }; " + cmd],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return pve.parse_mount_wait(res.stdout)
|
|
|
|
def test_a_mount_that_holds_is_measured(self):
|
|
self.assertEqual(self._attends(True)["verdict"], "MONTE")
|
|
|
|
def test_a_mount_that_disappears_is_a_flap(self):
|
|
"""La raison d'être de la reconfirmation.
|
|
|
|
reset-failed vient d'effacer la limite de relance, donc un pmxcfs qui
|
|
battait repart pour une salve entière : vu une fois, il peut mourir
|
|
dix secondes après notre ✓."""
|
|
self.assertEqual(
|
|
self._attends(True, disparait=True)["verdict"], "BATTEMENT"
|
|
)
|
|
|
|
def test_a_mount_that_never_comes_is_absent(self):
|
|
self.assertEqual(self._attends(False)["verdict"], "ABSENT")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=1)
|