Mathieu Benoit
770de6b0ca
La migration lance ses outils avec le venv de la version Odoo courante. Au premier palier c'est celui de la 12, en 3.7, où « dict | None » (3.10) et « tuple[str, str] » (3.9) sont ÉVALUÉS au chargement du module. La restauration du zip mourait donc sur un TypeError avant d'avoir rien fait, dans execute.py — importé par db_restore.py. `from __future__ import annotations` existe depuis 3.7 et le dépôt s'en sert déjà dans treize fichiers. Un test le vérifie maintenant sur toute la fermeture d'imports des outils que le pilote lance, points d'entrée lus dans le pilote pour que le script ajouté demain soit couvert. --- EN --- The migration runs its tools with the venv of the current Odoo version. At the first step that is 12's, on 3.7, where « dict | None » (3.10) and « tuple[str, str] » (3.9) are EVALUATED when the module loads. Restoring the zip therefore died on a TypeError before doing anything at all, in execute.py — imported by db_restore.py. `from __future__ import annotations` has existed since 3.7 and the repo already uses it in thirteen files. A test now checks the whole import closure of the tools the driver launches, with the entry points read from the driver so tomorrow's script is covered too. Assisted-by: Claude Opus 5
260 lines
9.9 KiB
Python
260 lines
9.9 KiB
Python
#!/usr/bin/env python3
|
|
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
|
|
|
# Annotations différées : la migration charge ce module sous le Python
|
|
# d'Odoo 12 — 3.7 — où « dict | None » et « tuple[str, str] » n'existent
|
|
# pas encore. Sans ceci, l'annotation est ÉVALUÉE au chargement et la
|
|
# migration meurt sur un TypeError avant d'avoir rien fait.
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
try:
|
|
import humanize
|
|
except ModuleNotFoundError as e:
|
|
humanize = None
|
|
|
|
VENV_ERPLIBRE = ".venv.erplibre"
|
|
|
|
# Une commande construite ailleurs peut porter un secret en clair : todo.py et
|
|
# kdbx_manager.py y mettent « --default_password_auth '<mot de passe KeePass>' ».
|
|
# Cette commande est affichée avant et après l'exécution, et journalisée en
|
|
# erreur : le secret finissait donc dans le terminal, dans les journaux et dans
|
|
# toute sortie CI qui les capture.
|
|
#
|
|
# Ce filtre reste le dernier rempart, pas le premier : un secret n'a rien à
|
|
# faire sur argv, que /proc/<pid>/cmdline expose à tout utilisateur de la
|
|
# machine et qu'aucun caviardage n'atteint. db_restore.py est passé à
|
|
# MASTER_PWD dans l'environnement pour cette raison.
|
|
#
|
|
# On caviarde la VALEUR, jamais le nom de l'option : la commande reste lisible et
|
|
# reproductible, il ne manque que ce qui ne doit pas être lu.
|
|
_SECRET_OPTION = re.compile(
|
|
r"(?P<opt>--?[\w-]*"
|
|
r"(?:password|passwd|pwd|secret|token|api[-_]?key)[\w-]*"
|
|
r"(?:\s+|=))"
|
|
r"(?P<val>'[^']*'|\"[^\"]*\"|\S+)",
|
|
re.IGNORECASE,
|
|
)
|
|
_SECRET_ENV = re.compile(
|
|
r"(?P<var>\b\w*(?:PASSWORD|PASSWD|SECRET|TOKEN)\w*=)"
|
|
r"(?P<val>'[^']*'|\"[^\"]*\"|\S+)"
|
|
)
|
|
|
|
|
|
def redact_secrets(text):
|
|
"""Remplace la valeur des options et variables porteuses de secret.
|
|
|
|
Appliqué à CHAQUE affichage d'une commande. Filtrer au point d'affichage
|
|
plutôt qu'à la construction est ce qui rend la garantie tenable : il n'y a
|
|
qu'une poignée de sorties ici, alors que les commandes se construisent
|
|
partout dans le dépôt.
|
|
"""
|
|
if not text:
|
|
return text
|
|
text = _SECRET_OPTION.sub(lambda m: m.group("opt") + "'***'", text)
|
|
return _SECRET_ENV.sub(lambda m: m.group("var") + "'***'", text)
|
|
|
|
|
|
new_path = os.path.normpath(
|
|
os.path.join(os.path.dirname(__file__), "..", "..")
|
|
)
|
|
sys.path.append(new_path)
|
|
|
|
|
|
logging.basicConfig(
|
|
format=(
|
|
"%(asctime)s,%(msecs)d %(levelname)-8s [%(filename)s:%(lineno)d]"
|
|
" %(message)s"
|
|
),
|
|
datefmt="%Y-%m-%d:%H:%M:%S",
|
|
level=logging.INFO,
|
|
)
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Execute:
|
|
def __init__(self) -> None:
|
|
self.cmd_source_erplibre: str = ""
|
|
self.cmd_source_default: str = ""
|
|
exec_path_gnome_terminal = shutil.which("gnome-terminal")
|
|
if exec_path_gnome_terminal:
|
|
self.cmd_source_erplibre = (
|
|
f"gnome-terminal -- bash -c 'source"
|
|
f" ./{VENV_ERPLIBRE}/bin/activate;%s'"
|
|
)
|
|
self.cmd_source_default = "gnome-terminal -- bash -c '" f"%s'"
|
|
else:
|
|
exec_path_tell = shutil.which("osascript")
|
|
if exec_path_tell:
|
|
self.cmd_source_erplibre = (
|
|
"osascript -e 'tell application \"Terminal\"'"
|
|
)
|
|
self.cmd_source_erplibre += " -e 'tell application \"System Events\" to keystroke \"t\" using {command down}' -e 'delay 0.1' -e 'do script \""
|
|
self.cmd_source_erplibre += f"cd {os.getcwd()}; source ./{VENV_ERPLIBRE}/bin/activate; %s\" in front window'"
|
|
self.cmd_source_erplibre += " -e 'end tell'"
|
|
else:
|
|
self.cmd_source_erplibre = (
|
|
f"source ./{VENV_ERPLIBRE}/bin/activate;%s"
|
|
)
|
|
|
|
def exec_command_live(
|
|
self,
|
|
command: str,
|
|
source_erplibre: bool = True,
|
|
quiet: bool = False,
|
|
single_source_erplibre: bool = False,
|
|
new_window: bool = False,
|
|
single_source_odoo: bool = False,
|
|
source_odoo: str = "",
|
|
new_env: dict | None = None,
|
|
return_status_and_command: bool = False,
|
|
return_status_and_output: bool = False,
|
|
return_status_and_output_and_command: bool = False,
|
|
) -> (
|
|
int
|
|
| tuple[int, str]
|
|
| tuple[int, list[str]]
|
|
| tuple[int, str, list[str]]
|
|
):
|
|
"""
|
|
Execute a command and display its output live.
|
|
|
|
Args:
|
|
command (str): The command to execute.
|
|
"""
|
|
|
|
my_env = os.environ.copy()
|
|
if new_env:
|
|
my_env.update(new_env)
|
|
|
|
process_start_time = time.time()
|
|
exit_code = None
|
|
if source_erplibre:
|
|
# command = f"source ./{VENV_ERPLIBRE}/bin/activate && " + command
|
|
# cmd = (
|
|
# f"gnome-terminal --tab -- bash -c 'source"
|
|
# f" ./{VENV_ERPLIBRE}/bin/activate;{command}'"
|
|
# )
|
|
command = self.cmd_source_erplibre % command
|
|
# os.system(f"./script/terminal/open_terminal.sh {command}")
|
|
elif single_source_erplibre:
|
|
command = f"source ./{VENV_ERPLIBRE}/bin/activate && %s" % command
|
|
elif single_source_odoo:
|
|
if not source_odoo and os.path.exists("./.erplibre-version"):
|
|
with open("./.erplibre-version") as f:
|
|
source_odoo = f.read()
|
|
if not source_odoo:
|
|
_logger.error(
|
|
"You cannot execute Odoo command if no version is"
|
|
f" installed. Command : {redact_secrets(command)}"
|
|
)
|
|
# Return the SAME shape the caller asked for. A bare int here
|
|
# made callers doing « status, cmd = exec_command_live(...) »
|
|
# crash with ValueError instead of seeing the failure.
|
|
if return_status_and_output_and_command:
|
|
return 1, command, []
|
|
if return_status_and_command:
|
|
return 1, command
|
|
if return_status_and_output:
|
|
return 1, []
|
|
return 1
|
|
command = f"source ./.venv.{source_odoo}/bin/activate && {command}"
|
|
if new_window and self.cmd_source_default:
|
|
command = self.cmd_source_default % command
|
|
|
|
if not quiet:
|
|
print("🏠 ⬇ Execute command :\n")
|
|
print(redact_secrets(command))
|
|
output_lines = []
|
|
|
|
try:
|
|
process = subprocess.Popen(
|
|
command,
|
|
shell=True,
|
|
executable="/bin/bash",
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
bufsize=1, # Disable buffering for live output
|
|
universal_newlines=True, # Handle line breaks correctly
|
|
env=my_env,
|
|
)
|
|
|
|
sink = getattr(self, "log_sink", None)
|
|
while True:
|
|
line = process.stdout.readline()
|
|
if not line:
|
|
break
|
|
# La sortie du sous-processus passe par le meme filtre que
|
|
# la commande : un outil qui reaffiche ses propres arguments
|
|
# (« set -x », une trace, odoo_bin.sh) y remettrait le secret
|
|
# que la ligne 165 venait d'ecarter.
|
|
line = redact_secrets(line)
|
|
if not quiet:
|
|
print(line, end="")
|
|
if sink:
|
|
# Chaque ligne passe DÉJÀ ici : c'est le seul endroit où
|
|
# journaliser sans rien changer à ce que le terminal
|
|
# montre. Une erreur d'écriture ne doit jamais faire
|
|
# échouer la commande qu'on est en train de suivre.
|
|
try:
|
|
sink.write(line)
|
|
except Exception:
|
|
sink = None
|
|
if (
|
|
return_status_and_output
|
|
or return_status_and_output_and_command
|
|
):
|
|
# Remove last \n char
|
|
output_lines.append(
|
|
line.removesuffix("\r\n")
|
|
.removesuffix("\n")
|
|
.removesuffix("\r")
|
|
)
|
|
|
|
process.wait()
|
|
exit_code = process.returncode
|
|
if process.returncode != 0 and not quiet:
|
|
print("Command returned error code:" f" {process.returncode}")
|
|
|
|
# An exception MUST report a failure. exit_code stays None otherwise,
|
|
# and None is falsy: callers testing « if not status: » would mark the
|
|
# step as done, and « if status and wait_at_error » would skip the error
|
|
# prompt. A crashed command was therefore recorded as a success.
|
|
except FileNotFoundError:
|
|
exit_code = 1
|
|
if not quiet:
|
|
print(f"Error: Command '{redact_secrets(command)}' not found.")
|
|
except Exception as e:
|
|
exit_code = 1
|
|
if not quiet:
|
|
print(f"An error occurred: {redact_secrets(str(e))}")
|
|
process_end_time = time.time()
|
|
duration_sec = process_end_time - process_start_time
|
|
if humanize:
|
|
duration_delta = datetime.timedelta(seconds=duration_sec)
|
|
human_time = humanize.precisedelta(duration_delta)
|
|
if not quiet:
|
|
print(f"🏠 ⬆ Executed ({human_time}) :\n")
|
|
else:
|
|
if not quiet:
|
|
print(f"🏠 ⬆ Executed ({duration_sec:.2f} sec.) :\n")
|
|
if not quiet:
|
|
print(redact_secrets(command))
|
|
print()
|
|
if return_status_and_output_and_command:
|
|
return exit_code, command, output_lines
|
|
if return_status_and_command:
|
|
return exit_code, command
|
|
if return_status_and_output:
|
|
return exit_code, output_lines
|
|
return exit_code
|