Compare commits

...

10 commits

5 changed files with 800 additions and 46 deletions

View file

@ -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
View 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()

View file

@ -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):

View file

@ -192,6 +192,7 @@ def main():
# Execute
run(args, selenium_tool)
selenium_tool.stop_record()
input("Appuie sur Entrée pour fermer...")
return 0