Compare commits
10 commits
89dc8e001f
...
dc42bb715a
| Author | SHA1 | Date | |
|---|---|---|---|
| dc42bb715a | |||
| 6f087d2a9b | |||
| 2d549025ec | |||
| b91878f06e | |||
| 160103a595 | |||
| 18c40790c8 | |||
| c47302d5c4 | |||
| af523fc1c2 | |||
| 970a373400 | |||
| 612c1e455a |
5 changed files with 800 additions and 46 deletions
|
|
@ -28,6 +28,10 @@ docker_show_databases:
|
|||
docker_show_logs_live:
|
||||
docker compose logs -f
|
||||
|
||||
.PHONY: docker_show_absolute_path
|
||||
docker_show_absolute_path:
|
||||
docker inspect $(docker ps -q) --format '{{.Name}}{{range .Mounts}} | {{.Source}} → {{.Destination}}{{end}}'
|
||||
|
||||
.PHONY: docker_show_process
|
||||
docker_show_process:
|
||||
docker compose ps
|
||||
|
|
|
|||
639
script/qemu/deploy_qemu.py
Executable file
639
script/qemu/deploy_qemu.py
Executable file
|
|
@ -0,0 +1,639 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Déploiement rapide de VM Ubuntu (cloud-image) avec qemu-img + cloud-init + virt-install.
|
||||
|
||||
Reprend le workflow des notes :
|
||||
1. Télécharge l'image cloud Ubuntu (si absente du cache -> pas de double téléchargement).
|
||||
2. Convertit/copie l'image en un qcow2 de travail dédié à la VM.
|
||||
3. Redimensionne le disque virtuel.
|
||||
4. Génère user-data / meta-data et construit le seed.iso (cidata).
|
||||
5. Lance virt-install en important le disque + le seed en CD-ROM.
|
||||
|
||||
Exemples
|
||||
--------
|
||||
# Déploiement minimal (clé SSH, aucun mot de passe)
|
||||
sudo ./qemu-deploy.py /var/lib/libvirt/images/iso/noble.img \\
|
||||
--name test-vm --ssh-key ~/.ssh/id_ed25519.pub
|
||||
|
||||
# Reproduction fidèle des notes (8 Go RAM, 8 vCPU, mot de passe demandé)
|
||||
sudo ./qemu-deploy.py /var/lib/libvirt/images/iso/noble.img \\
|
||||
--name test-vm --memory 8192 --vcpus 8 --disk-size 120G --ask-password
|
||||
|
||||
# Voir ce qui serait fait, sans rien exécuter
|
||||
./qemu-deploy.py /var/lib/libvirt/images/iso/noble.img --name test-vm --dry-run
|
||||
|
||||
sudo ./script/qemu/deploy_qemu.py /var/lib/libvirt/images/iso/noble.img --name test-vm --memory 8192 --vcpus 8 --disk-size 120G --ask-password --force
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import getpass
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import textwrap
|
||||
import time
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Table des versions Ubuntu -> (nom de code, valeur --osinfo/libosinfo)
|
||||
# Le nom de code sert à construire l'URL de l'image cloud. Les valeurs LTS sont
|
||||
# stables ; --codename et --osinfo permettent de surcharger si la table vieillit.
|
||||
# --------------------------------------------------------------------------- #
|
||||
UBUNTU_VERSIONS: dict[str, tuple[str, str]] = {
|
||||
"20.04": ("focal", "ubuntu20.04"),
|
||||
"22.04": ("jammy", "ubuntu22.04"),
|
||||
"24.04": ("noble", "ubuntu24.04"),
|
||||
"24.10": ("oracular", "ubuntu24.10"),
|
||||
"25.04": ("plucky", "ubuntu25.04"),
|
||||
"25.10": ("questing", "ubuntu25.10"),
|
||||
# "26.04": ("resolute", "ubuntu26.04"),
|
||||
}
|
||||
|
||||
CLOUD_IMG_BASE = "https://cloud-images.ubuntu.com"
|
||||
|
||||
|
||||
def image_url(codename: str, arch: str) -> str:
|
||||
"""URL de l'image cloud « current » pour un nom de code + architecture."""
|
||||
return f"{CLOUD_IMG_BASE}/{codename}/current/{codename}-server-cloudimg-{arch}.img"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Utilitaires d'exécution
|
||||
# --------------------------------------------------------------------------- #
|
||||
class Runner:
|
||||
"""Exécute (ou affiche, en dry-run) les commandes, avec sudo au besoin."""
|
||||
|
||||
def __init__(self, use_sudo: bool, dry_run: bool) -> None:
|
||||
self.use_sudo = use_sudo
|
||||
self.dry_run = dry_run
|
||||
|
||||
def run(
|
||||
self, cmd: list[str], *, privileged: bool = False, check: bool = True
|
||||
) -> None:
|
||||
if privileged and self.use_sudo:
|
||||
cmd = ["sudo", *cmd]
|
||||
printable = " ".join(cmd)
|
||||
if self.dry_run:
|
||||
print(f" [dry-run] {printable}")
|
||||
return
|
||||
print(f" $ {printable}")
|
||||
subprocess.run(cmd, check=check)
|
||||
|
||||
|
||||
def need_tool(name: str) -> None:
|
||||
if shutil.which(name) is None:
|
||||
sys.exit(f"Erreur : outil requis introuvable dans le PATH : {name!r}")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Étapes
|
||||
# --------------------------------------------------------------------------- #
|
||||
def download_image(url: str, dest: Path, dry_run: bool) -> None:
|
||||
"""Télécharge l'image seulement si elle n'existe pas déjà (cache)."""
|
||||
if dest.exists() and dest.stat().st_size > 0:
|
||||
size_mb = dest.stat().st_size / 1024 / 1024
|
||||
print(
|
||||
f" Image déjà présente ({size_mb:.0f} Mo), téléchargement ignoré : {dest}"
|
||||
)
|
||||
return
|
||||
if dry_run:
|
||||
print(f" [dry-run] téléchargement {url} -> {dest}")
|
||||
return
|
||||
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
print(f" Téléchargement {url}")
|
||||
tmp = dest.with_suffix(dest.suffix + ".part")
|
||||
|
||||
def _progress(block_num: int, block_size: int, total: int) -> None:
|
||||
if total > 0:
|
||||
pct = min(100, block_num * block_size * 100 // total)
|
||||
print(f"\r {pct:3d}%", end="", flush=True)
|
||||
|
||||
try:
|
||||
urllib.request.urlretrieve(
|
||||
url, tmp, _progress
|
||||
) # noqa: S310 (URL contrôlée)
|
||||
except Exception as exc: # pragma: no cover - dépend du réseau
|
||||
tmp.unlink(missing_ok=True)
|
||||
sys.exit(f"\nÉchec du téléchargement : {exc}")
|
||||
print()
|
||||
tmp.replace(dest)
|
||||
|
||||
|
||||
def verify_sha256(url: str, image: Path, dry_run: bool) -> None:
|
||||
"""Vérifie l'empreinte via le SHA256SUMS publié dans le même répertoire."""
|
||||
if dry_run:
|
||||
print(" [dry-run] vérification SHA256 ignorée")
|
||||
return
|
||||
sums_url = url.rsplit("/", 1)[0] + "/SHA256SUMS"
|
||||
filename = url.rsplit("/", 1)[1]
|
||||
print(f" Vérification SHA256 via {sums_url}")
|
||||
try:
|
||||
with urllib.request.urlopen(sums_url) as resp: # noqa: S310
|
||||
sums = resp.read().decode()
|
||||
except Exception as exc: # pragma: no cover
|
||||
sys.exit(f"Impossible de récupérer SHA256SUMS : {exc}")
|
||||
|
||||
expected = next(
|
||||
(
|
||||
line.split()[0]
|
||||
for line in sums.splitlines()
|
||||
if line.strip().endswith(filename)
|
||||
),
|
||||
None,
|
||||
)
|
||||
if expected is None:
|
||||
sys.exit(f"Empreinte introuvable pour {filename} dans SHA256SUMS")
|
||||
|
||||
h = hashlib.sha256()
|
||||
with image.open("rb") as fh:
|
||||
for chunk in iter(lambda: fh.read(1 << 20), b""):
|
||||
h.update(chunk)
|
||||
if h.hexdigest() != expected:
|
||||
sys.exit(
|
||||
f"SHA256 NON conforme !\n attendu : {expected}\n obtenu : {h.hexdigest()}"
|
||||
)
|
||||
print(" SHA256 conforme.")
|
||||
|
||||
|
||||
def hash_password(plain: str) -> str:
|
||||
"""Hash SHA-512 crypt ($6$...) via crypt (< 3.13) ou openssl en repli."""
|
||||
try:
|
||||
import crypt # supprimé en Python 3.13
|
||||
|
||||
return crypt.crypt(plain, crypt.mksalt(crypt.METHOD_SHA512))
|
||||
except (ImportError, AttributeError):
|
||||
need_tool("openssl")
|
||||
out = subprocess.run(
|
||||
["openssl", "passwd", "-6", "-stdin"],
|
||||
input=plain + "\n",
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
return out.stdout.strip()
|
||||
|
||||
|
||||
def build_cloud_config(
|
||||
args: argparse.Namespace, pw_hash: str | None, ssh_keys: list[str]
|
||||
) -> str:
|
||||
"""Construit le contenu #cloud-config (user-data)."""
|
||||
lines: list[str] = ["#cloud-config", f"hostname: {args.hostname}"]
|
||||
|
||||
user_block = [
|
||||
"users:",
|
||||
f" - name: {args.user}",
|
||||
" sudo: ALL=(ALL) NOPASSWD:ALL",
|
||||
" groups: users, admin",
|
||||
" shell: /bin/bash",
|
||||
" lock_passwd: false" if pw_hash else " lock_passwd: true",
|
||||
]
|
||||
if pw_hash:
|
||||
user_block.append(f' passwd: "{pw_hash}"')
|
||||
if ssh_keys:
|
||||
user_block.append(" ssh_authorized_keys:")
|
||||
user_block += [f" - {k}" for k in ssh_keys]
|
||||
lines += user_block
|
||||
|
||||
lines.append(f"ssh_pwauth: {'true' if pw_hash else 'false'}")
|
||||
lines.append(f"locale: {args.locale}")
|
||||
lines += [
|
||||
"keyboard:",
|
||||
f" layout: {args.keyboard_layout}",
|
||||
f" variant: {args.keyboard_variant}",
|
||||
]
|
||||
lines.append("package_update: true")
|
||||
lines.append(f"package_upgrade: {'false' if args.no_upgrade else 'true'}")
|
||||
|
||||
packages = ["qemu-guest-agent", *args.package]
|
||||
lines.append("packages:")
|
||||
lines += [
|
||||
f" - {p}" for p in dict.fromkeys(packages)
|
||||
] # dédoublonne, ordre gardé
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def build_seed(
|
||||
cloud_cfg: str, hostname: str, seed_dest: Path, runner: Runner
|
||||
) -> None:
|
||||
"""Génère le seed.iso (cidata) et le copie vers seed_dest."""
|
||||
meta_data = f"instance-id: {hostname}\nlocal-hostname: {hostname}\n"
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
tmp_path = Path(tmp)
|
||||
ud = tmp_path / "user-data"
|
||||
md = tmp_path / "meta-data"
|
||||
local_iso = tmp_path / "seed.iso"
|
||||
|
||||
if runner.dry_run:
|
||||
print(" [dry-run] user-data qui serait généré :")
|
||||
print(textwrap.indent(cloud_cfg, " "))
|
||||
else:
|
||||
ud.write_text(cloud_cfg)
|
||||
md.write_text(meta_data)
|
||||
|
||||
if runner.dry_run or shutil.which("cloud-localds"):
|
||||
runner.run(["cloud-localds", str(local_iso), str(ud), str(md)])
|
||||
else:
|
||||
need_tool("genisoimage")
|
||||
runner.run(
|
||||
[
|
||||
"genisoimage",
|
||||
"-output",
|
||||
str(local_iso),
|
||||
"-volid",
|
||||
"cidata",
|
||||
"-joliet",
|
||||
"-rock",
|
||||
str(ud),
|
||||
str(md),
|
||||
]
|
||||
)
|
||||
|
||||
(
|
||||
seed_dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
if not runner.dry_run
|
||||
else None
|
||||
)
|
||||
runner.run(["cp", str(local_iso), str(seed_dest)], privileged=True)
|
||||
|
||||
|
||||
def prepare_disk(
|
||||
image: Path, disk: Path, size: str, runner: Runner, force: bool
|
||||
) -> None:
|
||||
"""Convertit l'image cloud en qcow2 de travail puis redimensionne."""
|
||||
if disk.exists() and not force:
|
||||
sys.exit(
|
||||
f"Le disque {disk} existe déjà. Utilisez --force pour l'écraser."
|
||||
)
|
||||
runner.run(
|
||||
[
|
||||
"qemu-img",
|
||||
"convert",
|
||||
"-f",
|
||||
"qcow2",
|
||||
"-O",
|
||||
"qcow2",
|
||||
str(image),
|
||||
str(disk),
|
||||
],
|
||||
privileged=True,
|
||||
)
|
||||
runner.run(["qemu-img", "resize", str(disk), size], privileged=True)
|
||||
|
||||
|
||||
def network_name(network_arg: str) -> str | None:
|
||||
"""Extrait NAME de « network=NAME,... » ; None si c'est un bridge, etc."""
|
||||
for part in network_arg.split(","):
|
||||
if part.strip().startswith("network="):
|
||||
return part.split("=", 1)[1].strip()
|
||||
return None
|
||||
|
||||
|
||||
def ensure_network(name: str | None, runner: Runner) -> None:
|
||||
"""Active le réseau libvirt (idempotent : tolère « déjà actif »)."""
|
||||
if not name:
|
||||
return
|
||||
print(f" Activation du réseau libvirt '{name}' (si nécessaire)")
|
||||
runner.run(["virsh", "net-start", name], privileged=True, check=False)
|
||||
runner.run(["virsh", "net-autostart", name], privileged=True, check=False)
|
||||
|
||||
|
||||
def virt_install(
|
||||
args: argparse.Namespace,
|
||||
disk: Path,
|
||||
seed: Path,
|
||||
osinfo: str,
|
||||
runner: Runner,
|
||||
) -> None:
|
||||
cmd = [
|
||||
"virt-install",
|
||||
"--name",
|
||||
args.name,
|
||||
"--memory",
|
||||
str(args.memory),
|
||||
"--vcpus",
|
||||
str(args.vcpus),
|
||||
"--import",
|
||||
"--disk",
|
||||
f"path={disk},format=qcow2,bus=virtio",
|
||||
"--disk",
|
||||
f"path={seed},device=cdrom",
|
||||
"--osinfo",
|
||||
osinfo,
|
||||
"--network",
|
||||
args.network,
|
||||
"--graphics",
|
||||
args.graphics,
|
||||
"--console",
|
||||
"pty,target_type=serial",
|
||||
]
|
||||
if not args.attach_console:
|
||||
cmd.append("--noautoconsole")
|
||||
runner.run(cmd, privileged=True)
|
||||
|
||||
|
||||
def wait_for_ip(name: str, use_sudo: bool, timeout: int) -> str | None:
|
||||
"""Interroge les baux DHCP libvirt jusqu'à obtenir l'IPv4 de la VM."""
|
||||
base = (["sudo"] if use_sudo else []) + [
|
||||
"virsh",
|
||||
"domifaddr",
|
||||
name,
|
||||
"--source",
|
||||
"lease",
|
||||
]
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
res = subprocess.run(base, capture_output=True, text=True)
|
||||
m = re.search(r"(\d+\.\d+\.\d+\.\d+)", res.stdout)
|
||||
if m:
|
||||
return m.group(1)
|
||||
time.sleep(3)
|
||||
return None
|
||||
|
||||
|
||||
def ssh_command(user: str, ip: str, has_key: bool) -> str:
|
||||
"""Commande SSH adaptée : clé -> connexion simple ; mot de passe seul ->
|
||||
force le mot de passe pour éviter « Too many authentication failures »."""
|
||||
if has_key:
|
||||
return f"ssh {user}@{ip}"
|
||||
return (
|
||||
f"ssh -o IdentitiesOnly=yes -o PreferredAuthentications=password "
|
||||
f"{user}@{ip}"
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# CLI
|
||||
# --------------------------------------------------------------------------- #
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
versions_help = "\n".join(
|
||||
f" {v:<7} {code:<10} {image_url(code, 'amd64')}"
|
||||
for v, (code, _osinfo) in UBUNTU_VERSIONS.items()
|
||||
)
|
||||
p = argparse.ArgumentParser(
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
description=__doc__,
|
||||
epilog="Versions Ubuntu disponibles (--version) et URL de l'image cloud :\n"
|
||||
+ versions_help,
|
||||
)
|
||||
p.add_argument(
|
||||
"image_path",
|
||||
type=Path,
|
||||
help="Chemin de cache de l'image cloud (.img). "
|
||||
"Si le fichier existe, il n'est PAS re-téléchargé.",
|
||||
)
|
||||
|
||||
g_img = p.add_argument_group("Image")
|
||||
g_img.add_argument(
|
||||
"--version",
|
||||
default="24.04",
|
||||
choices=UBUNTU_VERSIONS,
|
||||
help="Version Ubuntu (défaut : 24.04). Voir la liste ci-dessous.",
|
||||
)
|
||||
g_img.add_argument(
|
||||
"--codename", help="Force le nom de code (surcharge --version)."
|
||||
)
|
||||
g_img.add_argument(
|
||||
"--arch",
|
||||
default="amd64",
|
||||
help="Architecture de l'image (défaut : amd64).",
|
||||
)
|
||||
g_img.add_argument(
|
||||
"--verify",
|
||||
action="store_true",
|
||||
help="Vérifie l'empreinte SHA256 après téléchargement (recommandé).",
|
||||
)
|
||||
|
||||
g_vm = p.add_argument_group("VM")
|
||||
g_vm.add_argument("--name", required=True, help="Nom de la VM (virsh).")
|
||||
g_vm.add_argument(
|
||||
"--hostname", help="Nom d'hôte interne (défaut : --name)."
|
||||
)
|
||||
g_vm.add_argument(
|
||||
"--memory", type=int, default=8192, help="RAM en Mo (défaut : 8192)."
|
||||
)
|
||||
g_vm.add_argument(
|
||||
"--vcpus", type=int, default=4, help="Nombre de vCPU (défaut : 4)."
|
||||
)
|
||||
g_vm.add_argument(
|
||||
"--disk-size",
|
||||
default="20G",
|
||||
help="Taille du disque virtuel, ex. 120G (défaut : 20G).",
|
||||
)
|
||||
g_vm.add_argument(
|
||||
"--disk-dir",
|
||||
type=Path,
|
||||
default=Path("/var/lib/libvirt/images"),
|
||||
help="Répertoire du qcow2 de travail (défaut : /var/lib/libvirt/images).",
|
||||
)
|
||||
g_vm.add_argument(
|
||||
"--seed-dir",
|
||||
type=Path,
|
||||
default=Path("/var/lib/libvirt/images/iso"),
|
||||
help="Répertoire du seed.iso (défaut : .../images/iso).",
|
||||
)
|
||||
g_vm.add_argument(
|
||||
"--network",
|
||||
default="network=default,model=virtio",
|
||||
help="Argument --network de virt-install.",
|
||||
)
|
||||
g_vm.add_argument(
|
||||
"--graphics",
|
||||
default="none",
|
||||
help="Argument --graphics (défaut : none).",
|
||||
)
|
||||
g_vm.add_argument(
|
||||
"--osinfo", help="Force la valeur --osinfo (sinon déduite)."
|
||||
)
|
||||
g_vm.add_argument(
|
||||
"--attach-console",
|
||||
action="store_true",
|
||||
help="Attache la console série (sinon --noautoconsole).",
|
||||
)
|
||||
|
||||
g_cloud = p.add_argument_group("cloud-init")
|
||||
g_cloud.add_argument(
|
||||
"--user",
|
||||
default="erplibre",
|
||||
help="Utilisateur créé (défaut : erplibre).",
|
||||
)
|
||||
g_cloud.add_argument(
|
||||
"--password", help="Mot de passe en clair (déconseillé : visible)."
|
||||
)
|
||||
g_cloud.add_argument(
|
||||
"--ask-password",
|
||||
action="store_true",
|
||||
help="Demande le mot de passe de façon interactive (sûr).",
|
||||
)
|
||||
g_cloud.add_argument(
|
||||
"--password-hash",
|
||||
help="Empreinte $6$... déjà calculée (openssl passwd -6).",
|
||||
)
|
||||
g_cloud.add_argument(
|
||||
"--ssh-key",
|
||||
action="append",
|
||||
default=[],
|
||||
metavar="FICHIER",
|
||||
help="Fichier de clé publique SSH à injecter (répétable).",
|
||||
)
|
||||
g_cloud.add_argument(
|
||||
"--locale",
|
||||
default="fr_CA.UTF-8",
|
||||
help="Locale (défaut : fr_CA.UTF-8).",
|
||||
)
|
||||
g_cloud.add_argument(
|
||||
"--keyboard-layout",
|
||||
default="ca",
|
||||
help="Disposition clavier (défaut : ca).",
|
||||
)
|
||||
g_cloud.add_argument(
|
||||
"--keyboard-variant",
|
||||
default="multix",
|
||||
help="Variante clavier (défaut : multix).",
|
||||
)
|
||||
g_cloud.add_argument(
|
||||
"--package",
|
||||
action="append",
|
||||
default=[],
|
||||
metavar="PKG",
|
||||
help="Paquet APT additionnel (répétable).",
|
||||
)
|
||||
g_cloud.add_argument(
|
||||
"--no-upgrade",
|
||||
action="store_true",
|
||||
help="N'exécute pas package_upgrade au premier boot.",
|
||||
)
|
||||
|
||||
g_run = p.add_argument_group("Exécution")
|
||||
g_run.add_argument(
|
||||
"--no-sudo",
|
||||
action="store_true",
|
||||
help="N'utilise pas sudo pour les étapes privilégiées.",
|
||||
)
|
||||
g_run.add_argument(
|
||||
"--force",
|
||||
action="store_true",
|
||||
help="Écrase le qcow2 de travail existant.",
|
||||
)
|
||||
g_run.add_argument(
|
||||
"--no-wait-ip",
|
||||
action="store_true",
|
||||
help="N'attend pas l'attribution de l'IP à la fin.",
|
||||
)
|
||||
g_run.add_argument(
|
||||
"--ip-timeout",
|
||||
type=int,
|
||||
default=90,
|
||||
metavar="SEC",
|
||||
help="Délai max d'attente de l'IP DHCP (défaut : 90 s).",
|
||||
)
|
||||
g_run.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Affiche les commandes et le user-data sans rien exécuter.",
|
||||
)
|
||||
return p
|
||||
|
||||
|
||||
def resolve_password(args: argparse.Namespace) -> str | None:
|
||||
if args.password_hash:
|
||||
return args.password_hash
|
||||
if args.ask_password:
|
||||
pw = getpass.getpass("Mot de passe pour l'utilisateur : ")
|
||||
if pw != getpass.getpass("Confirmer : "):
|
||||
sys.exit("Les mots de passe ne correspondent pas.")
|
||||
return hash_password(pw)
|
||||
if args.password:
|
||||
return hash_password(args.password)
|
||||
return None
|
||||
|
||||
|
||||
def load_ssh_keys(paths: list[str]) -> list[str]:
|
||||
keys: list[str] = []
|
||||
for path in paths:
|
||||
p = Path(path).expanduser()
|
||||
if not p.exists():
|
||||
sys.exit(f"Clé SSH introuvable : {p}")
|
||||
keys.append(p.read_text().strip())
|
||||
return keys
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = build_parser().parse_args()
|
||||
args.hostname = args.hostname or args.name
|
||||
|
||||
codename, default_osinfo = UBUNTU_VERSIONS[args.version]
|
||||
if args.codename:
|
||||
codename = args.codename
|
||||
osinfo = args.osinfo or default_osinfo
|
||||
url = image_url(codename, args.arch)
|
||||
|
||||
pw_hash = resolve_password(args)
|
||||
ssh_keys = load_ssh_keys(args.ssh_key)
|
||||
if not pw_hash and not ssh_keys:
|
||||
print(
|
||||
"ATTENTION : ni mot de passe ni clé SSH -> connexion impossible à la VM.\n"
|
||||
" Ajoutez --ssh-key, --ask-password ou --password-hash.\n",
|
||||
file=sys.stderr,
|
||||
)
|
||||
|
||||
runner = Runner(
|
||||
use_sudo=not args.no_sudo and os.geteuid() != 0, dry_run=args.dry_run
|
||||
)
|
||||
disk = args.disk_dir / f"{args.name}.qcow2"
|
||||
seed = args.seed_dir / f"{args.name}-seed.iso"
|
||||
|
||||
if not args.dry_run:
|
||||
for tool in ("qemu-img", "virt-install"):
|
||||
need_tool(tool)
|
||||
|
||||
print(f"\n== 1/5 Image cloud ({args.version} / {codename}) ==")
|
||||
download_image(url, args.image_path, args.dry_run)
|
||||
if args.verify:
|
||||
verify_sha256(url, args.image_path, args.dry_run)
|
||||
|
||||
print(f"\n== 2-3/5 Disque de travail {disk} ({args.disk_size}) ==")
|
||||
prepare_disk(args.image_path, disk, args.disk_size, runner, args.force)
|
||||
|
||||
print(f"\n== 4/5 Seed cloud-init {seed} ==")
|
||||
cloud_cfg = build_cloud_config(args, pw_hash, ssh_keys)
|
||||
build_seed(cloud_cfg, args.hostname, seed, runner)
|
||||
|
||||
print(f"\n== 5/5 virt-install (--osinfo {osinfo}) ==")
|
||||
ensure_network(network_name(args.network), runner)
|
||||
virt_install(args, disk, seed, osinfo, runner)
|
||||
|
||||
has_key = bool(ssh_keys)
|
||||
print("\nTerminé. Suivi :")
|
||||
print(" virsh list --all")
|
||||
print(f" virsh console {args.name} # Ctrl+] pour quitter")
|
||||
|
||||
if args.dry_run:
|
||||
print("\n Connexion SSH (IP attribuée au 1er boot) :")
|
||||
print(f" {ssh_command(args.user, '<IP>', has_key)}")
|
||||
return
|
||||
if args.no_wait_ip:
|
||||
print(
|
||||
f"\n Récupérer l'IP : virsh domifaddr {args.name} --source lease"
|
||||
)
|
||||
return
|
||||
|
||||
print(f"\n Attente de l'IP (bail DHCP, max {args.ip_timeout} s)…")
|
||||
ip = wait_for_ip(args.name, runner.use_sudo, args.ip_timeout)
|
||||
if ip:
|
||||
print(f" IP : {ip}")
|
||||
print(" Connexion SSH :")
|
||||
print(f" {ssh_command(args.user, ip, has_key)}")
|
||||
else:
|
||||
print(" IP non obtenue dans le délai (cloud-init encore en cours ?).")
|
||||
print(f" Réessayez : virsh domifaddr {args.name} --source lease")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Binary file not shown.
|
|
@ -70,8 +70,6 @@ MONTHS_FR = {
|
|||
"decembre": 12,
|
||||
}
|
||||
|
||||
defaut_timeout = 60
|
||||
|
||||
|
||||
class SeleniumLib(object):
|
||||
def __init__(self, config):
|
||||
|
|
@ -81,6 +79,7 @@ class SeleniumLib(object):
|
|||
self.kdbx = None
|
||||
self.video_recorder = None
|
||||
self.default_copy_download_dir_path = None
|
||||
self.defaut_timeout = 60
|
||||
|
||||
date_str = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
tmp_dir = tempfile.mkdtemp(
|
||||
|
|
@ -343,10 +342,6 @@ class SeleniumLib(object):
|
|||
# Ajout de l'enregistrement
|
||||
if self.config.record_mode:
|
||||
# Create recording
|
||||
new_path = os.path.normpath(
|
||||
os.path.join(os.path.dirname(__file__), "..", "..")
|
||||
)
|
||||
sys.path.append(new_path)
|
||||
from script.selenium.selenium_video import VideoRecorder
|
||||
|
||||
if not os.path.isdir(self.dirname_recording):
|
||||
|
|
@ -362,16 +357,23 @@ class SeleniumLib(object):
|
|||
if self.config.use_firefox_driver and not self.config.use_network:
|
||||
if not self.config.no_dark_mode:
|
||||
self.driver.install_addon(
|
||||
"./script/selenium/darkreader-firefox.xpi", temporary=True
|
||||
os.path.join(
|
||||
new_path, "script/selenium/darkreader-firefox.xpi"
|
||||
),
|
||||
temporary=True,
|
||||
)
|
||||
self.driver.install_addon(
|
||||
"./script/selenium/odoo_debug-4.0.xpi", temporary=True
|
||||
os.path.join(
|
||||
new_path, "script/selenium/odoo_debug-4.0.xpi"
|
||||
),
|
||||
temporary=True,
|
||||
)
|
||||
|
||||
if (
|
||||
not self.config.not_private_mode
|
||||
and not self.config.no_dark_mode
|
||||
):
|
||||
# self.driver.set_context("chrome")
|
||||
self.driver.get("about:addons")
|
||||
# Enable Dark Reader into incognito
|
||||
# Click Extensions
|
||||
|
|
@ -469,7 +471,9 @@ class SeleniumLib(object):
|
|||
word = ""
|
||||
return word
|
||||
|
||||
def click(self, xpath, timeout=defaut_timeout):
|
||||
def click(self, xpath, timeout=None):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
wait = WebDriverWait(self.driver, timeout)
|
||||
button = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
|
|
@ -482,15 +486,31 @@ class SeleniumLib(object):
|
|||
button.click()
|
||||
return button
|
||||
|
||||
def get_elements(self, by: str = By.ID, value: str = None, timeout=None):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
wait = WebDriverWait(self.driver, timeout)
|
||||
elements = wait.until(
|
||||
EC.visibility_of_any_elements_located(
|
||||
(
|
||||
by,
|
||||
value,
|
||||
)
|
||||
)
|
||||
)
|
||||
return elements
|
||||
|
||||
def get_element(
|
||||
self,
|
||||
by: str = By.ID,
|
||||
value: str = None,
|
||||
timeout=defaut_timeout,
|
||||
timeout=None,
|
||||
wait_clickable=False,
|
||||
wait_is_invisible=False,
|
||||
is_visible=True,
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
only_one = False
|
||||
wait = WebDriverWait(self.driver, timeout)
|
||||
if wait_clickable:
|
||||
|
|
@ -536,8 +556,10 @@ class SeleniumLib(object):
|
|||
return ele[0]
|
||||
|
||||
def get_element_not_visible(
|
||||
self, by: str = By.ID, value: str = None, timeout=defaut_timeout
|
||||
self, by: str = By.ID, value: str = None, timeout=None
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
wait = WebDriverWait(self.driver, timeout)
|
||||
ele = wait.until(
|
||||
EC.presence_of_element_located(
|
||||
|
|
@ -551,8 +573,10 @@ class SeleniumLib(object):
|
|||
return ele
|
||||
|
||||
def get_elements_not_visible(
|
||||
self, by: str = By.ID, value: str = None, timeout=defaut_timeout
|
||||
self, by: str = By.ID, value: str = None, timeout=None
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
wait = WebDriverWait(self.driver, timeout)
|
||||
ele = wait.until(
|
||||
EC.presence_of_all_elements_located(
|
||||
|
|
@ -566,14 +590,18 @@ class SeleniumLib(object):
|
|||
return ele
|
||||
|
||||
def get_text_from_element(
|
||||
self, by: str = By.ID, value: str = None, timeout=defaut_timeout
|
||||
self, by: str = By.ID, value: str = None, timeout: int = None
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
ele = self.get_element(by, value, timeout=timeout)
|
||||
return ele.text
|
||||
|
||||
def get_all_element(
|
||||
self, by: str = By.ID, value: str = None, timeout=defaut_timeout
|
||||
self, by: str = By.ID, value: str = None, timeout=None
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
wait = WebDriverWait(self.driver, timeout)
|
||||
ele = wait.until(
|
||||
EC.visibility_of_all_elements_located(
|
||||
|
|
@ -591,9 +619,11 @@ class SeleniumLib(object):
|
|||
by: str = By.ID,
|
||||
value: str = None,
|
||||
delay_wait=1,
|
||||
timeout=defaut_timeout,
|
||||
timeout=None,
|
||||
inject_cursor=False,
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
new_element = self.wait_new_element(
|
||||
by=by, value=value, delay_wait=delay_wait, timeout=timeout
|
||||
)
|
||||
|
|
@ -608,9 +638,11 @@ class SeleniumLib(object):
|
|||
by: str = By.ID,
|
||||
value: str = None,
|
||||
delay_wait=1,
|
||||
timeout=defaut_timeout,
|
||||
timeout=None,
|
||||
timeout_wait=None,
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
element = None
|
||||
# TODO support timeout_wait
|
||||
while element is None:
|
||||
|
|
@ -627,8 +659,10 @@ class SeleniumLib(object):
|
|||
by: str = By.ID,
|
||||
value: str = None,
|
||||
delay_wait=1,
|
||||
timeout=defaut_timeout,
|
||||
timeout=None,
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
new_element = self.wait_add_new_element(
|
||||
by=by, value=value, delay_wait=delay_wait, timeout=timeout
|
||||
)
|
||||
|
|
@ -640,8 +674,10 @@ class SeleniumLib(object):
|
|||
by: str = By.ID,
|
||||
value: str = None,
|
||||
delay_wait=1,
|
||||
timeout=defaut_timeout,
|
||||
timeout=None,
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
all_element_init = self.get_all_element(by=by, value=value)
|
||||
len_all_element_init = len(all_element_init)
|
||||
len_all_element = len_all_element_init
|
||||
|
|
@ -659,8 +695,10 @@ class SeleniumLib(object):
|
|||
by: str = By.ID,
|
||||
value: str = None,
|
||||
delay_wait=1,
|
||||
timeout=defaut_timeout,
|
||||
timeout=None,
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
element_init = self.get_element(by=by, value=value, timeout=timeout)
|
||||
actual_number = int(element_init.text)
|
||||
number_goal = actual_number + 1
|
||||
|
|
@ -678,7 +716,7 @@ class SeleniumLib(object):
|
|||
self,
|
||||
by: str = By.ID,
|
||||
value: str = None,
|
||||
timeout: int = defaut_timeout,
|
||||
timeout: int = None,
|
||||
no_scroll: bool = False,
|
||||
viewport_ele_by: str = By.ID,
|
||||
viewport_ele_value: str = None,
|
||||
|
|
@ -686,21 +724,41 @@ class SeleniumLib(object):
|
|||
with_index: bool = False,
|
||||
is_visible=True,
|
||||
index_of_list: int = 0,
|
||||
support_multiple_click: bool = False,
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
# Exemple Detect button by text value
|
||||
# Replace CLASS_NAME
|
||||
# By.CSS_SELECTOR, "input.CLASS_NAME[value='Fermer']"
|
||||
|
||||
# ele = self.driver.find_element(by, value)
|
||||
if element:
|
||||
ele = element
|
||||
else:
|
||||
ele = self.get_element(by, value, timeout, is_visible=is_visible)
|
||||
if support_multiple_click:
|
||||
ele = self.get_elements(by, value, timeout)
|
||||
else:
|
||||
ele = self.get_element(
|
||||
by, value, timeout, is_visible=is_visible
|
||||
)
|
||||
if not no_scroll:
|
||||
viewport_ele = None
|
||||
if viewport_ele_value:
|
||||
viewport_ele = self.get_element(
|
||||
viewport_ele_by, viewport_ele_value, timeout
|
||||
)
|
||||
self.scrollto_element(ele, viewport_ele=viewport_ele)
|
||||
# ActionChains(self.driver).move_to_element(ele).click().perform()
|
||||
ActionChains(self.driver).move_to_element(ele).perform()
|
||||
if support_multiple_click:
|
||||
for one_ele in ele:
|
||||
self.scrollto_element(one_ele, viewport_ele=viewport_ele)
|
||||
# ActionChains(self.driver).move_to_element(one_ele).click().perform()
|
||||
ActionChains(self.driver).move_to_element(
|
||||
one_ele
|
||||
).perform()
|
||||
else:
|
||||
self.scrollto_element(ele, viewport_ele=viewport_ele)
|
||||
# ActionChains(self.driver).move_to_element(ele).click().perform()
|
||||
ActionChains(self.driver).move_to_element(ele).perform()
|
||||
time.sleep(self.config.selenium_default_delay)
|
||||
wait = WebDriverWait(self.driver, timeout)
|
||||
if element:
|
||||
|
|
@ -717,36 +775,71 @@ class SeleniumLib(object):
|
|||
)
|
||||
button = buttons[index_of_list]
|
||||
else:
|
||||
button = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(
|
||||
by,
|
||||
value,
|
||||
if support_multiple_click:
|
||||
button = wait.until(
|
||||
# EC.visibility_of_all_elements_located(
|
||||
EC.presence_of_all_elements_located(
|
||||
(
|
||||
by,
|
||||
value,
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
else:
|
||||
button = wait.until(
|
||||
EC.element_to_be_clickable(
|
||||
(
|
||||
by,
|
||||
value,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
try:
|
||||
button.click()
|
||||
if support_multiple_click:
|
||||
for btn in button:
|
||||
try:
|
||||
btn.click()
|
||||
# This button work!
|
||||
return btn
|
||||
except Exception as e:
|
||||
pass
|
||||
else:
|
||||
button.click()
|
||||
except ElementClickInterceptedException as e:
|
||||
try:
|
||||
print(e)
|
||||
self.driver.execute_script(
|
||||
"arguments[0].scrollIntoView(true);", button
|
||||
)
|
||||
button.click()
|
||||
if support_multiple_click:
|
||||
for btn in button:
|
||||
self.driver.execute_script(
|
||||
"arguments[0].scrollIntoView(true);", btn
|
||||
)
|
||||
btn.click()
|
||||
else:
|
||||
self.driver.execute_script(
|
||||
"arguments[0].scrollIntoView(true);", button
|
||||
)
|
||||
button.click()
|
||||
except ElementClickInterceptedException as e:
|
||||
print(e)
|
||||
self.driver.execute_script("arguments[0].click();", button)
|
||||
if support_multiple_click:
|
||||
for btn in button:
|
||||
self.driver.execute_script(
|
||||
"arguments[0].click();", btn
|
||||
)
|
||||
else:
|
||||
self.driver.execute_script("arguments[0].click();", button)
|
||||
return button
|
||||
|
||||
def click_canvas_form(
|
||||
self,
|
||||
by: str = By.ID,
|
||||
value: str = None,
|
||||
timeout: int = defaut_timeout,
|
||||
timeout: int = None,
|
||||
form="carre",
|
||||
):
|
||||
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
# ele = self.driver.find_element(by, value)
|
||||
ele = self.get_element(by, value, timeout)
|
||||
actions = ActionChains(self.driver)
|
||||
|
|
@ -767,13 +860,15 @@ class SeleniumLib(object):
|
|||
value: str = None,
|
||||
text_value: str = "",
|
||||
delay_after: int = -1,
|
||||
timeout: int = defaut_timeout,
|
||||
timeout: int = None,
|
||||
no_scroll: bool = False,
|
||||
viewport_ele_by: str = By.ID,
|
||||
viewport_ele_value: str = None,
|
||||
clear_before: bool = False,
|
||||
click_before: bool = False,
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
# ele = self.driver.find_element(by, value)
|
||||
ele = self.get_element(by, value)
|
||||
if not no_scroll:
|
||||
|
|
@ -1006,8 +1101,13 @@ class SeleniumLib(object):
|
|||
# Switch to the new window and open new URL
|
||||
self.driver.switch_to.window(self.driver.window_handles[index])
|
||||
|
||||
def refresh(self):
|
||||
def refresh(self, wait_ready=False):
|
||||
self.driver.refresh()
|
||||
if wait_ready:
|
||||
WebDriverWait(self.driver, 15).until(
|
||||
lambda d: d.execute_script("return document.readyState")
|
||||
== "complete"
|
||||
)
|
||||
|
||||
def check_bot_chat_and_close(self):
|
||||
try:
|
||||
|
|
@ -1140,7 +1240,7 @@ class SeleniumLib(object):
|
|||
" odoo_website_menu_click"
|
||||
)
|
||||
button = self.click_with_mouse_move(
|
||||
By.LINK_TEXT, from_text, timeout=defaut_timeout
|
||||
By.LINK_TEXT, from_text, timeout=self.defaut_timeout
|
||||
)
|
||||
return button
|
||||
|
||||
|
|
@ -1282,8 +1382,10 @@ class SeleniumLib(object):
|
|||
return button
|
||||
|
||||
def odoo_web_form_click_statusbar_button_status(
|
||||
self, status_label, timeout=defaut_timeout
|
||||
self, status_label, timeout=None
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
try:
|
||||
status_button = self.click_with_mouse_move(
|
||||
By.XPATH,
|
||||
|
|
@ -1328,8 +1430,10 @@ class SeleniumLib(object):
|
|||
toggle_data_menu("color_scheme.switch", enable_dark_mode)
|
||||
|
||||
def odoo_web_form_click_statusbar_button_status_floating(
|
||||
self, status_label, timeout=defaut_timeout
|
||||
self, status_label, timeout=None
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
status_button = self.click_with_mouse_move(
|
||||
By.XPATH,
|
||||
"//span[contains(@class, 'o_arrow_button') and contains(text(),"
|
||||
|
|
@ -1340,8 +1444,10 @@ class SeleniumLib(object):
|
|||
return status_button
|
||||
|
||||
def odoo_web_form_click_statusbar_button_status_plus(
|
||||
self, status_label, timeout=defaut_timeout
|
||||
self, status_label, timeout=None
|
||||
):
|
||||
if timeout is None:
|
||||
timeout = self.defaut_timeout
|
||||
status_button = self.click_with_mouse_move(
|
||||
By.XPATH,
|
||||
"//button[contains(@class, 'o_arrow_button') and"
|
||||
|
|
@ -1897,11 +2003,13 @@ class SeleniumLib(object):
|
|||
else:
|
||||
return set()
|
||||
lst_unique_file = fichiers
|
||||
print(len(fichiers))
|
||||
print(fichiers)
|
||||
dct_document[filename] = os.path.join(
|
||||
self.default_download_dir_path, new_file_name
|
||||
)
|
||||
print(len(fichiers))
|
||||
print(fichiers)
|
||||
print("--")
|
||||
print(dct_document)
|
||||
if self.config.use_network:
|
||||
self.driver.download_file(
|
||||
new_file_name, target_directory=self.default_download_dir_path
|
||||
|
|
@ -1911,9 +2019,11 @@ class SeleniumLib(object):
|
|||
|
||||
def print_download_file_unique(self, dct_document, detect_error=False):
|
||||
dct_output = {"document": dct_document, "has_error": detect_error}
|
||||
print()
|
||||
print("=DATA=")
|
||||
print(json.dumps(dct_output))
|
||||
print("=ENDDATA=")
|
||||
print()
|
||||
|
||||
|
||||
def get_args(parser):
|
||||
|
|
|
|||
|
|
@ -192,6 +192,7 @@ def main():
|
|||
# Execute
|
||||
run(args, selenium_tool)
|
||||
selenium_tool.stop_record()
|
||||
input("Appuie sur Entrée pour fermer...")
|
||||
return 0
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue