[FIX] proxmox : sh: 1: Syntax error: "(" unexpected

Rapporté. La chaîne, en trois maillons : sur un hôte sans pont, « ip -o link
show type bridge » ne rend RIEN, la sortie ne contient donc que
l'avertissement de ssh sur la clé d'hôte — que le lecteur 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 ce que l'utilisateur a lu. Le bruit de
ssh est maintenant retiré à la source, et un pont doit avoir la forme d'un
lien pour en être un.

Éprouvé sur l'hôte réel, VM créée puis détruite : le pont ne montait pas
(ifupdown2 accuse « another instance » quand /run/network manque — un
mensonge), le noyau Debian n'a ni module bridge ni table NAT, et une VM en
adresse fixe n'avait aucun résolveur. Le déploiement écrit désormais un
journal par VM sous ~/.erplibre/proxmox-deploy et en donne le chemin.

--- EN ---

Reported. The chain, in three links: on a host with no bridge, "ip -o link
show type bridge" returns NOTHING, so the output holds only ssh's host-key
warning — which the parser took for a bridge name. "(ED25519)" landed in
"--net0 virtio,bridge=…", wrapped in "sudo sh -c", and dash answered what the
user read. Ssh's noise is now stripped at the source, and a bridge must have
the shape of a link to be one.

Proven on the real host, VM created then destroyed: the bridge would not come
up (ifupdown2 claims "another instance" when /run/network is missing — a lie),
the Debian kernel has neither the bridge module nor the NAT table, and a
statically addressed VM had no resolver at all. Deployment now writes one log
per VM under ~/.erplibre/proxmox-deploy and prints its path.

Assisted-by: Claude Opus 5
This commit is contained in:
Mathieu Benoit 2026-08-24 04:17:36 -04:00
parent 651cfd4583
commit 1a8f1525b1
5 changed files with 513 additions and 24 deletions

View file

@ -80,6 +80,67 @@ def wrap_privilege(remote: str, prefix: str) -> str:
return "sudo sh -c " + shlex.quote(remote) return "sudo sh -c " + shlex.quote(remote)
# Ce que ssh écrit de lui-même, et qui n'est pas la réponse de l'hôte. Retiré
# à la source : un avertissement laissé dans la sortie a été pris pour un nom
# de pont par `parse_bridges`, et « (ED25519) » s'est retrouvé dans un
# « qm create » enrobé de « sudo sh -c » — d'où le « sh: 1: Syntax error:
# "(" unexpected » rapporté. Filtrer chez chaque lecteur aurait laissé le
# suivant retomber dans le piège.
_BRUIT_SSH = (
"Warning: Permanently added",
"Pseudo-terminal will not be allocated",
"Connection to ",
"Shared connection to ",
"Killed by signal",
"mesg: ttyname failed",
"stdin: is not a tty",
)
def strip_ssh_noise(text: str) -> str:
"""La sortie de l'hôte, débarrassée de ce que ssh y a ajouté.
Ce sont des lignes de ssh lui-même (clé d'hôte enregistrée, pseudo-terminal
refusé, connexion fermée) : elles n'apprennent rien sur la commande et
n'ont donc rien à faire dans ce qu'on analyse ou affiche.
"""
gardees = [
ligne
for ligne in (text or "").splitlines()
if not ligne.strip().startswith(_BRUIT_SSH)
]
return "\n".join(gardees) + ("\n" if gardees else "")
# Les lignes d'AVANCEMENT : « transferred 1.2 GiB of 3.0 GiB (40%) » répété
# cent fois par « qm set --import-from », les points de wget. Elles ne disent
# qu'une chose, et la dernière la dit aussi bien.
_RE_PROGRES = re.compile(
r"^\s*(transferred\s+[\d.]+|\d+K\s+\.|.*\.{10}.*\d+%)"
)
def collapse_progress(text: str) -> str:
"""Ne garde que la DERNIÈRE ligne de chaque salve d'avancement.
Le journal du premier essai réel faisait 136 lignes, dont cent
« transferred … » : l'erreur utile se lisait au chausse-pied. Un
avancement compte pendant qu'il défile, pas dans un fichier qu'on relit.
"""
sortie, salve = [], 0
for ligne in (text or "").splitlines():
if _RE_PROGRES.match(ligne):
salve += 1
continue
if salve:
sortie.append(f" … {salve} lignes d'avancement …")
salve = 0
sortie.append(ligne)
if salve:
sortie.append(f" … {salve} lignes d'avancement …")
return "\n".join(sortie)
def run(host: dict, remote: str, timeout: int = 120) -> tuple: def run(host: dict, remote: str, timeout: int = 120) -> tuple:
"""(code, sortie) de `remote` exécuté sur l'hôte. Ne lève jamais. """(code, sortie) de `remote` exécuté sur l'hôte. Ne lève jamais.
@ -98,7 +159,9 @@ def run(host: dict, remote: str, timeout: int = 120) -> tuple:
return 255, "timeout" return 255, "timeout"
except (OSError, subprocess.SubprocessError) as exc: except (OSError, subprocess.SubprocessError) as exc:
return 255, str(exc) return 255, str(exc)
return res.returncode, (res.stdout or "") + (res.stderr or "") return res.returncode, strip_ssh_noise(
(res.stdout or "") + (res.stderr or "")
)
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
@ -115,6 +178,46 @@ def parse_pveversion(text: str) -> str:
return m.group(1) if m else "" return m.group(1) if m else ""
# D'abord le fichier de systemd-resolved, qui porte les serveurs RÉELS :
# /etc/resolv.conf n'y renvoie qu'un stub sur 127.0.0.53, inutilisable pour un
# invité. On tente les deux, dans cet ordre.
RESOLV_CMD = (
"cat /run/systemd/resolve/resolv.conf 2>/dev/null || cat /etc/resolv.conf"
)
def parse_nameservers(text: str) -> list:
"""Résolveurs UTILISABLES PAR UN INVITÉ, tirés d'un resolv.conf.
Les adresses de boucle sont écartées : « nameserver 127.0.0.53 » est le
stub de systemd-resolved, qui n'existe que sur l'hôte. Une VM qui le
reçoit n'a pas de DNS — mesuré, la VM d'essai ne résolvait rien alors que
le NAT marchait, et « apt update » aurait échoué sans rien expliquer.
"""
serveurs = []
for ligne in (text or "").splitlines():
parts = ligne.split()
if len(parts) >= 2 and parts[0] == "nameserver":
adresse = parts[1].strip()
if adresse.startswith("127.") or adresse in ("::1", "localhost"):
continue
if adresse not in serveurs:
serveurs.append(adresse)
return serveurs
def parse_kernel(text: str) -> str:
"""Noyau ANNONCÉ par pveversion, ou ''.
« pve-manager/9.2.11/abc (running kernel: 6.12.95+deb13-cloud-amd64) » ->
« 6.12.95+deb13-cloud-amd64 ». Ce n'est pas un détail : tant que l'hôte
tourne le noyau de la distribution, il n'a ni le module bridge ni la table
NAT, donc pas de pont et pas de VM.
"""
trouve = re.search(r"running kernel:\s*([^)\s]+)", text or "")
return trouve.group(1) if trouve else ""
def parse_qm_list(text: str) -> list: def parse_qm_list(text: str) -> list:
"""Sortie de « qm list » -> [{vmid, name, status, mem, disk}]. """Sortie de « qm list » -> [{vmid, name, status, mem, disk}].
@ -169,15 +272,24 @@ def parse_storages(text: str) -> list:
return out return out
# « 2: vmbr0: <BROADCAST,MULTICAST,UP> mtu 1500 … » — l'index, le nom, les
# drapeaux. Exiger cette forme, et pas « quelque chose avant deux-points » :
# n'importe quelle ligne de bruit devenait sinon un nom de pont.
_RE_LIEN = re.compile(r"^\s*\d+:\s*([A-Za-z0-9][A-Za-z0-9._@-]*):\s*<")
def parse_bridges(text: str) -> list: def parse_bridges(text: str) -> list:
"""Sortie de « ip -o link show type bridge » -> ['vmbr0', …].""" """Sortie de « ip -o link show type bridge » -> ['vmbr0', …].
Rien d'autre ne passe : un avertissement de ssh a déjà été pris pour un
pont, et son « (ED25519) » a fait échouer le « qm create » qui suivait sur
une erreur de syntaxe shell incompréhensible.
"""
ponts = [] ponts = []
for ligne in (text or "").splitlines(): for ligne in (text or "").splitlines():
parts = ligne.split(":") trouve = _RE_LIEN.match(ligne)
if len(parts) > 1: if trouve:
nom = parts[1].strip().split("@")[0] ponts.append(trouve.group(1).split("@")[0])
if nom:
ponts.append(nom)
return ponts return ponts
@ -354,7 +466,18 @@ def bridge_setup_cmds(
) )
# ifup plutôt qu'« ifreload -a » : recharger TOUTE la configuration d'un # ifup plutôt qu'« ifreload -a » : recharger TOUTE la configuration d'un
# hôte distant peut emporter l'interface qui porte la session. # hôte distant peut emporter l'interface qui porte la session.
cmds.append(f"ifup {nom} 2>/dev/null || ifreload -a") #
# « mkdir -p /run/network » d'abord : ifupdown2 y pose son verrou, et
# quand le répertoire manque il annonce « Another instance of this program
# is already running » — son lockFile() attrape aussi le fichier
# introuvable. Le message est un MENSONGE, et il a caché deux heures la
# vraie panne. Sur une Debian installée en image cloud, networking.service
# n'a jamais démarré, donc personne n'a créé le répertoire.
#
# Et l'erreur d'ifup n'est PAS masquée : « 2>/dev/null » cachait
# « operation failed with 'Operation not supported' » — le noyau cloud n'a
# pas le module bridge, et c'est ce qu'il fallait lire.
cmds.append(f"mkdir -p /run/network; ifup {nom} || ifreload -a")
return cmds return cmds
@ -398,7 +521,7 @@ def image_fetch_cmd(url: str, nom: str, repertoire: str = IMAGE_DIR) -> str:
f"mkdir -p {shlex.quote(repertoire)} && " f"mkdir -p {shlex.quote(repertoire)} && "
f"if [ -s {shlex.quote(cible)} ]; then " f"if [ -s {shlex.quote(cible)} ]; then "
f'echo "image déjà présente : {cible}"; else ' f'echo "image déjà présente : {cible}"; else '
f"wget -q --show-progress -O {shlex.quote(cible)} {shlex.quote(url)}; " f"wget -nv -O {shlex.quote(cible)} {shlex.quote(url)}; "
f"fi" f"fi"
) )
@ -446,6 +569,11 @@ def create_cmds(vmid: int, spec: dict) -> list:
if spec.get("password"): if spec.get("password"):
ci += f" --cipassword {shlex.quote(spec['password'])}" ci += f" --cipassword {shlex.quote(spec['password'])}"
ci += f" --ipconfig0 {spec.get('ipconfig') or 'ip=dhcp'}" ci += f" --ipconfig0 {spec.get('ipconfig') or 'ip=dhcp'}"
# « --ipconfig0 » ne porte PAS le DNS : une VM en adresse fixe n'a alors
# aucun résolveur, et rien ne le dit. En DHCP le bail s'en charge.
serveurs = [s for s in (spec.get("nameservers") or ()) if s]
if serveurs and "dhcp" not in (spec.get("ipconfig") or "dhcp"):
ci += f" --nameserver {shlex.quote(' '.join(serveurs))}"
cmds.append(ci) cmds.append(ci)
# 5. La taille. L'image cloud fait 2 Gio : sans agrandissement, il ne reste # 5. La taille. L'image cloud fait 2 Gio : sans agrandissement, il ne reste
# rien pour installer quoi que ce soit. # rien pour installer quoi que ce soit.

View file

@ -81,6 +81,9 @@ def build_spec(vms, existants, form):
"host": form["host"], "host": form["host"],
"storage": form["storage"], "storage": form["storage"],
"bridge": form["bridge"], "bridge": form["bridge"],
# Les résolveurs de l'hôte suivent la spec : une VM en adresse fixe
# n'a pas de DNS sans eux.
"nameservers": form.get("nameservers") or (),
"res_label": form["res_label"], "res_label": form["res_label"],
"vms": [vm for vm in vms if vm["name"] not in connus], "vms": [vm for vm in vms if vm["name"] not in connus],
"existing": [vm["name"] for vm in vms if vm["name"] in connus], "existing": [vm["name"] for vm in vms if vm["name"] in connus],
@ -692,6 +695,7 @@ def run_proxmox_form(ctx, run_app: bool = True):
"host": ctx["host"], "host": ctx["host"],
"storage": self._storage(), "storage": self._storage(),
"bridge": self._bridge(), "bridge": self._bridge(),
"nameservers": ctx.get("nameservers") or (),
"res_label": res_label(self.profile), "res_label": res_label(self.profile),
"ssh_key": os.path.expanduser(cle) if cle else "", "ssh_key": os.path.expanduser(cle) if cle else "",
"start": self.query_one("#f_start", Checkbox).value, "start": self.query_one("#f_start", Checkbox).value,

View file

@ -316,6 +316,16 @@ class ProxmoxMenuMixin:
print(f" ✓ sudo") print(f" ✓ sudo")
host = dict(host, version=version, sudo=prefixe) host = dict(host, version=version, sudo=prefixe)
print(f" ✓ Proxmox VE {version}") print(f" ✓ Proxmox VE {version}")
# Le noyau DÉCIDE de ce qui marche : sans le noyau Proxmox, ni module
# bridge ni table NAT — donc aucun pont à créer et aucune VM à
# démarrer. Vécu sur l'hôte d'essai, où ifupdown2 répondait
# « Another instance of this program is already running » au lieu de
# « Operation not supported ». On le dit ici, une fois, plutôt que de
# laisser chercher.
noyau = pve.parse_kernel(out)
if noyau and "-pve" not in noyau:
print(f" ⚠ {t('Still on the distribution kernel:')} {noyau}")
print(f" → {t('Reboot the host: no bridge, no NAT until then.')}")
self._pve_remember_host(host) self._pve_remember_host(host)
return host return host
@ -673,6 +683,11 @@ class ProxmoxMenuMixin:
_c, cfg = self._pve_show("cat /etc/network/interfaces", quiet=True) _c, cfg = self._pve_show("cat /etc/network/interfaces", quiet=True)
infos = pve.parse_bridge_config(cfg) infos = pve.parse_bridge_config(cfg)
cpu, ram_libre = self._pve_capacity() cpu, ram_libre = self._pve_capacity()
# Le DNS de l'hôte, pour les VM en adresse fixe : sans lui elles
# routent mais ne résolvent rien, et « apt update » échoue sans que
# rien ne l'explique. Mesuré sur la VM d'essai.
_c, resolv = self._pve_show(pve.RESOLV_CMD, quiet=True)
serveurs_dns = pve.parse_nameservers(resolv)
def ipconfig(pont, vmid): def ipconfig(pont, vmid):
return pve.ipconfig_for(infos.get(pont, {}), vmid) return pve.ipconfig_for(infos.get(pont, {}), vmid)
@ -701,6 +716,7 @@ class ProxmoxMenuMixin:
"bridges": ponts, "bridges": ponts,
"bridge": pve.pick_bridge(ponts), "bridge": pve.pick_bridge(ponts),
"ipconfig": ipconfig, "ipconfig": ipconfig,
"nameservers": serveurs_dns,
"build_command": build_command, "build_command": build_command,
"branches": self._qemu_branch_list() or ["master"], "branches": self._qemu_branch_list() or ["master"],
"install_profiles": self._qemu_install_profiles(), "install_profiles": self._qemu_install_profiles(),
@ -767,6 +783,9 @@ class ProxmoxMenuMixin:
"user": spec.get("user") or "erplibre", "user": spec.get("user") or "erplibre",
"start": spec.get("start", True), "start": spec.get("start", True),
"ipconfig": vm.get("ipconfig") or "ip=dhcp", "ipconfig": vm.get("ipconfig") or "ip=dhcp",
# Le DNS de l'hôte : « --ipconfig0 » ne le porte pas, et une VM
# en adresse fixe se retrouvait sans résolveur.
"nameservers": spec.get("nameservers") or (),
} }
if spec.get("sshkey_path"): if spec.get("sshkey_path"):
detail["sshkey_path"] = spec["sshkey_path"] detail["sshkey_path"] = spec["sshkey_path"]
@ -784,6 +803,17 @@ class ProxmoxMenuMixin:
from script.proxmox import proxmox_deploy as pve from script.proxmox import proxmox_deploy as pve
from script.todo.deploy_form_lib import run_deploy_progress from script.todo.deploy_form_lib import run_deploy_progress
# Le stockage et le pont AVANT tout : l'écran les vérifie déjà, mais
# cette méthode s'appelle aussi d'ailleurs. Sans ce garde-fou, on
# téléchargeait 350 Mio d'image pour finir sur « net0: invalid format
# - missing key » — vécu sur l'hôte d'essai.
for valeur, message in (
(spec.get("storage"), t("No storage able to hold a VM disk.")),
(spec.get("bridge"), t("No bridge on the host.")),
):
if not valeur:
print(f"\n ✗ {message}")
return
if not dry_run and not self._pve_confirm_spec(host, spec): if not dry_run and not self._pve_confirm_spec(host, spec):
print(t("Cancelled.")) print(t("Cancelled."))
return return
@ -793,9 +823,10 @@ class ProxmoxMenuMixin:
spec["sshkey_path"] = "/root/.ssh/erplibre-deploy.pub" spec["sshkey_path"] = "/root/.ssh/erplibre-deploy.pub"
else: else:
print(f" ⚠ {t('SSH key not pushed: password login only.')}") print(f" ⚠ {t('SSH key not pushed: password login only.')}")
travaux = [] travaux, commandes = [], {}
for vm in spec["vms"]: for vm in spec["vms"]:
cmds = self._pve_vm_commands(mod, vm, spec) cmds = self._pve_vm_commands(mod, vm, spec)
commandes[vm["name"]] = cmds
if dry_run: if dry_run:
print(f"\n── {vm['name']} ({t('VMID')} {vm['vmid']}) ──") print(f"\n── {vm['name']} ({t('VMID')} {vm['vmid']}) ──")
for cmd in cmds: for cmd in cmds:
@ -821,16 +852,93 @@ class ProxmoxMenuMixin:
print(f" ⏭ {t('already there')} : {', '.join(spec['existing'])}") print(f" ⏭ {t('already there')} : {', '.join(spec['existing'])}")
if not travaux: if not travaux:
return return
# Le journal AVANT de lancer : la vue de progression se referme et
# emporte tout ce qu'elle montrait. Rapporté — « il manque plein
# d'informations qu'il y avait avant, où est le fichier de log ? ».
# L'ancienne voie par questions imprimait chaque commande et sa
# sortie ; celle-ci les écrit, ce qui vaut mieux qu'un défilement.
session = self._pve_log_dir()
print(f"\n {t('Log:')} {session}")
resultats = run_deploy_progress(travaux, spec.get("parallelism") or 1) resultats = run_deploy_progress(travaux, spec.get("parallelism") or 1)
reussies = [nom for nom, code, _o, _d in resultats if code == 0] reussies = [nom for nom, code, _o, _d in resultats if code == 0]
for nom, code, sortie, _duree in resultats: for nom, code, sortie, duree in resultats:
chemin = self._pve_write_log(
session,
nom,
spec,
commandes.get(nom) or [],
code,
sortie,
duree,
)
marque = "✓" if code == 0 else "✗"
print(f" {marque} {nom} : {chemin}")
if code: if code:
print(f"\n ✗ {nom} : {t('exit code')} {code}") print(f" {t('exit code')} {code}")
print("\n".join(sortie.rstrip().splitlines()[-12:])) # Les dernières lignes à l'écran, le reste dans le journal :
# c'est l'échec qu'on veut lire tout de suite.
propre = pve.collapse_progress(pve.strip_ssh_noise(sortie))
for ligne in propre.rstrip().splitlines()[-12:]:
print(f" {ligne}")
if not reussies: if not reussies:
return return
self._pve_after_create(host, spec, reussies, cle_locale) self._pve_after_create(host, spec, reussies, cle_locale)
@staticmethod
def _pve_log_dir():
"""Répertoire de journaux de CE déploiement, créé au besoin.
Même esprit que ~/.erplibre/qemu-install : une session par
déploiement, un fichier par VM. La vue de progression se referme ; le
journal reste, et c'est lui qu'on relit quand une étape a cédé."""
session = os.path.join(
os.path.expanduser("~/.erplibre/proxmox-deploy"),
time.strftime("%Y%m%d-%H%M%S"),
)
os.makedirs(session, exist_ok=True)
return session
@staticmethod
def _pve_write_log(session, nom, spec, cmds, code, sortie, duree):
"""Écrit le journal d'UNE VM et rend son chemin.
Les commandes AVANT leur sortie : c'est ce qui rend l'étape rejouable
à la main, et c'est ainsi que les pannes de ce module ont été
diagnostiquées."""
from script.proxmox import proxmox_deploy as pve
chemin = os.path.join(session, f"{nom}.log")
hote = (spec.get("host") or {}).get("target", "?")
vm = next((v for v in spec.get("vms") or [] if v["name"] == nom), {})
entete = [
"=" * 64,
" ERPLibre — création d'une VM sur Proxmox VE",
f" Date : {time.strftime('%Y-%m-%d %H:%M:%S')}",
f" VM : {nom} VMID {vm.get('vmid', '?')}",
f" Hôte : {hote}",
f" Stockage : {spec.get('storage')} "
f"{t('bridge')} : {spec.get('bridge')}",
f" Adresse : {(vm.get('ipconfig') or '').replace('ip=', '')}",
f" Ressources: {vm.get('vcpus', '?')} vCPU "
f"{vm.get('ram', '?')} Mo {vm.get('disk', '?')}",
"=" * 64,
"",
"---- commandes ----",
]
entete += [f" {c}" for c in cmds]
propre = pve.collapse_progress(pve.strip_ssh_noise(sortie or ""))
entete += ["", "---- sortie ----", propre.rstrip(), ""]
entete += [
(
f"---- fin : code {code}, {duree:.0f} s ----"
if isinstance(duree, (int, float))
else f"---- fin : code {code} ----"
)
]
with open(chemin, "w", encoding="utf-8") as fh:
fh.write("\n".join(entete) + "\n")
return chemin
def _pve_confirm_spec(self, host, spec): def _pve_confirm_spec(self, host, spec):
"""Récapitulatif puis confirmation, dans le TERMINAL. """Récapitulatif puis confirmation, dans le TERMINAL.
@ -890,23 +998,45 @@ class ProxmoxMenuMixin:
print(f" ✓ ~/.ssh/config : ssh {vm['name']}") print(f" ✓ ~/.ssh/config : ssh {vm['name']}")
joignables.append(vm) joignables.append(vm)
install = spec.get("install") install = spec.get("install")
if not install or not joignables: # Le suivi vient du DÉPLOIEMENT, pas de l'installation — même règle
# qu'en QEMU/KVM. Sans elle, la case « Suivre l'installation » ne
# commandait rien : décochée, le tableau de bord s'ouvrait quand
# même ; cochée sans rien à installer, il ne s'ouvrait jamais.
suivi = spec.get("monitor", True)
if not joignables or not (install or suivi):
return return
noms = [vm["name"] for vm in joignables] noms = [vm["name"] for vm in joignables]
print(f" {install.get('label') or ''}") if install:
print(f" {install.get('label') or ''}")
# Une commande PAR VM dès qu'elles diffèrent : un Proxmox imbriqué # Une commande PAR VM dès qu'elles diffèrent : un Proxmox imbriqué
# installe son hyperviseur, ses voisines ERPLibre. Une commande # installe son hyperviseur, ses voisines ERPLibre. Une commande
# unique en aurait imposé une aux deux. # unique en aurait imposé une aux deux.
commun = install.get("cmd") or "" commun = (install or {}).get("cmd") or ""
cartes = { cartes = {
vm["name"]: (vm.get("install_cmd") or commun) for vm in joignables vm["name"]: (vm.get("install_cmd") or commun) for vm in joignables
} }
self._qemu_install_erplibre_monitored( finale = cartes if self._qemu_per_vm(cartes, commun) else commun
noms, branche = (install or {}).get("branch") or ""
install.get("branch") or "master", if suivi:
{n: n for n in noms}, # Rien à installer ? La commande distante regarde alors la VM
cartes if self._qemu_per_vm(cartes, commun) else commun, # ARRIVER (cloud-init, puis relevé système) : c'est ce que le
) # tableau de bord montre.
self._qemu_install_erplibre_monitored(
noms, branche, {n: n for n in noms}, finale
)
return
# Sans suivi mais avec quelque chose à installer : en série, sortie à
# l'écran. C'est le pendant exact de la voie QEMU/KVM.
print(f"\n{t('Installing ERPLibre on each VM')} ({branche})…")
for vm in joignables:
self._qemu_install_erplibre_vm(
vm["name"],
cle_locale,
branche,
pve.ip_from_ipconfig(vm.get("ipconfig") or "") or vm["name"],
vm.get("install_cmd") or commun,
False,
)
def _pve_deploy_prompts(self, dry_run=False): def _pve_deploy_prompts(self, dry_run=False):
"""Déploie une VM SUR l'hôte Proxmox choisi, par questions. """Déploie une VM SUR l'hôte Proxmox choisi, par questions.

View file

@ -31,6 +31,7 @@ from unittest import mock
sys.argv = ["todo.py"] sys.argv = ["todo.py"]
from script.proxmox import proxmox_deploy as pve # noqa: E402 from script.proxmox import proxmox_deploy as pve # noqa: E402
from script.todo.todo import TODO # 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. # Sorties RÉELLES relevées sur l'hôte d'essai.
PVEVERSION = ( PVEVERSION = (
@ -42,6 +43,17 @@ AVERTISSEMENT = (
"Warning: Permanently added '192.168.123.227' (ED25519) to the list " "Warning: Permanently added '192.168.123.227' (ED25519) to the list "
"of known hosts.\n" "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 QM_LIST = """ VMID NAME STATUS MEM(MB) BOOTDISK(GB) PID
100 vm-essai running 2048 16.00 2726 100 vm-essai running 2048 16.00 2726
101 avec un espace stopped 4096 32.00 0 101 avec un espace stopped 4096 32.00 0
@ -151,6 +163,143 @@ class TestLectureDesSorties(unittest.TestCase):
self.assertIn("999", orph[0][0]) 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): class TestLesChoix(unittest.TestCase):
def test_the_vmid_skips_the_taken_ones(self): def test_the_vmid_skips_the_taken_ones(self):
"""Proxmox refuse un VMID pris, et le dit APRÈS le téléchargement de """Proxmox refuse un VMID pris, et le dit APRÈS le téléchargement de
@ -359,10 +508,13 @@ class TestChoixDeLHote(unittest.TestCase):
] ]
) )
self.assertIsNone(host) self.assertIsNone(host)
self.assertIn("joignable", sortie.lower()) # 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) self.assertIn("install_proxmox.sh", sortie)
# Et surtout : ne plus envoyer chercher un problème de réseau. # Et surtout : ne plus envoyer chercher un problème de réseau.
self.assertNotIn("SSH ne passe pas", sortie) self.assertNotIn(t("SSH does not get through:"), sortie)
def test_an_unreachable_machine_says_ssh_does_not_get_through(self): 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" panne = "ssh: connect to host 10.0.0.9 port 22: No route to host"

