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_show_logs_live:
|
||||||
docker compose logs -f
|
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
|
.PHONY: docker_show_process
|
||||||
docker_show_process:
|
docker_show_process:
|
||||||
docker compose ps
|
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,
|
"decembre": 12,
|
||||||
}
|
}
|
||||||
|
|
||||||
defaut_timeout = 60
|
|
||||||
|
|
||||||
|
|
||||||
class SeleniumLib(object):
|
class SeleniumLib(object):
|
||||||
def __init__(self, config):
|
def __init__(self, config):
|
||||||
|
|
@ -81,6 +79,7 @@ class SeleniumLib(object):
|
||||||
self.kdbx = None
|
self.kdbx = None
|
||||||
self.video_recorder = None
|
self.video_recorder = None
|
||||||
self.default_copy_download_dir_path = None
|
self.default_copy_download_dir_path = None
|
||||||
|
self.defaut_timeout = 60
|
||||||
|
|
||||||
date_str = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
date_str = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||||
tmp_dir = tempfile.mkdtemp(
|
tmp_dir = tempfile.mkdtemp(
|
||||||
|
|
@ -343,10 +342,6 @@ class SeleniumLib(object):
|
||||||
# Ajout de l'enregistrement
|
# Ajout de l'enregistrement
|
||||||
if self.config.record_mode:
|
if self.config.record_mode:
|
||||||
# Create recording
|
# 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
|
from script.selenium.selenium_video import VideoRecorder
|
||||||
|
|
||||||
if not os.path.isdir(self.dirname_recording):
|
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 self.config.use_firefox_driver and not self.config.use_network:
|
||||||
if not self.config.no_dark_mode:
|
if not self.config.no_dark_mode:
|
||||||
self.driver.install_addon(
|
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(
|
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 (
|
if (
|
||||||
not self.config.not_private_mode
|
not self.config.not_private_mode
|
||||||
and not self.config.no_dark_mode
|
and not self.config.no_dark_mode
|
||||||
):
|
):
|
||||||
|
# self.driver.set_context("chrome")
|
||||||
self.driver.get("about:addons")
|
self.driver.get("about:addons")
|
||||||
# Enable Dark Reader into incognito
|
# Enable Dark Reader into incognito
|
||||||
# Click Extensions
|
# Click Extensions
|
||||||
|
|
@ -469,7 +471,9 @@ class SeleniumLib(object):
|
||||||
word = ""
|
word = ""
|
||||||
return 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)
|
wait = WebDriverWait(self.driver, timeout)
|
||||||
button = wait.until(
|
button = wait.until(
|
||||||
EC.element_to_be_clickable(
|
EC.element_to_be_clickable(
|
||||||
|
|
@ -482,15 +486,31 @@ class SeleniumLib(object):
|
||||||
button.click()
|
button.click()
|
||||||
return button
|
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(
|
def get_element(
|
||||||
self,
|
self,
|
||||||
by: str = By.ID,
|
by: str = By.ID,
|
||||||
value: str = None,
|
value: str = None,
|
||||||
timeout=defaut_timeout,
|
timeout=None,
|
||||||
wait_clickable=False,
|
wait_clickable=False,
|
||||||
wait_is_invisible=False,
|
wait_is_invisible=False,
|
||||||
is_visible=True,
|
is_visible=True,
|
||||||
):
|
):
|
||||||
|
if timeout is None:
|
||||||
|
timeout = self.defaut_timeout
|
||||||
only_one = False
|
only_one = False
|
||||||
wait = WebDriverWait(self.driver, timeout)
|
wait = WebDriverWait(self.driver, timeout)
|
||||||
if wait_clickable:
|
if wait_clickable:
|
||||||
|
|
@ -536,8 +556,10 @@ class SeleniumLib(object):
|
||||||
return ele[0]
|
return ele[0]
|
||||||
|
|
||||||
def get_element_not_visible(
|
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)
|
wait = WebDriverWait(self.driver, timeout)
|
||||||
ele = wait.until(
|
ele = wait.until(
|
||||||
EC.presence_of_element_located(
|
EC.presence_of_element_located(
|
||||||
|
|
@ -551,8 +573,10 @@ class SeleniumLib(object):
|
||||||
return ele
|
return ele
|
||||||
|
|
||||||
def get_elements_not_visible(
|
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)
|
wait = WebDriverWait(self.driver, timeout)
|
||||||
ele = wait.until(
|
ele = wait.until(
|
||||||
EC.presence_of_all_elements_located(
|
EC.presence_of_all_elements_located(
|
||||||
|
|
@ -566,14 +590,18 @@ class SeleniumLib(object):
|
||||||
return ele
|
return ele
|
||||||
|
|
||||||
def get_text_from_element(
|
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)
|
ele = self.get_element(by, value, timeout=timeout)
|
||||||
return ele.text
|
return ele.text
|
||||||
|
|
||||||
def get_all_element(
|
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)
|
wait = WebDriverWait(self.driver, timeout)
|
||||||
ele = wait.until(
|
ele = wait.until(
|
||||||
EC.visibility_of_all_elements_located(
|
EC.visibility_of_all_elements_located(
|
||||||
|
|
@ -591,9 +619,11 @@ class SeleniumLib(object):
|
||||||
by: str = By.ID,
|
by: str = By.ID,
|
||||||
value: str = None,
|
value: str = None,
|
||||||
delay_wait=1,
|
delay_wait=1,
|
||||||
timeout=defaut_timeout,
|
timeout=None,
|
||||||
inject_cursor=False,
|
inject_cursor=False,
|
||||||
):
|
):
|
||||||
|
if timeout is None:
|
||||||
|
timeout = self.defaut_timeout
|
||||||
new_element = self.wait_new_element(
|
new_element = self.wait_new_element(
|
||||||
by=by, value=value, delay_wait=delay_wait, timeout=timeout
|
by=by, value=value, delay_wait=delay_wait, timeout=timeout
|
||||||
)
|
)
|
||||||
|
|
@ -608,9 +638,11 @@ class SeleniumLib(object):
|
||||||
by: str = By.ID,
|
by: str = By.ID,
|
||||||
value: str = None,
|
value: str = None,
|
||||||
delay_wait=1,
|
delay_wait=1,
|
||||||
timeout=defaut_timeout,
|
timeout=None,
|
||||||
timeout_wait=None,
|
timeout_wait=None,
|
||||||
):
|
):
|
||||||
|
if timeout is None:
|
||||||
|
timeout = self.defaut_timeout
|
||||||
element = None
|
element = None
|
||||||
# TODO support timeout_wait
|
# TODO support timeout_wait
|
||||||
while element is None:
|
while element is None:
|
||||||
|
|
@ -627,8 +659,10 @@ class SeleniumLib(object):
|
||||||
by: str = By.ID,
|
by: str = By.ID,
|
||||||
value: str = None,
|
value: str = None,
|
||||||
delay_wait=1,
|
delay_wait=1,
|
||||||
timeout=defaut_timeout,
|
timeout=None,
|
||||||
):
|
):
|
||||||
|
if timeout is None:
|
||||||
|
timeout = self.defaut_timeout
|
||||||
new_element = self.wait_add_new_element(
|
new_element = self.wait_add_new_element(
|
||||||
by=by, value=value, delay_wait=delay_wait, timeout=timeout
|
by=by, value=value, delay_wait=delay_wait, timeout=timeout
|
||||||
)
|
)
|
||||||
|
|
@ -640,8 +674,10 @@ class SeleniumLib(object):
|
||||||
by: str = By.ID,
|
by: str = By.ID,
|
||||||
value: str = None,
|
value: str = None,
|
||||||
delay_wait=1,
|
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)
|
all_element_init = self.get_all_element(by=by, value=value)
|
||||||
len_all_element_init = len(all_element_init)
|
len_all_element_init = len(all_element_init)
|
||||||
len_all_element = len_all_element_init
|
len_all_element = len_all_element_init
|
||||||
|
|
@ -659,8 +695,10 @@ class SeleniumLib(object):
|
||||||
by: str = By.ID,
|
by: str = By.ID,
|
||||||
value: str = None,
|
value: str = None,
|
||||||
delay_wait=1,
|
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)
|
element_init = self.get_element(by=by, value=value, timeout=timeout)
|
||||||
actual_number = int(element_init.text)
|
actual_number = int(element_init.text)
|
||||||
number_goal = actual_number + 1
|
number_goal = actual_number + 1
|
||||||
|
|
@ -678,7 +716,7 @@ class SeleniumLib(object):
|
||||||
self,
|
self,
|
||||||
by: str = By.ID,
|
by: str = By.ID,
|
||||||
value: str = None,
|
value: str = None,
|
||||||
timeout: int = defaut_timeout,
|
timeout: int = None,
|
||||||
no_scroll: bool = False,
|
no_scroll: bool = False,
|
||||||
viewport_ele_by: str = By.ID,
|
viewport_ele_by: str = By.ID,
|
||||||
viewport_ele_value: str = None,
|
viewport_ele_value: str = None,
|
||||||
|
|
@ -686,21 +724,41 @@ class SeleniumLib(object):
|
||||||
with_index: bool = False,
|
with_index: bool = False,
|
||||||
is_visible=True,
|
is_visible=True,
|
||||||
index_of_list: int = 0,
|
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)
|
# ele = self.driver.find_element(by, value)
|
||||||
if element:
|
if element:
|
||||||
ele = element
|
ele = element
|
||||||
else:
|
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:
|
if not no_scroll:
|
||||||
viewport_ele = None
|
viewport_ele = None
|
||||||
if viewport_ele_value:
|
if viewport_ele_value:
|
||||||
viewport_ele = self.get_element(
|
viewport_ele = self.get_element(
|
||||||
viewport_ele_by, viewport_ele_value, timeout
|
viewport_ele_by, viewport_ele_value, timeout
|
||||||
)
|
)
|
||||||
self.scrollto_element(ele, viewport_ele=viewport_ele)
|
if support_multiple_click:
|
||||||
# ActionChains(self.driver).move_to_element(ele).click().perform()
|
for one_ele in ele:
|
||||||
ActionChains(self.driver).move_to_element(ele).perform()
|
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)
|
time.sleep(self.config.selenium_default_delay)
|
||||||
wait = WebDriverWait(self.driver, timeout)
|
wait = WebDriverWait(self.driver, timeout)
|
||||||
if element:
|
if element:
|
||||||
|
|
@ -717,36 +775,71 @@ class SeleniumLib(object):
|
||||||
)
|
)
|
||||||
button = buttons[index_of_list]
|
button = buttons[index_of_list]
|
||||||
else:
|
else:
|
||||||
button = wait.until(
|
if support_multiple_click:
|
||||||
EC.element_to_be_clickable(
|
button = wait.until(
|
||||||
(
|
# EC.visibility_of_all_elements_located(
|
||||||
by,
|
EC.presence_of_all_elements_located(
|
||||||
value,
|
(
|
||||||
|
by,
|
||||||
|
value,
|
||||||
|
)
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
)
|
else:
|
||||||
|
button = wait.until(
|
||||||
|
EC.element_to_be_clickable(
|
||||||
|
(
|
||||||
|
by,
|
||||||
|
value,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
try:
|
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:
|
except ElementClickInterceptedException as e:
|
||||||
try:
|
try:
|
||||||
print(e)
|
print(e)
|
||||||
self.driver.execute_script(
|
if support_multiple_click:
|
||||||
"arguments[0].scrollIntoView(true);", button
|
for btn in button:
|
||||||
)
|
self.driver.execute_script(
|
||||||
button.click()
|
"arguments[0].scrollIntoView(true);", btn
|
||||||
|
)
|
||||||
|
btn.click()
|
||||||
|
else:
|
||||||
|
self.driver.execute_script(
|
||||||
|
"arguments[0].scrollIntoView(true);", button
|
||||||
|
)
|
||||||
|
button.click()
|
||||||
except ElementClickInterceptedException as e:
|
except ElementClickInterceptedException as e:
|
||||||
print(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
|
return button
|
||||||
|
|
||||||
def click_canvas_form(
|
def click_canvas_form(
|
||||||
self,
|
self,
|
||||||
by: str = By.ID,
|
by: str = By.ID,
|
||||||
value: str = None,
|
value: str = None,
|
||||||
timeout: int = defaut_timeout,
|
timeout: int = None,
|
||||||
form="carre",
|
form="carre",
|
||||||
):
|
):
|
||||||
|
if timeout is None:
|
||||||
|
timeout = self.defaut_timeout
|
||||||
# ele = self.driver.find_element(by, value)
|
# ele = self.driver.find_element(by, value)
|
||||||
ele = self.get_element(by, value, timeout)
|
ele = self.get_element(by, value, timeout)
|
||||||
actions = ActionChains(self.driver)
|
actions = ActionChains(self.driver)
|
||||||
|
|
@ -767,13 +860,15 @@ class SeleniumLib(object):
|
||||||
value: str = None,
|
value: str = None,
|
||||||
text_value: str = "",
|
text_value: str = "",
|
||||||
delay_after: int = -1,
|
delay_after: int = -1,
|
||||||
timeout: int = defaut_timeout,
|
timeout: int = None,
|
||||||
no_scroll: bool = False,
|
no_scroll: bool = False,
|
||||||
viewport_ele_by: str = By.ID,
|
viewport_ele_by: str = By.ID,
|
||||||
viewport_ele_value: str = None,
|
viewport_ele_value: str = None,
|
||||||
clear_before: bool = False,
|
clear_before: bool = False,
|
||||||
click_before: bool = False,
|
click_before: bool = False,
|
||||||
):
|
):
|
||||||
|
if timeout is None:
|
||||||
|
timeout = self.defaut_timeout
|
||||||
# ele = self.driver.find_element(by, value)
|
# ele = self.driver.find_element(by, value)
|
||||||
ele = self.get_element(by, value)
|
ele = self.get_element(by, value)
|
||||||
if not no_scroll:
|
if not no_scroll:
|
||||||
|
|
@ -1006,8 +1101,13 @@ class SeleniumLib(object):
|
||||||
# Switch to the new window and open new URL
|
# Switch to the new window and open new URL
|
||||||
self.driver.switch_to.window(self.driver.window_handles[index])
|
self.driver.switch_to.window(self.driver.window_handles[index])
|
||||||
|
|
||||||
def refresh(self):
|
def refresh(self, wait_ready=False):
|
||||||
self.driver.refresh()
|
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):
|
def check_bot_chat_and_close(self):
|
||||||
try:
|
try:
|
||||||
|
|
@ -1140,7 +1240,7 @@ class SeleniumLib(object):
|
||||||
" odoo_website_menu_click"
|
" odoo_website_menu_click"
|
||||||
)
|
)
|
||||||
button = self.click_with_mouse_move(
|
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
|
return button
|
||||||
|
|
||||||
|
|
@ -1282,8 +1382,10 @@ class SeleniumLib(object):
|
||||||
return button
|
return button
|
||||||
|
|
||||||
def odoo_web_form_click_statusbar_button_status(
|
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:
|
try:
|
||||||
status_button = self.click_with_mouse_move(
|
status_button = self.click_with_mouse_move(
|
||||||
By.XPATH,
|
By.XPATH,
|
||||||
|
|
@ -1328,8 +1430,10 @@ class SeleniumLib(object):
|
||||||
toggle_data_menu("color_scheme.switch", enable_dark_mode)
|
toggle_data_menu("color_scheme.switch", enable_dark_mode)
|
||||||
|
|
||||||
def odoo_web_form_click_statusbar_button_status_floating(
|
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(
|
status_button = self.click_with_mouse_move(
|
||||||
By.XPATH,
|
By.XPATH,
|
||||||
"//span[contains(@class, 'o_arrow_button') and contains(text(),"
|
"//span[contains(@class, 'o_arrow_button') and contains(text(),"
|
||||||
|
|
@ -1340,8 +1444,10 @@ class SeleniumLib(object):
|
||||||
return status_button
|
return status_button
|
||||||
|
|
||||||
def odoo_web_form_click_statusbar_button_status_plus(
|
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(
|
status_button = self.click_with_mouse_move(
|
||||||
By.XPATH,
|
By.XPATH,
|
||||||
"//button[contains(@class, 'o_arrow_button') and"
|
"//button[contains(@class, 'o_arrow_button') and"
|
||||||
|
|
@ -1897,11 +2003,13 @@ class SeleniumLib(object):
|
||||||
else:
|
else:
|
||||||
return set()
|
return set()
|
||||||
lst_unique_file = fichiers
|
lst_unique_file = fichiers
|
||||||
print(len(fichiers))
|
|
||||||
print(fichiers)
|
|
||||||
dct_document[filename] = os.path.join(
|
dct_document[filename] = os.path.join(
|
||||||
self.default_download_dir_path, new_file_name
|
self.default_download_dir_path, new_file_name
|
||||||
)
|
)
|
||||||
|
print(len(fichiers))
|
||||||
|
print(fichiers)
|
||||||
|
print("--")
|
||||||
|
print(dct_document)
|
||||||
if self.config.use_network:
|
if self.config.use_network:
|
||||||
self.driver.download_file(
|
self.driver.download_file(
|
||||||
new_file_name, target_directory=self.default_download_dir_path
|
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):
|
def print_download_file_unique(self, dct_document, detect_error=False):
|
||||||
dct_output = {"document": dct_document, "has_error": detect_error}
|
dct_output = {"document": dct_document, "has_error": detect_error}
|
||||||
|
print()
|
||||||
print("=DATA=")
|
print("=DATA=")
|
||||||
print(json.dumps(dct_output))
|
print(json.dumps(dct_output))
|
||||||
print("=ENDDATA=")
|
print("=ENDDATA=")
|
||||||
|
print()
|
||||||
|
|
||||||
|
|
||||||
def get_args(parser):
|
def get_args(parser):
|
||||||
|
|
|
||||||
|
|
@ -192,6 +192,7 @@ def main():
|
||||||
# Execute
|
# Execute
|
||||||
run(args, selenium_tool)
|
run(args, selenium_tool)
|
||||||
selenium_tool.stop_record()
|
selenium_tool.stop_record()
|
||||||
|
input("Appuie sur Entrée pour fermer...")
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue