[ADD] proxmox : un écran pour déployer sur un hôte distant

Le déploiement Proxmox se faisait par vingt questions, l'une après
l'autre, sans jamais voir le plan entier. Il a maintenant le même écran
que QEMU/KVM — catalogue à gauche, plan à droite, totaux dessous — et une
touche pour revenir aux questions. Il n'en redit rien : 690 lignes contre
1 340, tout le reste vient du socle commun.

Ce que Proxmox a en propre est montré AVANT de lancer : le VMID, parce
que l'hôte ne dit « déjà pris » qu'après avoir téléchargé l'image, et
l'adresse qui s'en déduit sur un pont interne. Le stockage et le pont sont
lus sur l'hôte, jamais devinés. L'icône 🗄 distingue l'entrée du menu de
celle qui déploie ici même.

--- EN ---

Deploying on Proxmox meant twenty questions, one after another, never
seeing the whole plan. It now has the same screen as QEMU/KVM — catalogue
left, plan right, totals below — and a key to fall back to the questions.
It repeats none of it: 690 lines against 1,340, everything else comes
from the shared foundation.

What belongs to Proxmox is shown BEFORE launching: the VMID, because the
host only says "already taken" after downloading the image, and the
address derived from it on an internal bridge. Storage and bridge are read
from the host, never guessed. The 🗄 icon tells the menu entry apart from
the one that deploys right here.

Assisted-by: Claude Opus 5
This commit is contained in:
Mathieu Benoit 2026-08-23 18:01:28 -04:00
parent 0d480af24a
commit 9767ad540f
8 changed files with 1507 additions and 3 deletions

View file

@ -43,6 +43,7 @@ if [[ ${#FILES[@]} -eq 0 ]]; then
# Aucun argument : tout ce que le dépôt sait tester sans base de données.
mapfile -t FILES < <(ls test/test_qemu_*.py test/test_mobile_*.py \
test/test_todo_*.py test/test_install_*.py test/test_run_*.py \
test/test_proxmox_*.py \
2>/dev/null)
fi

View file

@ -13,6 +13,40 @@ The `mail/` package is the mail client reachable from `Assistant > Mail`:
several IMAP/SMTP accounts, a local cache, and a Textual TUI. See
[../../doc/EMAIL.md](../../doc/EMAIL.md).
## Where the code lives
`todo.py` carries the menus and the general helpers. Everything around a
single subject sits in its own file, and the whole thing is assembled by
mixins on the `TODO` class — one file, one subject, its header states its
boundary.
| File | What it owns |
|------|--------------|
| `todo.py` | the menus, the configuration, the general helpers |
| `qemu_menu.py` | the QEMU/KVM menu, the image catalogue, the statistics |
| `qemu_deploy.py` | deciding then running a deployment |
| `qemu_install.py` | the recipes run inside a VM |
| `qemu_manage.py` | lifecycle, disks, hardware, cleanup, addresses |
| `qemu_access.py` | SSH, tunnels, consoles, Android emulator |
| `proxmox_menu.py` | the same, on a REMOTE Proxmox VE host |
The two deployment forms — libvirt here, Proxmox over there — ask the same
questions, so they share a foundation rather than each holding a copy:
| File | What it owns |
|------|--------------|
| `deploy_form_lib.py` | pure logic (sizes, plan, totals, spec), the shared CSS, the resource-row factory, the progress view |
| `deploy_form_plan.py` | the plan's gestures: overrides, locks, copies, renaming, free values |
| `qemu_deploy_form.py` | what QEMU/KVM adds: desktops, tools, branches, install profiles |
| `proxmox_deploy_form.py` | what Proxmox adds: host, storage, bridge, VMID, address |
A form inherits `PlanMixin` and provides three hooks: which presets each
resource offers, the name a VM would fall back to, and what a lock freezes.
`test_todo_deploy_form_lib.py` fails if a form redefines a gesture the
foundation already carries — that is what keeps the architecture from drifting
back into two copies.
<!-- [fr] -->
TODO est un robot assistant pour utiliser ERPLibre
Exécutez-le avec `./script/todo/todo.py` ou `make todo`.
@ -22,3 +56,37 @@ Pour un nouveau projet, copiez todo_example.json vers private/todo/todo_override
Le paquet `mail/` est le client courriel accessible depuis
`Assistant > Courriel` : plusieurs comptes IMAP/SMTP, un cache local, et un
TUI Textual. Voir [../../doc/EMAIL.fr.md](../../doc/EMAIL.fr.md).
## Où vit le code
`todo.py` porte les menus et les aides générales. Tout ce qui tourne autour
d'un même sujet vit dans son fichier, et l'ensemble est assemblé par des
mixins sur la classe `TODO` — un fichier, un sujet, son en-tête dit sa
frontière.
| Fichier | Ce qu'il porte |
|---------|----------------|
| `todo.py` | les menus, la configuration, les aides générales |
| `qemu_menu.py` | le menu QEMU/KVM, le catalogue d'images, les statistiques |
| `qemu_deploy.py` | décider puis exécuter un déploiement |
| `qemu_install.py` | les recettes exécutées DANS une VM |
| `qemu_manage.py` | cycle de vie, disques, matériel, nettoyage, adresses |
| `qemu_access.py` | SSH, tunnels, consoles, émulateur Android |
| `proxmox_menu.py` | la même chose, sur un hôte Proxmox VE DISTANT |
Les deux formulaires de déploiement — libvirt ici, Proxmox ailleurs — posent
les mêmes questions : ils partagent donc un socle au lieu d'en garder chacun
une copie.
| Fichier | Ce qu'il porte |
|---------|----------------|
| `deploy_form_lib.py` | la logique pure (tailles, plan, totaux, spec), le CSS commun, la fabrique des rangées de ressources, la vue de progression |
| `deploy_form_plan.py` | les gestes du plan : surcharges, verrous, exemplaires, renommage, valeurs libres |
| `qemu_deploy_form.py` | ce que QEMU/KVM ajoute : bureaux, outils, branches, profils d'installation |
| `proxmox_deploy_form.py` | ce que Proxmox ajoute : hôte, stockage, pont, VMID, adresse |
Un formulaire hérite de `PlanMixin` et fournit trois crochets : les
préréglages de chaque ressource, le nom auquel une VM retombe, et ce qu'un
verrou fige. `test_todo_deploy_form_lib.py` échoue si un formulaire redit un
geste que le socle porte déjà — c'est ce qui empêche l'architecture de
retomber en deux copies.

View file

@ -6,4 +6,38 @@ Pour un nouveau projet, copiez todo_example.json vers private/todo/todo_override
Le paquet `mail/` est le client courriel accessible depuis
`Assistant > Courriel` : plusieurs comptes IMAP/SMTP, un cache local, et un
TUI Textual. Voir [../../doc/EMAIL.fr.md](../../doc/EMAIL.fr.md).
TUI Textual. Voir [../../doc/EMAIL.fr.md](../../doc/EMAIL.fr.md).
## Où vit le code
`todo.py` porte les menus et les aides générales. Tout ce qui tourne autour
d'un même sujet vit dans son fichier, et l'ensemble est assemblé par des
mixins sur la classe `TODO` — un fichier, un sujet, son en-tête dit sa
frontière.
| Fichier | Ce qu'il porte |
|---------|----------------|
| `todo.py` | les menus, la configuration, les aides générales |
| `qemu_menu.py` | le menu QEMU/KVM, le catalogue d'images, les statistiques |
| `qemu_deploy.py` | décider puis exécuter un déploiement |
| `qemu_install.py` | les recettes exécutées DANS une VM |
| `qemu_manage.py` | cycle de vie, disques, matériel, nettoyage, adresses |
| `qemu_access.py` | SSH, tunnels, consoles, émulateur Android |
| `proxmox_menu.py` | la même chose, sur un hôte Proxmox VE DISTANT |
Les deux formulaires de déploiement — libvirt ici, Proxmox ailleurs — posent
les mêmes questions : ils partagent donc un socle au lieu d'en garder chacun
une copie.
| Fichier | Ce qu'il porte |
|---------|----------------|
| `deploy_form_lib.py` | la logique pure (tailles, plan, totaux, spec), le CSS commun, la fabrique des rangées de ressources, la vue de progression |
| `deploy_form_plan.py` | les gestes du plan : surcharges, verrous, exemplaires, renommage, valeurs libres |
| `qemu_deploy_form.py` | ce que QEMU/KVM ajoute : bureaux, outils, branches, profils d'installation |
| `proxmox_deploy_form.py` | ce que Proxmox ajoute : hôte, stockage, pont, VMID, adresse |
Un formulaire hérite de `PlanMixin` et fournit trois crochets : les
préréglages de chaque ressource, le nom auquel une VM retombe, et ce qu'un
verrou fige. `test_todo_deploy_form_lib.py` échoue si un formulaire redit un
geste que le socle porte déjà — c'est ce qui empêche l'architecture de
retomber en deux copies.

View file

@ -7,3 +7,37 @@ For a new project, copy todo_example.json to private/todo/todo_override.json | p
The `mail/` package is the mail client reachable from `Assistant > Mail`:
several IMAP/SMTP accounts, a local cache, and a Textual TUI. See
[../../doc/EMAIL.md](../../doc/EMAIL.md).
## Where the code lives
`todo.py` carries the menus and the general helpers. Everything around a
single subject sits in its own file, and the whole thing is assembled by
mixins on the `TODO` class — one file, one subject, its header states its
boundary.
| File | What it owns |
|------|--------------|
| `todo.py` | the menus, the configuration, the general helpers |
| `qemu_menu.py` | the QEMU/KVM menu, the image catalogue, the statistics |
| `qemu_deploy.py` | deciding then running a deployment |
| `qemu_install.py` | the recipes run inside a VM |
| `qemu_manage.py` | lifecycle, disks, hardware, cleanup, addresses |
| `qemu_access.py` | SSH, tunnels, consoles, Android emulator |
| `proxmox_menu.py` | the same, on a REMOTE Proxmox VE host |
The two deployment forms — libvirt here, Proxmox over there — ask the same
questions, so they share a foundation rather than each holding a copy:
| File | What it owns |
|------|--------------|
| `deploy_form_lib.py` | pure logic (sizes, plan, totals, spec), the shared CSS, the resource-row factory, the progress view |
| `deploy_form_plan.py` | the plan's gestures: overrides, locks, copies, renaming, free values |
| `qemu_deploy_form.py` | what QEMU/KVM adds: desktops, tools, branches, install profiles |
| `proxmox_deploy_form.py` | what Proxmox adds: host, storage, bridge, VMID, address |
A form inherits `PlanMixin` and provides three hooks: which presets each
resource offers, the name a VM would fall back to, and what a lock freezes.
`test_todo_deploy_form_lib.py` fails if a form redefines a gesture the
foundation already carries — that is what keeps the architecture from drifting
back into two copies.

View file

@ -0,0 +1,687 @@
#!/usr/bin/env python3
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
"""Formulaire de déploiement sur un hôte Proxmox VE.
Même écran que pour QEMU/KVM — catalogue à gauche, plan à droite, totaux
dessous — parce que c'est le même travail : choisir des systèmes, régler des
ressources, vérifier avant de lancer. Tout ce qui est commun vient de
`deploy_form_lib` (logique pure, socle CSS, fabrique des ressources) et de
`deploy_form_plan` (surcharges, verrous, exemplaires, renommage). Ne reste
ici que ce que Proxmox a en propre :
* l'hôte, choisi AVANT d'ouvrir l'écran — il faut ssh et sudo, et une invite
de mot de passe pendant que Textual affiche casserait le terminal ;
* le stockage et le pont, LUS SUR L'HÔTE : « local-lvm » n'existe pas partout
et un pont inventé fait échouer « qm create » ;
* le VMID, et l'adresse qui s'en déduit sur un pont interne.
Le formulaire ne touche à rien : il rend une spec. C'est l'appelant
(`ProxmoxMenuMixin._pve_deploy`) qui exécute.
"""
import os
from script.todo.deploy_form_lib import (
CSS_BASE,
FREE,
RES_FIELDS,
SELECT_TO_FIELD,
build_vms,
entry_key,
plan_rows,
plan_totals,
res_row_widgets,
t,
)
from script.todo.deploy_form_plan import PlanMixin, preview_screen
# Aucun disque orphelin à craindre : les disques d'un Proxmox distant vivent
# dans un stockage que seul l'hôte connaît, jamais dans /var/lib/libvirt.
PAS_D_ORPHELIN = None
def assign_vmids(rows, used, start, ipconfig):
"""Pose un VMID libre et son adresse sur chaque VM À CRÉER.
Proxmox refuse un VMID déjà pris, et il 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. Les VM qui
existent déjà sont sautées — elles ont le leur.
`ipconfig(vmid)` rend la ligne cloud-init : « ip=dhcp » sur un pont qui
donne sur le LAN, une adresse fixe dérivée du VMID sur un pont interne.
"""
pris = {int(v) for v in used or () if str(v).isdigit()}
suivant = max(int(start or 0), 100)
for r in rows:
if r["state"] == "exists":
continue
while suivant in pris:
suivant += 1
pris.add(suivant)
r["vm"]["vmid"] = suivant
r["vm"]["ipconfig"] = ipconfig(suivant) if ipconfig else "ip=dhcp"
suivant += 1
return rows
def res_label(profile) -> str:
"""Comment le plan nomme le réglage commun choisi."""
return t("custom") if profile == "custom" else f"x{profile}"
def build_spec(vms, existants, form):
"""La spec que le déploiement exécutera. Une VM qui existe déjà n'y entre
pas : Proxmox refuserait le VMID, et on ne veut surtout pas l'écraser."""
connus = set(existants)
return {
"host": form["host"],
"storage": form["storage"],
"bridge": form["bridge"],
"res_label": form["res_label"],
"vms": [vm for vm in vms if vm["name"] not in connus],
"existing": [vm["name"] for vm in vms if vm["name"] in connus],
"ssh_key": form["ssh_key"],
"user": form.get("user") or "erplibre",
"start": form["start"],
"add_ssh_config": form["add_ssh_config"],
"install": form["install"],
"monitor": form["monitor"],
"parallelism": form["parallelism"],
}
def run_proxmox_form(ctx, run_app: bool = True):
"""Formulaire Proxmox. Renvoie une spec, None si annulé, {} pour retomber
sur les invites textuelles. `run_app=False` rend l'instance sans la lancer
(tests headless)."""
from textual.app import App, ComposeResult
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.widgets import (
Button,
Checkbox,
Footer,
Header,
Input,
RadioButton,
RadioSet,
Select,
SelectionList,
Static,
)
SELECT_NULL = getattr(Select, "NULL", Select.BLANK)
arches = ctx["arches"]
catalog = ctx["catalog"]
noms_pris = ctx["names"]
vmids_pris = ctx["vmids"]
branches = ctx.get("branches") or ["master"]
profiles = ctx.get("install_profiles") or []
stockages = ctx.get("storages") or []
ponts = ctx.get("bridges") or []
def entry_label(e):
return f"{e['distro']} {e['version']} [{e['arch']}] {e['name']}"
class ProxmoxForm(PlanMixin, App):
TITLE = t("Deploy one or more ERPLibre VMs on Proxmox VE!")
BINDINGS = [
("f5", "deploy", t("Deploy")),
("f4", "clear_vm", t("Reset VM")),
("f3", "preview", t("Preview")),
("f6", "select_all", t("All")),
("f8", "select_none", t("None")),
("escape", "cancel", t("Cancel")),
]
# Le socle porte la mise en page et les modales ; ne reste ici que ce
# qui nomme les widgets propres à Proxmox.
CSS = (
CSS_BASE
+ """
SelectionList { height: 10; border: solid $panel; }
RadioSet { height: auto; layout: horizontal; }
.vmrow Select.vmbranch { width: 34; }
#hostline { height: 1; color: $accent; padding: 0 1; }
"""
)
def __init__(self):
super().__init__()
self.arch = ctx.get("native") or arches[0]
self.profile = "1"
self.custom = {}
self.overrides = {}
self.locked = set()
self.copies = {}
self.rows = []
self.vms = []
self.result = None
# Génération des widgets de rangée : un message qui arrive d'un
# jeu périmé ne doit pas être pris pour une saisie.
self._gen = 0
self._shown_ids = ()
self._syncing = False
# ---------------------------------------------------------------- #
# L'écran
# ---------------------------------------------------------------- #
def compose(self) -> ComposeResult:
yield Header()
hote = ctx["host"].get("label") or ctx["host"]["target"]
yield Static(
f" {t('Proxmox host')} : {hote}"
f" {t('node')} : {ctx.get('node') or '?'}",
id="hostline",
)
with Horizontal(id="body"):
with VerticalScroll(id="fields"):
yield Static(t("Architecture"), classes="grouptitle")
with RadioSet(id="f_arch"):
for a in arches:
label = a if a != "all" else t("all archs")
yield RadioButton(label, value=a == self.arch)
yield Static(t("Catalog"), classes="grouptitle")
yield SelectionList(id="f_catalog")
yield Static(
t("Resources — applied to ALL VMs"),
classes="grouptitle",
)
with RadioSet(id="f_profile"):
for label in ("x1", "x2", "x3", "x4"):
yield RadioButton(label, value=label == "x1")
yield RadioButton(t("custom"))
# Les mêmes trois ressources qu'ailleurs, montées par la
# même fabrique : « libre… » révèle la saisie du dessous.
for champ, presets, etiquette in (
("vcpus", ctx["cpu_presets"], t("vCPU")),
("ram", ctx["ram_presets"], t("RAM: 2048 or 8G")),
("disk", ctx["disk_presets"], t("Disk")),
):
yield Select(
[
(
(
f"{v // 1024}G"
if champ == "ram"
else str(v)
),
v,
)
for v in presets
]
+ [(t("free value…"), FREE)],
prompt=etiquette,
id=RES_FIELDS[champ][0][1:],
disabled=True,
)
yield Input(
placeholder=etiquette,
id=RES_FIELDS[champ][1][1:],
classes="freeval",
disabled=True,
)
yield Static(t("Proxmox VE"), classes="grouptitle")
yield Select(
[(s, s) for s in stockages],
value=(
ctx.get("storage")
or (stockages[0] if stockages else SELECT_NULL)
),
prompt=t("Storage"),
allow_blank=not stockages,
id="f_storage",
)
yield Select(
[(b, b) for b in ponts],
value=(
ctx.get("bridge")
or (ponts[0] if ponts else SELECT_NULL)
),
prompt=t("Bridge"),
allow_blank=not ponts,
id="f_bridge",
)
yield Static(f" {t('First VMID')}")
yield Input(
value=str(ctx.get("next_vmid") or 100),
placeholder="100",
id="f_vmid",
)
yield Static(t("Access"), classes="grouptitle")
yield Static(f" {t('SSH public key')}")
yield Input(
value=ctx.get("ssh_key") or "",
placeholder="~/.ssh/id_ed25519.pub",
id="f_key",
)
yield Checkbox(
t("Start the VM after creating it"),
value=True,
id="f_start",
)
yield Checkbox(
t("Add an entry to ~/.ssh/config"),
value=True,
id="f_sshcfg",
)
yield Static("ERPLibre", classes="grouptitle")
yield Checkbox(
t("Install ERPLibre"), value=True, id="f_install"
)
yield Select(
[(lbl, i) for i, (lbl, _c) in enumerate(profiles)],
value=0 if profiles else SELECT_NULL,
allow_blank=not profiles,
id="f_profile_install",
)
yield Select(
[(b, b) for b in branches],
value=branches[0],
allow_blank=False,
id="f_branch",
)
yield Checkbox(
t("Follow the installation (dashboard)"),
value=True,
id="f_monitor",
)
yield Static(t("Parallelism"), classes="grouptitle")
yield Select(
[(str(n), n) for n in (1, 2, 3, 4)],
value=1,
allow_blank=False,
id="f_par",
)
with Vertical(id="right"):
yield VerticalScroll(id="plan")
yield Static("", id="totals")
with Horizontal(id="actions"):
yield Button(t("Deploy"), variant="primary", id="go")
yield Button(t("Text prompts"), id="prompts")
yield Button(t("Cancel"), id="no")
yield Footer()
def on_mount(self) -> None:
self._reload_catalog()
# ---------------------------------------------------------------- #
# Le plan
# ---------------------------------------------------------------- #
def _entries(self):
return catalog.get(self.arch) or []
def _selected_entries(self):
choisis = set(self.query_one("#f_catalog", SelectionList).selected)
return [e for e in self._entries() if entry_key(e) in choisis]
def _presets(self):
return {
"vcpus": ctx["cpu_presets"],
"ram": ctx["ram_presets"],
"disk": ctx["disk_presets"],
}
def _reload_catalog(self) -> None:
liste = self.query_one("#f_catalog", SelectionList)
garde = set(liste.selected)
liste.clear_options()
for e in self._entries():
cle = entry_key(e)
liste.add_option((entry_label(e), cle, cle in garde))
self._recompute()
self._mount_rows()
def _recompute(self) -> None:
entries = self._plan_entries()
self.vms = build_vms(
entries,
self.profile,
ctx["base_vcpus"],
ctx["host_cpu"],
self.custom,
self.overrides,
)
grow = (
ctx.get("extra_disk_gb", 0)
if self.query_one("#f_install", Checkbox).value
else 0
)
self.rows = plan_rows(
self.vms, noms_pris, grow, orphelin=lambda _n: False
)
for row, entry in zip(self.rows, entries):
cle = entry_key(entry)
row["custom"] = bool(self.overrides.get(cle))
row["locked"] = cle in self.locked
assign_vmids(
self.rows,
vmids_pris,
self._vmid_start(),
lambda vmid: (ctx.get("ipconfig") or (lambda _v: "ip=dhcp"))(
self._bridge(), vmid
),
)
self._render_plan()
def _vmid_start(self):
brut = self.query_one("#f_vmid", Input).value.strip()
return (
int(brut) if brut.isdigit() else (ctx.get("next_vmid") or 100)
)
def _bridge(self):
valeur = self.query_one("#f_bridge", Select).value
return "" if valeur is SELECT_NULL else valeur
def _storage(self):
valeur = self.query_one("#f_storage", Select).value
return "" if valeur is SELECT_NULL else valeur
def _row_head(self, index, row):
"""La ligne de titre du socle, plus ce que Proxmox ajoute : le
VMID et l'adresse. Les deux sont décidés ICI et pas par l'hôte —
les montrer avant de lancer est le seul moyen de les vérifier."""
base = PlanMixin._row_head(self, index, row)
vm = row["vm"]
if row["state"] == "exists":
return base
adresse = (vm.get("ipconfig") or "").replace("ip=", "")
return f"{base} VMID {vm.get('vmid', '?')} {adresse}"
def _mount_rows(self) -> None:
"""(Re)construit le panneau droit.
Le verrou couvre TOUT le montage : poser « value= » sur un Select
fait émettre un Changed à Textual, que on_select_changed prendrait
pour une saisie."""
self._syncing = True
self._gen += 1
plan = self.query_one("#plan", VerticalScroll)
plan.remove_children()
cartes = []
for i, r in enumerate(self.rows):
vm = r["vm"]
cle = self._row_key(i)
item = self._plan_entries()[i]
rangee = Horizontal(
Button("+", id=f"p{i}", classes="vmcopy"),
Button("✎", id=f"r{i}", classes="vmcopy"),
Button(
"🔒" if cle in self.locked else "🔓",
id=f"l{i}",
variant=(
"success" if cle in self.locked else "default"
),
classes="vmlock",
),
*res_row_widgets(
i,
vm,
self._presets(),
labels={"vcpus": t("vCPU")},
null=SELECT_NULL,
),
(
Button("−", id=f"m{i}", classes="vmcopy")
if item.get("instance")
else Static("", classes="vmcopy")
),
classes="vmrow",
)
cartes.append(
Vertical(
Static(self._row_head(i, r), id=f"h{i}"),
rangee,
classes=(
"vmcard locked" if cle in self.locked else "vmcard"
),
)
)
# Marque de génération sur CHAQUE widget : « walk_children() » ne
# voit rien avant le montage, les enfants attendent dans
# « _pending_children ».
def marquer(node):
node._el_gen = self._gen
for child in getattr(node, "_pending_children", None) or []:
marquer(child)
for carte in cartes:
marquer(carte)
plan.mount_all(cartes)
self._shown_ids = self._row_ids()
self.call_after_refresh(self._after_mount_rows)
def _after_mount_rows(self) -> None:
self._sync_free_inputs()
self._syncing = False
def _render_plan(self) -> None:
for i, r in enumerate(self.rows):
try:
self.query_one(f"#h{i}", Static).update(
self._row_head(i, r)
)
except Exception:
pass
n, cpu, ram, disque = plan_totals(self.rows)
libre = ctx.get("free_ram") or 0
alerte = (
f" ⚠ {t('more RAM than the host has free')}"
if libre and ram > libre
else ""
)
self.query_one("#totals", Static).update(
f" {n} {t('VM')} {cpu} vCPU {ram} Mo RAM "
f"{disque} Go [{res_label(self.profile)}]"
f" {t('storage')} {self._storage() or '?'}"
f" {t('bridge')} {self._bridge() or '?'}{alerte}"
)
def _refresh_after(self, remonter=False) -> None:
"""Recalcule, et ne remonte les rangées que si le JEU a changé :
un remontage à chaque frappe volerait le focus."""
self._recompute()
if remonter or self._row_ids() != self._shown_ids:
self._mount_rows()
else:
self._sync_free_inputs()
# ---------------------------------------------------------------- #
# Les messages
# ---------------------------------------------------------------- #
def on_selection_list_selected_changed(self, _event) -> None:
self._refresh_after()
def on_radio_set_changed(self, event) -> None:
if event.radio_set.id == "f_arch":
self.arch = arches[event.index]
self._reload_catalog()
return
if event.radio_set.id == "f_profile":
choix = ("1", "2", "3", "4", "custom")[event.index]
self.profile = choix
sur_mesure = choix == "custom"
for champ in RES_FIELDS:
self.query_one(RES_FIELDS[champ][0], Select).disabled = (
not sur_mesure
)
if not sur_mesure:
self._show_free(champ, False)
# Un réglage commun reprend la main sur les VM non figées :
# c'est le sens même du mot « commun ».
self._clear_overrides(tuple(RES_FIELDS))
self._refresh_after(remonter=True)
def on_checkbox_changed(self, event) -> None:
if event.checkbox.id == "f_install":
# Le disque d'ERPLibre entre — ou sort — du total.
self._refresh_after()
def on_input_changed(self, event) -> None:
if event.input.id == "f_vmid":
self._refresh_after()
def on_input_submitted(self, event) -> None:
ident = event.input.id or ""
if ident in {RES_FIELDS[c][1][1:] for c in RES_FIELDS}:
self._apply_free(ident.split("_", 1)[1])
self._refresh_after(remonter=True)
return
if ident.startswith("c") and "_" in ident:
rang, champ = ident[1:].split("_", 1)
if rang.isdigit():
self._set_override(
int(rang), champ, self._read_row_free(int(rang), champ)
)
self._refresh_after()
def on_select_changed(self, event) -> None:
if self._syncing or not self._is_current(event.select):
return
ident = event.select.id or ""
if ident in ("f_storage", "f_bridge"):
self._refresh_after()
return
# Réglage commun : « libre… » révèle la saisie, une valeur
# s'applique à toutes les VM non figées.
champ = SELECT_TO_FIELD.get(ident)
if champ:
if event.value is FREE:
self._show_free(champ, True)
self.query_one(RES_FIELDS[champ][1], Input).focus()
elif event.value is not SELECT_NULL:
self._show_free(champ, False)
self.custom[champ] = event.value
self._clear_overrides((champ,))
self._refresh_after(remonter=True)
return
# Réglage d'UNE rangée.
if ident.startswith("v") and "_" in ident:
rang, champ = ident[1:].split("_", 1)
if not rang.isdigit() or champ not in RES_FIELDS:
return
index = int(rang)
if event.value is FREE:
self._row_free(index, champ, True)
self._set_override(
index, champ, self._read_row_free(index, champ)
)
elif event.value is not SELECT_NULL:
# L'écho du montage n'est pas une saisie : sans ce test,
# les trois champs de chaque VM se surchargeaient dès
# l'affichage et toutes les rangées portaient la marque ✎.
if self._row_echo(index, champ, event.value):
return
self._row_free(index, champ, False)
self._set_override(index, champ, event.value)
self._refresh_after()
def on_button_pressed(self, event) -> None:
ident = event.button.id or ""
if ident == "go":
self.action_deploy()
elif ident == "no":
self.action_cancel()
elif ident == "prompts":
# Retour aux invites textuelles : {} n'est pas None, et
# l'appelant sait faire la différence entre « annulé » et
# « pose-moi les questions à l'ancienne ».
self.result = {}
self.exit()
elif ident.startswith("p") and ident[1:].isdigit():
self._add_copy(int(ident[1:]), 1)
elif ident.startswith("m") and ident[1:].isdigit():
self._add_copy(int(ident[1:]), -1)
elif ident.startswith("r") and ident[1:].isdigit():
self._rename(int(ident[1:]))
elif ident.startswith("l") and ident[1:].isdigit():
index = int(ident[1:])
self._set_lock(index, self._row_key(index) not in self.locked)
# ---------------------------------------------------------------- #
# Les actions
# ---------------------------------------------------------------- #
def _install(self):
if not self.query_one("#f_install", Checkbox).value:
return None
index = self.query_one("#f_profile_install", Select).value
label, cmd = (
profiles[index]
if profiles and isinstance(index, int)
else ("", "")
)
return {
"branch": self.query_one("#f_branch", Select).value,
"label": label,
"cmd": cmd,
}
def _form_values(self):
cle = self.query_one("#f_key", Input).value.strip()
return {
"host": ctx["host"],
"storage": self._storage(),
"bridge": self._bridge(),
"res_label": res_label(self.profile),
"ssh_key": os.path.expanduser(cle) if cle else "",
"start": self.query_one("#f_start", Checkbox).value,
"add_ssh_config": self.query_one("#f_sshcfg", Checkbox).value,
"install": self._install(),
# Le suivi est demandé au NIVEAU DU DÉPLOIEMENT : une VM sans
# ERPLibre se suit aussi (cloud-init, puis relevé système).
"monitor": self.query_one("#f_monitor", Checkbox).value,
"parallelism": self.query_one("#f_par", Select).value,
}
def action_deploy(self) -> None:
spec = build_spec(self.vms, noms_pris, self._form_values())
if not spec["vms"]:
self.notify(t("Nothing to deploy."), severity="warning")
return
if not spec["storage"]:
self.notify(
t("No storage able to hold a VM disk."), severity="error"
)
return
if not spec["bridge"]:
self.notify(t("No bridge on the host."), severity="error")
return
self.result = spec
self.exit()
def action_preview(self) -> None:
build = ctx.get("build_command")
if not build:
return
spec = build_spec(self.vms, noms_pris, self._form_values())
lignes = ["\n".join(build(vm, spec)) for vm in spec["vms"]] or [
t("Nothing selected.")
]
self.push_screen(preview_screen()(lignes))
def action_clear_vm(self) -> None:
"""Rend au réglage commun la VM sous le curseur (et son verrou)."""
index = self._focused_row()
if index is None:
return
cle = self._row_key(index)
self.locked.discard(cle)
self.overrides.pop(cle, None)
self._refresh_after(remonter=True)
def action_select_all(self) -> None:
self.query_one("#f_catalog", SelectionList).select_all()
def action_select_none(self) -> None:
self.query_one("#f_catalog", SelectionList).deselect_all()
def action_cancel(self) -> None:
self.result = None
self.exit()
app = ProxmoxForm()
if not run_app:
return app
app.run()
return app.result

View file

@ -3148,8 +3148,11 @@ TRANSLATIONS = {
"en": "Watch the VMs start (no install)",
},
"Proxmox VE - Deploy a VM on a remote host": {
"fr": "Proxmox VE - Déployer une VM sur un hôte distant",
"en": "Proxmox VE - Deploy a VM on a remote host",
# 🗄 et non 💻 : l'entrée d'à côté déploie sur CETTE machine, celle-ci
# sur une baie ailleurs. L'icône est ce qui distingue les deux d'un
# coup d'œil dans le menu.
"fr": "🗄 Proxmox VE - Déployer une VM sur un hôte distant",
"en": "🗄 Proxmox VE - Deploy a VM on a remote host",
},
"Deploy a virtual machine on Proxmox VE!": {
"fr": "Déployer une machine virtuelle sur Proxmox VE !",
@ -3299,6 +3302,95 @@ TRANSLATIONS = {
"fr": "Aucun pont réseau sur cet hôte.",
"en": "No network bridge on this host.",
},
# --- Écran de déploiement Proxmox VE (formulaire TUI) ---
"Deploy one or more ERPLibre VMs on Proxmox VE!": {
"fr": "Déployer une ou plusieurs VM ERPLibre sur Proxmox VE !",
"en": "Deploy one or more ERPLibre VMs on Proxmox VE!",
},
"Proxmox host": {
"fr": "Hôte Proxmox",
"en": "Proxmox host",
},
"node": {
"fr": "nœud",
"en": "node",
},
"Proxmox VE": {
"fr": "Proxmox VE",
"en": "Proxmox VE",
},
"Bridge": {
"fr": "Pont",
"en": "Bridge",
},
"First VMID": {
"fr": "Premier VMID",
"en": "First VMID",
},
"Access": {
"fr": "Accès",
"en": "Access",
},
"SSH public key": {
"fr": "Clé publique SSH",
"en": "SSH public key",
},
"Start the VM after creating it": {
"fr": "Démarrer la VM après sa création",
"en": "Start the VM after creating it",
},
"Add an entry to ~/.ssh/config": {
"fr": "Ajouter une entrée à ~/.ssh/config",
"en": "Add an entry to ~/.ssh/config",
},
"Follow the installation (dashboard)": {
"fr": "Suivre l'installation (tableau de bord)",
"en": "Follow the installation (dashboard)",
},
"Text prompts": {
"fr": "Questions texte",
"en": "Text prompts",
},
"Nothing to deploy.": {
"fr": "Rien à déployer.",
"en": "Nothing to deploy.",
},
"No bridge on the host.": {
"fr": "Aucun pont sur l'hôte.",
"en": "No bridge on the host.",
},
"more RAM than the host has free": {
"fr": "plus de RAM que l'hôte n'en a de libre",
"en": "more RAM than the host has free",
},
"Loading (host, storage, bridges, VMs)...": {
"fr": "Chargement (hôte, stockage, ponts, VM)…",
"en": "Loading (host, storage, bridges, VMs)...",
},
"TUI unavailable": {
"fr": "Interface graphique indisponible",
"en": "TUI unavailable",
},
"VM": {
"fr": "VM",
"en": "VM",
},
"VMID": {
"fr": "VMID",
"en": "VMID",
},
"address": {
"fr": "adresse",
"en": "address",
},
"storage": {
"fr": "stockage",
"en": "storage",
},
"other…": {
"fr": "autre…",
"en": "other…",
},
"offered": {
"fr": "proposés",
"en": "offered",

300
test/test_proxmox_form.py Normal file
View file

@ -0,0 +1,300 @@
#!/usr/bin/env python3
# © 2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
"""L'écran de déploiement Proxmox VE.
Ce que Proxmox a en plus de QEMU/KVM, et qui doit donc être éprouvé ici : le
VMID choisi AVANT le téléchargement de l'image (l'hôte ne dit « déjà pris »
qu'après), l'adresse qui s'en déduit sur un pont interne, le stockage et le
pont lus sur l'hôte, et une VM qui existe déjà et qu'on ne doit surtout pas
écraser.
Le rendu est vérifié sans terminal (`run_test`), sur un contexte synthétique :
aucun hôte Proxmox n'est joint.
"""
import asyncio
import sys
import unittest
sys.argv = ["todo.py"]
from script.todo.proxmox_deploy_form import ( # noqa: E402
assign_vmids,
build_spec,
res_label,
run_proxmox_form,
)
try:
import textual # noqa: F401
TEXTUAL = True
except Exception: # pragma: no cover - dépend de l'environnement
TEXTUAL = False
def rangee(nom, etat="new"):
return {
"vm": {
"name": nom,
"distro": "debian",
"version": "13",
"arch": "amd64",
"vcpus": 2,
"ram": 2048,
"disk": "32G",
},
"state": etat,
"note": "",
"disk_gb": 32,
}
class TestVmid(unittest.TestCase):
"""Proxmox refuse un VMID déjà pris, et il le dit APRÈS avoir téléchargé
l'image : le choix se fait donc avant, d'après ce que l'hôte déclare."""
def test_taken_ids_are_skipped(self):
rows = [rangee("a"), rangee("b")]
assign_vmids(rows, [100, 101, 103], 100, lambda v: "ip=dhcp")
self.assertEqual([r["vm"]["vmid"] for r in rows], [102, 104])
def test_an_existing_vm_keeps_its_own(self):
rows = [rangee("a", "exists"), rangee("b")]
assign_vmids(rows, [], 100, lambda v: "ip=dhcp")
self.assertNotIn("vmid", rows[0]["vm"])
self.assertEqual(rows[1]["vm"]["vmid"], 100)
def test_the_first_vmid_is_honoured(self):
rows = [rangee("a")]
assign_vmids(rows, [], 250, lambda v: "ip=dhcp")
self.assertEqual(rows[0]["vm"]["vmid"], 250)
def test_a_vmid_never_goes_below_100(self):
# Proxmox réserve les VMID sous 100.
rows = [rangee("a")]
assign_vmids(rows, [], 7, lambda v: "ip=dhcp")
self.assertEqual(rows[0]["vm"]["vmid"], 100)
def test_the_address_is_derived_from_the_vmid(self):
rows = [rangee("a"), rangee("b")]
assign_vmids(
rows, [], 100, lambda v: f"ip=10.10.10.{50 + v % 200}/24"
)
self.assertEqual(rows[0]["vm"]["ipconfig"], "ip=10.10.10.150/24")
self.assertEqual(rows[1]["vm"]["ipconfig"], "ip=10.10.10.151/24")
def test_without_a_bridge_rule_it_falls_back_to_dhcp(self):
rows = [rangee("a")]
assign_vmids(rows, [], 100, None)
self.assertEqual(rows[0]["vm"]["ipconfig"], "ip=dhcp")
class TestSpec(unittest.TestCase):
def _form(self, **extra):
base = {
"host": {"target": "erplibre@10.0.0.5"},
"storage": "local-lvm",
"bridge": "vmbr0",
"res_label": "x1",
"ssh_key": "/home/x/.ssh/id_ed25519.pub",
"start": True,
"add_ssh_config": True,
"install": {"branch": "develop", "label": "Odoo 18", "cmd": "make"},
"monitor": True,
"parallelism": 2,
}
base.update(extra)
return base
def test_an_existing_vm_is_never_recreated(self):
vms = [{"name": "a"}, {"name": "b"}]
spec = build_spec(vms, ["b"], self._form())
self.assertEqual([v["name"] for v in spec["vms"]], ["a"])
self.assertEqual(spec["existing"], ["b"])
def test_the_user_defaults_to_erplibre(self):
spec = build_spec([], [], self._form())
self.assertEqual(spec["user"], "erplibre")
def test_the_monitor_choice_reaches_the_spec(self):
# Le suivi est demandé au NIVEAU DU DÉPLOIEMENT : une VM sans
# ERPLibre se suit aussi.
spec = build_spec([], [], self._form(install=None, monitor=True))
self.assertIsNone(spec["install"])
self.assertTrue(spec["monitor"])
def test_the_resource_label_names_the_common_setting(self):
self.assertEqual(res_label("3"), "x3")
self.assertNotEqual(res_label("custom"), "xcustom")
def contexte():
def entree(distro, version, arch="amd64"):
return {
"name": f"erplibre-{distro}-{version}",
"distro": distro,
"version": version,
"arch": arch,
"ram": 2048,
"disk": "32G",
}
return {
"host": {"target": "erplibre@10.0.0.5", "sudo": "sudo ", "label": "pve"},
"node": "pve1",
"catalog": {
"amd64": [
entree("ubuntu", "26.04"),
entree("debian", "13"),
entree("fedora", "44"),
],
"arm64": [entree("debian", "13", "arm64")],
},
"arches": ["amd64", "arm64"],
"native": "amd64",
"names": ["erplibre-debian-13"],
"vmids": [100, 101],
"next_vmid": 102,
"storages": ["local-lvm", "local"],
"storage": "local-lvm",
"bridges": ["vmbr0"],
"bridge": "vmbr0",
"ipconfig": lambda pont, vmid: f"ip=10.10.10.{50 + vmid % 200}/24",
"build_command": lambda vm, spec: [f"qm create {vm['vmid']}"],
"branches": ["develop", "master"],
"install_profiles": [("ERPLibre + Odoo 18", "make install_odoo_18")],
"ssh_key": "/home/x/.ssh/id_ed25519.pub",
"cpu_presets": [2, 4, 8],
"ram_presets": [2048, 4096, 8192],
"disk_presets": ["32G", "64G"],
"base_vcpus": 2,
"host_cpu": 8,
"free_ram": 12000,
"extra_disk_gb": 10,
}
@unittest.skipUnless(TEXTUAL, "Textual absent")
class TestEcran(unittest.TestCase):
"""Le formulaire, monté sans terminal."""
def _rendu(self, gestes):
ctx = contexte()
resultat = {}
async def scenario():
from textual.widgets import SelectionList
app = run_proxmox_form(ctx, run_app=False)
async with app.run_test(size=(200, 50)) as pilote:
await pilote.pause()
liste = app.query_one(SelectionList)
for i in range(3):
liste.select(liste.get_option_at_index(i).value)
await pilote.pause()
await pilote.pause()
await gestes(app, pilote)
resultat["app"] = app
asyncio.run(scenario())
return resultat["app"]
def test_the_plan_shows_a_row_per_selected_system(self):
async def rien(app, pilote):
pass
app = self._rendu(rien)
self.assertEqual(len(app.rows), 3)
def test_the_head_line_carries_the_vmid_and_the_address(self):
async def rien(app, pilote):
pass
app = self._rendu(rien)
tete = app._row_head(0, app.rows[0])
self.assertIn("VMID", tete)
self.assertIn("10.10.10.", tete)
def test_an_existing_vm_is_marked_and_gets_no_vmid(self):
async def rien(app, pilote):
pass
app = self._rendu(rien)
deja = [r for r in app.rows if r["state"] == "exists"]
self.assertEqual(len(deja), 1)
self.assertNotIn("VMID", app._row_head(1, deja[0]))
def test_mounting_does_not_mark_every_row_as_custom(self):
# Poser « value= » sur un Select fait émettre un Changed : pris pour
# une saisie, il surchargeait les trois champs de CHAQUE VM et toutes
# les rangées portaient la marque ✎ avant qu'on ne touche à rien.
async def rien(app, pilote):
pass
app = self._rendu(rien)
self.assertEqual(app.overrides, {})
self.assertNotIn("✎", app._row_head(0, app.rows[0]))
def test_a_lock_survives_a_common_setting(self):
async def gestes(app, pilote):
app._set_lock(0, True)
await pilote.pause()
app.custom["ram"] = 8192
app.profile = "custom"
app._clear_overrides(("ram",))
app._recompute()
await pilote.pause()
app = self._rendu(gestes)
self.assertEqual(app.rows[0]["vm"]["ram"], 2048)
def test_a_copy_adds_a_vm_with_its_own_vmid(self):
async def gestes(app, pilote):
app._add_copy(0, 1)
await pilote.pause()
app = self._rendu(gestes)
self.assertEqual(len(app.rows), 4)
vmids = [
r["vm"]["vmid"] for r in app.rows if r["state"] != "exists"
]
self.assertEqual(len(vmids), len(set(vmids)))
def test_deploying_yields_a_spec_the_engine_can_run(self):
async def gestes(app, pilote):
app.action_deploy()
app = self._rendu(gestes)
spec = app.result
self.assertEqual(len(spec["vms"]), 2)
self.assertEqual(spec["existing"], ["erplibre-debian-13"])
self.assertEqual(spec["storage"], "local-lvm")
self.assertEqual(spec["bridge"], "vmbr0")
for vm in spec["vms"]:
self.assertIn("vmid", vm)
self.assertIn("ip=", vm["ipconfig"])
self.assertEqual(spec["install"]["branch"], "develop")
def test_text_prompts_are_not_a_cancellation(self):
# {} n'est pas None : l'appelant distingue « annulé » de
# « pose-moi les questions à l'ancienne ».
async def gestes(app, pilote):
from textual.widgets import Button
app.on_button_pressed(
type("E", (), {"button": Button("x", id="prompts")})()
)
app = self._rendu(gestes)
self.assertEqual(app.result, {})
def test_cancelling_yields_nothing(self):
async def gestes(app, pilote):
app.action_cancel()
self.assertIsNone(self._rendu(gestes).result)
if __name__ == "__main__":
unittest.main(verbosity=2)

View file

@ -0,0 +1,288 @@
#!/usr/bin/env python3
# © 2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
"""Le socle commun aux deux formulaires de déploiement.
Le formulaire QEMU/KVM et le formulaire Proxmox posent le même travail :
choisir des systèmes, régler des ressources, vérifier le plan, lancer. Ce qui
est commun vit dans `deploy_form_lib` (logique pure, socle CSS, fabrique des
ressources) et `deploy_form_plan` (surcharges, verrous, exemplaires,
renommage). Ces tests gardent DEUX propriétés :
* le socle fait ce qu'il dit — une ressource libre, un verrou, un écho de
montage pris pour une saisie ;
* les deux formulaires s'en servent VRAIMENT, au lieu de le redire chacun de
son côté. C'est la seule chose qui empêche l'architecture de retomber en
deux copies qui divergent.
"""
import ast
import pathlib
import sys
import unittest
sys.argv = ["todo.py"]
from script.todo import deploy_form_lib as lib # noqa: E402
from script.todo import deploy_form_plan as plan # noqa: E402
RACINE = pathlib.Path(__file__).resolve().parent.parent
TODO_DIR = RACINE / "script" / "todo"
class TestRessourcesLibres(unittest.TestCase):
"""« autre… » ne vaut pas zéro : il révèle une saisie."""
def test_the_free_sentinel_closes_the_list(self):
choix = lib.res_choices([2, 4, 8])
self.assertEqual(choix[-1][1], lib.FREE)
self.assertEqual([v for _l, v in choix[:-1]], ["2", "4", "8"])
def test_a_formatter_shapes_the_labels(self):
choix = lib.res_choices([2048, 4096], fmt=lambda v: f"{v // 1024}G")
self.assertEqual([lbl for lbl, _v in choix[:-1]], ["2G", "4G"])
def test_free_without_typing_keeps_the_default(self):
# Le piège payé : valider un formulaire à peine ouvert rétrécissait
# les machines à zéro.
self.assertEqual(lib.res_value(lib.FREE, "", 4096), 4096)
self.assertEqual(lib.res_value(None, " ", 2), 2)
def test_a_typed_value_wins(self):
self.assertEqual(lib.res_value(lib.FREE, "12", 2), 12)
self.assertEqual(lib.res_value("8", "", 2), 8)
class TestEtatDuNom(unittest.TestCase):
"""Un nom déjà pris, un disque resté seul : deux gravités différentes."""
def test_a_defined_vm_is_skipped(self):
etat, note = lib.vm_status("erplibre-debian-13", ["erplibre-debian-13"])
self.assertEqual(etat, "exists")
self.assertTrue(note)
def test_the_orphan_probe_is_replaceable(self):
# Proxmox distant : ses disques vivent dans un stockage que seul
# l'hôte connaît, jamais dans /var/lib/libvirt.
etat, _n = lib.vm_status("x", [], orphelin=lambda _n: False)
self.assertEqual(etat, "new")
etat, _n = lib.vm_status("x", [], orphelin=lambda _n: True)
self.assertEqual(etat, "orphan")
def test_plan_rows_carries_the_probe(self):
vms = [{"name": "x", "vcpus": 2, "ram": 2048, "disk": "32G"}]
rows = lib.plan_rows(vms, [], 0, orphelin=lambda _n: True)
self.assertEqual(rows[0]["state"], "orphan")
class FauxPlan(plan.PlanMixin):
"""Le strict contrat du mixin, sans Textual : de quoi éprouver ce qui ne
touche pas à l'écran."""
def __init__(self, entrees):
self.entrees = entrees
self.copies = {}
self.overrides = {}
self.locked = set()
self.custom = {}
self.rows = []
self._gen = 0
self._shown_ids = ()
def _selected_entries(self):
return self.entrees
def entree(distro, version, arch="amd64"):
return {
"name": f"erplibre-{distro}-{version}",
"distro": distro,
"version": version,
"arch": arch,
"ram": 2048,
"disk": "32G",
}
class TestSocleDuPlan(unittest.TestCase):
def setUp(self):
self.app = FauxPlan([entree("debian", "13"), entree("ubuntu", "26.04")])
self.app.rows = [
{
"vm": {
"name": e["name"],
"distro": e["distro"],
"version": e["version"],
"arch": e["arch"],
"vcpus": 2,
"ram": 2048,
"disk": "32G",
},
"state": "new",
"note": "",
"disk_gb": 32,
}
for e in self.app.entrees
]
def test_an_override_is_written_then_dropped(self):
self.app._set_override(0, "ram", 8192)
cle = self.app._row_key(0)
self.assertEqual(self.app.overrides[cle]["ram"], 8192)
# Une saisie vidée RETIRE la surcharge : écrire zéro donnerait une VM
# à 0 vCPU.
self.app._set_override(0, "ram", 0)
self.assertNotIn(cle, self.app.overrides)
def test_a_common_setting_reclaims_the_unlocked_rows(self):
self.app._set_override(0, "ram", 8192)
self.app._set_override(1, "ram", 8192)
self.app.locked.add(self.app._row_key(1))
self.app._clear_overrides(("ram",))
self.assertNotIn(self.app._row_key(0), self.app.overrides)
self.assertIn(self.app._row_key(1), self.app.overrides)
def test_clearing_is_per_field(self):
self.app._set_override(0, "ram", 8192)
self.app._set_override(0, "disk", "64G")
self.app._clear_overrides(("ram",))
self.assertEqual(
self.app.overrides[self.app._row_key(0)], {"disk": "64G"}
)
def test_a_command_drags_its_label_along(self):
cle = self.app._row_key(0)
self.app.overrides[cle] = {
"install_cmd": "make install_odoo_18",
"install_label": "Odoo 18",
}
self.app._clear_overrides(("install_cmd",))
self.assertNotIn(cle, self.app.overrides)
def test_the_mount_echo_is_not_input(self):
# La valeur que le modèle porte déjà : c'est l'écho du montage.
self.assertTrue(self.app._row_echo(0, "ram", 2048))
self.assertFalse(self.app._row_echo(0, "ram", 4096))
# Rang hors plan : rien à appliquer.
self.assertTrue(self.app._row_echo(99, "ram", 1))
def test_copies_expand_the_selection(self):
self.app.copies[("debian", "13", "amd64")] = 1
entrees = self.app._plan_entries()
self.assertEqual(len(entrees), 3)
self.assertEqual(len(set(lib.entry_key(e) for e in entrees)), 3)
def test_the_head_line_names_what_is_special(self):
self.app.rows[0]["locked"] = True
self.assertIn("🔒", self.app._row_head(0, self.app.rows[0]))
self.app.rows[1]["custom"] = True
self.assertIn("✎", self.app._row_head(1, self.app.rows[1]))
self.assertIn("debian", self.app._row_head(0, self.app.rows[0]))
def test_locking_copies_the_resources(self):
gele = self.app._lock_fields(0)
self.assertEqual(gele["ram"], 2048)
self.assertEqual(gele["disk"], "32G")
self.assertEqual(gele["vcpus"], 2)
def membres(chemin, classe=None):
"""Noms des fonctions/méthodes définies dans un fichier (ou une classe)."""
arbre = ast.parse(pathlib.Path(chemin).read_text(encoding="utf-8"))
trouves = set()
for n in ast.walk(arbre):
if isinstance(n, ast.ClassDef) and (classe is None or n.name == classe):
trouves |= {
m.name for m in n.body if isinstance(m, ast.FunctionDef)
}
if classe is None and isinstance(n, ast.FunctionDef):
trouves.add(n.name)
return trouves
class TestPasDeDoublon(unittest.TestCase):
"""L'architecture demandée : un seul endroit par idée.
Ces tests échouent le jour où quelqu'un recopie un geste du plan dans un
formulaire plutôt que de l'appeler. C'est exactement ce qu'ils gardent.
"""
PARTAGES = {
"_plan_entries",
"_row_ids",
"_row_key",
"_is_current",
"_focused_row",
"_set_override",
"_clear_overrides",
"_set_lock",
"_add_copy",
"_rename",
"_row_free",
"_read_row_free",
"_show_free",
"_apply_free",
"_sync_free_inputs",
"_row_echo",
}
def test_the_mixin_owns_the_plan_gestures(self):
self.assertTrue(
self.PARTAGES <= membres(TODO_DIR / "deploy_form_plan.py")
)
def test_neither_form_redefines_them(self):
for fichier in ("qemu_deploy_form.py", "proxmox_deploy_form.py"):
redits = self.PARTAGES & membres(TODO_DIR / fichier)
self.assertEqual(
redits,
set(),
f"{fichier} redit ce que le socle porte déjà : {redits}",
)
def test_both_forms_inherit_the_mixin(self):
for fichier in ("qemu_deploy_form.py", "proxmox_deploy_form.py"):
src = pathlib.Path(TODO_DIR / fichier).read_text(encoding="utf-8")
self.assertIn("PlanMixin, App", src, fichier)
def test_the_pure_logic_lives_in_one_place(self):
pures = {
"parse_ram",
"parse_disk",
"positive_int",
"plan_rows",
"plan_totals",
"build_vms",
"apply_profile",
"apply_overrides",
"expand_copies",
"entry_key",
"vm_name",
"clean_hostname",
"clip_payload",
"run_deploy_progress",
}
self.assertTrue(pures <= membres(TODO_DIR / "deploy_form_lib.py"))
# Réexportées par l'ancien module : les appelants historiques (et les
# tests déjà écrits) importent encore ces noms LÀ.
from script.todo import qemu_deploy_form as ancien
for nom in pures:
self.assertTrue(hasattr(ancien, nom), nom)
def test_the_row_factory_is_shared(self):
for fichier in ("qemu_deploy_form.py", "proxmox_deploy_form.py"):
src = pathlib.Path(TODO_DIR / fichier).read_text(encoding="utf-8")
self.assertIn("res_row_widgets(", src, fichier)
def test_the_css_base_is_shared(self):
self.assertIn("#plan", lib.CSS_BASE)
self.assertIn("PreviewScreen", lib.CSS_BASE)
for fichier in ("qemu_deploy_form.py", "proxmox_deploy_form.py"):
src = pathlib.Path(TODO_DIR / fichier).read_text(encoding="utf-8")
self.assertIn("CSS_BASE", src, fichier)
# Les règles communes ne doivent PAS être recopiées à côté.
self.assertNotIn("#totals { height: auto;", src, fichier)
if __name__ == "__main__":
unittest.main(verbosity=2)