View file

@ -400,5 +400,80 @@ class TestEcran(unittest.TestCase):
self.assertIsNone(self._rendu(gestes).result) self.assertIsNone(self._rendu(gestes).result)
class TestLeSuivi(unittest.TestCase):
"""La case « Suivre l'installation » doit commander quelque chose.
Elle ne commandait rien : décochée, le tableau de bord s'ouvrait quand
même ; cochée sans rien à installer, il ne s'ouvrait jamais. Le suivi
vient du DÉPLOIEMENT, pas de l'installation — c'est la règle déjà tirée du
côté QEMU/KVM après le même rapport.
"""
def _apres_creation(self, install, monitor):
"""Rejoue l'épilogue du déploiement et dit quelle voie a été prise."""
import sys
sys.argv = ["todo.py"]
from script.todo.todo import TODO
todo = TODO.__new__(TODO)
vus = {}
todo._qemu_install_erplibre_monitored = lambda *a, **k: vus.setdefault(
"tableau", a
)
todo._qemu_install_erplibre_vm = lambda *a, **k: vus.setdefault(
"serie", a
)
todo._write_ssh_config_entry = lambda *a, **k: None
todo._ssh_private_key = lambda k: None
todo._pve_guest_ip = lambda vmid, attente=120: ""
spec = {
"host": {"target": "pve1"},
"vms": [
{
"name": "vm-a",
"vmid": 100,
"ipconfig": "ip=10.10.10.150/24,gw=10.10.10.1",
"install_cmd": "",
}
],
"add_ssh_config": False,
"user": "erplibre",
"install": install,
"monitor": monitor,
}
import contextlib
import io
with contextlib.redirect_stdout(io.StringIO()):
todo._pve_after_create(spec["host"], spec, ["vm-a"], "")
return vus
def test_ticked_without_anything_to_install_still_opens_it(self):
# La commande distante regarde alors la VM ARRIVER : c'est justement
# ce qu'on veut voir sur une VM déployée nue.
vus = self._apres_creation(install=None, monitor=True)
self.assertIn("tableau", vus)
self.assertNotIn("serie", vus)
def test_unticked_installs_without_the_dashboard(self):
vus = self._apres_creation(
install={"branch": "develop", "cmd": "make x", "label": "X"},
monitor=False,
)
self.assertIn("serie", vus)
self.assertNotIn("tableau", vus)
def test_unticked_and_nothing_to_install_does_nothing(self):
self.assertEqual(self._apres_creation(install=None, monitor=False), {})
def test_ticked_with_an_install_opens_it(self):
vus = self._apres_creation(
install={"branch": "develop", "cmd": "make x", "label": "X"},
monitor=True,
)
self.assertIn("tableau", vus)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main(verbosity=2) unittest.main(verbosity=2)