[ADD] migration: an auto-run that takes the default after five seconds

A migration asks dozens of questions whose answer is almost always the
one offered, and someone has to stand there pressing Enter for hours.
Asked before the version choice — so it covers that choice too, now
that it has a default — and off by default: it decides in your place.

Every prompt of the migration goes through one reader, seventeen of
them. A prompt left as a bare input() would know nothing of auto mode
and would block the run with nothing to say why; a test reads the
function and rejects any that appears.

click.prompt had to go: it cannot hand back control after a delay, so
auto would have stopped at the first question. select() rather than a
thread or an alarm — a thread would leave a blocked input() behind it,
stealing the next keystroke.

--- FR ---

[ADD] migration : une auto-exécution qui prend le défaut après cinq secondes

Une migration pose des dizaines de questions dont la réponse est presque
toujours celle proposée, et il faut rester devant à taper Entrée pendant
des heures. Posée avant le choix de version — donc elle vaut aussi pour
lui, maintenant qu'il a un défaut — et éteinte par défaut : elle décide
à votre place.

Toutes les invites de la migration passent par une seule lecture,
dix-sept d'un coup. Une invite laissée en input() nu ne saurait rien du
mode auto et bloquerait la migration sans rien dire ; un test lit la
fonction et refuse toute apparition.

click.prompt devait partir : il ne sait pas rendre la main après un
délai, l'auto se serait arrêté à la première question. select() plutôt
qu'un fil ou une alarme — un fil laisserait un input() bloqué derrière
lui, qui volerait la frappe suivante.

Assisted-by: Claude Opus 5
This commit is contained in:
Mathieu Benoit 2026-08-17 17:35:31 -04:00
parent 1831b234ea
commit cb77394869
4 changed files with 353 additions and 37 deletions

View file

@ -5246,6 +5246,26 @@ TRANSLATIONS = {
"fr": "Nettoyer la base avant de tester les pages ?",
"en": "Clean the database before testing the pages?",
},
"Auto-run: take the default answer after": {
"fr": "Auto-exécution : prendre la réponse par défaut après",
"en": "Auto-run: take the default answer after",
},
"seconds?": {
"fr": "secondes ?",
"en": "seconds?",
},
"Auto-run on: each prompt waits": {
"fr": "Auto-exécution activée : chaque invite attend",
"en": "Auto-run on: each prompt waits",
},
"seconds, then takes its default.": {
"fr": "secondes, puis prend son défaut.",
"en": "seconds, then takes its default.",
},
"(default)": {
"fr": "(défaut)",
"en": "(default)",
},
"Nothing to decide yet": {
"fr": "Rien à décider pour l'instant",
"en": "Nothing to decide yet",

View file

@ -13,7 +13,6 @@ import sys
import zipfile
from uuid import uuid4
import click
from script.todo import todo_file_browser
from script.todo.version_manager import get_odoo_version
@ -761,6 +760,61 @@ class TodoUpgrade:
]
return dct_kept
AUTO_DELAY = 5
def prompt_auto_execute(self):
"""Proposer que les invites prennent leur défaut après un délai.
Une migration pose des dizaines de questions dont la réponse est
presque toujours celle proposée. Les enchaîner à la main immobilise
quelqu'un pendant des heures pour taper Entrée.
Posée AVANT le choix de version, donc avant la première décision :
activée, elle vaut pour toutes les suivantes, celle-là comprise.
"""
answer = (
input(
f"💬 {t('Auto-run: take the default answer after')}"
f" {self.AUTO_DELAY} {t('seconds?')} (y/N) : "
)
.strip()
.lower()
)
self.auto_execute = answer == "y"
if self.auto_execute:
print(
f"⏱ {t('Auto-run on: each prompt waits')}"
f" {self.AUTO_DELAY} {t('seconds, then takes its default.')}"
)
def ask(self, prompt, default=""):
"""Lire une réponse ; en mode auto, rendre le défaut après le délai.
`select` plutôt qu'un fil ou une alarme : on veut savoir si quelque
chose est LISIBLE, et rendre la main sinon. Un fil laisserait un
`input()` bloqué derrière lui, qui volerait la frappe suivante.
"""
if not getattr(self, "auto_execute", False):
return input(prompt)
import select
sys.stdout.write(prompt)
sys.stdout.flush()
try:
ready, _, _ = select.select([sys.stdin], [], [], self.AUTO_DELAY)
except (OSError, ValueError):
# stdin n'est pas sélectionnable : on ne devine pas, on demande.
return input("")
if ready:
answer = sys.stdin.readline().rstrip("\n")
# Une réponse VIDE vaut « prends le défaut » : c'est tout le
# propos du mode auto. Un stdin fermé — exécution non
# interactive — est lisible tout de suite et rend justement
# une ligne vide ; sans ceci, le défaut ne servirait jamais là.
return answer or default
print(f" ⏱ → {default or t('(default)')}")
return default
def ask_gate(self, prompt):
"""Une invite d'attente, avec une porte de sortie vers l'arrière.
@ -771,7 +825,7 @@ class TodoUpgrade:
il rembobine l'état, l'écrit, et s'arrête en disant quoi relancer.
"""
while True:
answer = input(prompt)
answer = self.ask(prompt)
if (answer or "").strip().lower() != "b":
return answer
if self.rewind_to_chosen_step():
@ -1161,6 +1215,7 @@ class TodoUpgrade:
# TODO Redeploy new production after upgrade
# 2 upgrades version = 5 environnement. 0-prod init, 1-dev init, 2-dev01, 3-dev02, 4-prod final
print(t("Welcome to the Odoo database upgrade with ERPLibre") + " 🤖")
self.prompt_auto_execute()
self.lst_command_executed = []
self.dct_module_per_version = {}
self.dct_module_per_dct_version_path = {}
@ -1197,7 +1252,7 @@ class TodoUpgrade:
print("")
print(t("Select the zip file of your database backup."))
self.file_path = input(
self.file_path = self.ask(
f"💬 {t('Give the path of the file, or empty to use a file')}"
f" {t('browser, or type')} 'remote'"
f" {t('to download from production')} : "
@ -1283,14 +1338,17 @@ class TodoUpgrade:
odoo_target_version = None
cmd_no_found = True
while cmd_no_found:
# click affiche « [6] » et rend « 6 » sur Entrée : le chemin
# normal traite la réponse, sans cas particulier à tenir
# d'accord avec lui.
status = click.prompt(
help_info,
# `self.ask` et non click.prompt : c'est lui qui porte le
# mode auto, et click ne sait pas rendre la main après un
# délai — l'auto se serait arrêté à la première question.
# La marque « [6] » est donc posée à la main.
marque = f" [{default_index}]" if default_index else ""
status = self.ask(
f"{help_info}{marque}: ",
default=str(default_index) if default_index else "",
show_default=bool(default_index),
)
if not str(status).strip() and default_index:
status = str(default_index)
try:
int_cmd = int(status)
if 0 < int_cmd <= len(lst_odoo_version):
@ -1353,7 +1411,7 @@ class TodoUpgrade:
iter_range_version = odoo_version_to_install.replace(
"odoo", ""
).replace(".0", "")
want_continue = input(
want_continue = self.ask(
f"💬 {t('Would you like to install')}"
f" '{odoo_version_to_install}' (y/Y) : "
)
@ -1426,7 +1484,7 @@ class TodoUpgrade:
if lst_module_duplicate:
print(f"{t('Duplicate module')} :")
print(lst_module_duplicate)
want_continue = input(
want_continue = self.ask(
f"💬 {t('Missing or duplicate module detected at init,')}"
f" {t('do you want to continue?')} (Y/N) : "
)
@ -1453,7 +1511,7 @@ class TodoUpgrade:
database_name = self.dct_progression.get("config_database_name")
if not database_name:
database_name = (
input(
self.ask(
f"💬 {t('Which database name do you want to work with?')}"
f" {t('Default')} ({default_database_name}) : "
).strip()
@ -1492,7 +1550,7 @@ class TodoUpgrade:
if not shutil._samefile(self.file_path, image_db_file_path):
do_copy = False
if os.path.exists(image_db_file_path):
status_overwrite_image_db = input(
status_overwrite_image_db = self.ask(
f"{str_will_copy}, "
f"{t('a file already exists, do you want to')}"
f" {t('continue?')} (y/Y) : "
@ -1528,7 +1586,7 @@ class TodoUpgrade:
f" {t('neutralized by Odoo if supported)')}"
)
wait_continue = (
input(
self.ask(
f"💬 {t('Do you need to upgrade before neutralizing the')}"
f" {t('database? Press enter to ignore')} : "
)
@ -1731,7 +1789,7 @@ class TodoUpgrade:
if status.lower().strip() == "y":
self.todo.prompt_execute_selenium_and_run_db(database_name)
status = input(
status = self.ask(
f"💬 {t('Press enter to continue step 3')} : "
).strip()
@ -2060,7 +2118,7 @@ class TodoUpgrade:
if lst_module_duplicate:
print(f"{t('Duplicate module in Odoo')}{next_version} : ")
print(lst_module_duplicate)
input(
self.ask(
f"💬 {t('Duplicate module error detected, handle it')}"
f" {t('manually then press enter to continue.')}"
)
@ -2085,7 +2143,7 @@ class TodoUpgrade:
print(f"[e] {t('Add an extra custom one')}")
want_continue = (
input(
self.ask(
f"💬 {t('List the missing modules to delete,')}"
f" {t('separated by commas. The others will be')}"
f" {t('migrated')} : "
@ -2149,7 +2207,7 @@ class TodoUpgrade:
if "e" in lst_want_continue:
want_continue = (
input(
self.ask(
f"💬 {t('List the module names to delete,')}"
f" {t('separated by commas')} : "
)
@ -2577,7 +2635,7 @@ class TodoUpgrade:
)
status = (
input(
self.ask(
f"💬 {t('Do you want to upgrade all')}"
f"{str_wait_next_version} ?"
f" {t('Press y/Y to upgrade all addons of the')}"
@ -2610,7 +2668,7 @@ class TodoUpgrade:
print(f"[y] {t('Open the server with Selenium')}")
status = (
input(
self.ask(
f"💬 {t('Do you want to test this upgrade? Choose')}"
f" {t('or press enter to ignore it')} : "
)
@ -2622,7 +2680,7 @@ class TodoUpgrade:
self.todo.prompt_execute_selenium_and_run_db(
database_name_upgrade
)
status = input(
status = self.ask(
f"💬 {t('Press enter to continue')} 4.{index} : "
).strip()
else:
@ -2632,7 +2690,7 @@ class TodoUpgrade:
)
#
# waiting_input = input("💬 Press any keyboard key to continue...")
# waiting_input = self.ask("💬 Press any keyboard key to continue...")
print("")
msg = "5 - Cleaning up database after upgrade"
@ -2643,7 +2701,7 @@ class TodoUpgrade:
f"✨ {t('Re-update i18n, purge the data and the tables')}"
f" ({t('except mail_test and mail_test_full')})"
)
# waiting_input = input("💬print Press any keyboard key to continue...")
# waiting_input = self.ask("💬print Press any keyboard key to continue...")
msg = "6 - Migration finished"
self.print_step(msg)
self.add_comment_progression(msg)
@ -2651,7 +2709,7 @@ class TodoUpgrade:
cmd_backup_template = f"./odoo_bin.sh db --backup --database {database_name_upgrade} --restore_image"
cmd_backup = f"{cmd_backup_template} {database_name_upgrade}_finish_{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}"
print(f"✨ {t('A backup can be created')} :\n{cmd_backup}")
status = input(
status = self.ask(
f"💬 {t('Press y/Y or type filename.zip to export, or')}"
f" {t('enter to continue')} : "
).strip()
@ -2660,7 +2718,7 @@ class TodoUpgrade:
cmd_backup = f"{cmd_backup_template} {status}"
self.todo_upgrade_execute(cmd_backup)
status = input(f"💬 {t('Test the migration, press y/Y')} : ")
status = self.ask(f"💬 {t('Test the migration, press y/Y')} : ")
if status.lower().strip() == "y":
self.todo.prompt_execute_selenium_and_run_db(database_name_upgrade)

225
test/test_auto_execute.py Executable file
View file

@ -0,0 +1,225 @@
#!/usr/bin/env python3
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
"""Enchaîner la migration sans rester devant à taper Entrée.
Une migration pose des dizaines de questions dont la réponse est presque
toujours celle proposée. Le mode auto attend cinq secondes puis prend le
défaut — assez pour reprendre la main, assez peu pour ne pas immobiliser
quelqu'un des heures.
La question est posée AVANT le choix de version, donc avant la première
décision : activée, elle vaut pour toutes les suivantes, celle-là comprise.
Le test qui compte le plus est le dernier : une invite écrite plus tard en
`input()` nu ne saurait rien du mode auto, et bloquerait la migration sans
que rien ne le signale.
"""
import ast
import inspect
import io
import os
import sys
import unittest
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
from script.todo import todo_i18n # noqa: E402
from script.todo.todo_upgrade import TodoUpgrade # noqa: E402
def upgrade(auto, delay=0.2):
obj = TodoUpgrade.__new__(TodoUpgrade)
obj.auto_execute = auto
obj.AUTO_DELAY = delay
return obj
class TestTheTimedRead(unittest.TestCase):
def setUp(self):
self.original = sys.stdin
self.addCleanup(setattr, sys, "stdin", self.original)
def pipe_stdin(self, payload=b""):
"""Un vrai descripteur : `select` ne sait pas lire un faux objet."""
read_fd, write_fd = os.pipe()
if payload:
os.write(write_fd, payload)
os.close(write_fd)
handle = os.fdopen(read_fd)
self.addCleanup(handle.close)
sys.stdin = handle
def test_without_auto_it_just_asks(self):
import builtins
original = builtins.input
builtins.input = lambda prompt="": "tapé"
self.addCleanup(setattr, builtins, "input", original)
self.assertEqual(upgrade(False).ask("q : ", default="6"), "tapé")
def test_with_auto_an_answer_wins(self):
self.pipe_stdin(b"2\n")
out = io.StringIO()
real = sys.stdout
sys.stdout = out
try:
got = upgrade(True).ask("q : ", default="6")
finally:
sys.stdout = real
self.assertEqual(got, "2")
def test_a_closed_stdin_gives_the_default(self):
# Exécution non interactive : le tuyau fermé est lisible TOUT DE
# SUITE et rend une ligne vide. Sans traiter le vide comme le
# défaut, le mode auto n'y servirait à rien.
self.pipe_stdin()
out = io.StringIO()
real = sys.stdout
sys.stdout = out
try:
got = upgrade(True).ask("q : ", default="6")
finally:
sys.stdout = real
self.assertEqual(got, "6")
def test_silence_gives_the_default_after_the_delay(self):
# Le tuyau reste OUVERT : rien à lire, et select doit rendre la main
# au bout du délai plutôt que d'attendre indéfiniment.
import time
read_fd, write_fd = os.pipe()
self.addCleanup(os.close, write_fd)
handle = os.fdopen(read_fd)
self.addCleanup(handle.close)
sys.stdin = handle
out = io.StringIO()
real = sys.stdout
sys.stdout = out
debut = time.time()
try:
got = upgrade(True, delay=0.3).ask("q : ", default="6")
finally:
sys.stdout = real
self.assertEqual(got, "6")
self.assertGreaterEqual(time.time() - debut, 0.25)
def test_an_unselectable_stdin_falls_back_to_asking(self):
# Ne pas deviner quand on ne peut pas mesurer : mieux vaut demander
# que rendre un défaut que personne n'a voulu.
import builtins
class NotSelectable:
def fileno(self):
raise ValueError("pas de descripteur")
sys.stdin = NotSelectable()
original = builtins.input
builtins.input = lambda prompt="": "demandé"
self.addCleanup(setattr, builtins, "input", original)
out = io.StringIO()
real = sys.stdout
sys.stdout = out
try:
got = upgrade(True).ask("q : ", default="6")
finally:
sys.stdout = real
self.assertEqual(got, "demandé")
class TestTheQuestionThatEnablesIt(unittest.TestCase):
def setUp(self):
self.addCleanup(
setattr, todo_i18n, "_current_lang", todo_i18n._current_lang
)
todo_i18n._current_lang = "en"
def answer(self, typed):
import builtins
import contextlib
obj = TodoUpgrade.__new__(TodoUpgrade)
original = builtins.input
builtins.input = lambda prompt="": typed
self.addCleanup(setattr, builtins, "input", original)
out = io.StringIO()
with contextlib.redirect_stdout(out):
obj.prompt_auto_execute()
return obj.auto_execute, out.getvalue()
def test_yes_turns_it_on(self):
on, text = self.answer("y")
self.assertTrue(on)
self.assertIn("Auto-run on", text)
def test_the_default_is_off(self):
# Elle prend des décisions à votre place : ce n'est pas le défaut.
on, text = self.answer("")
self.assertFalse(on)
self.assertEqual(text, "")
def test_anything_else_is_off(self):
self.assertFalse(self.answer("peut-être")[0])
def test_the_delay_is_five_seconds(self):
self.assertEqual(TodoUpgrade.AUTO_DELAY, 5)
class TestWhereItIsAsked(unittest.TestCase):
def source(self):
return inspect.getsource(TodoUpgrade.execute_odoo_upgrade)
def test_it_comes_before_the_version_choice(self):
# « avant de choisir la version d'Odoo » : activée, elle vaut aussi
# pour ce choix-là, qui a désormais un défaut.
source = self.source()
self.assertLess(
source.index("prompt_auto_execute"),
source.index("Which version do you want to upgrade to?"),
)
def test_the_version_choice_goes_through_it(self):
# click.prompt ne sait pas rendre la main après un délai : le laisser
# là aurait fait un mode auto qui s'arrête à la première question.
source = self.source()
self.assertNotIn("click.prompt(", source)
class TestNoPromptEscapesIt(unittest.TestCase):
"""Une invite en `input()` nu ne saurait rien du mode auto.
Elle bloquerait la migration sans rien signaler — et c'est précisément
le genre de chose qu'on n'ajoute pas exprès, mais par habitude.
"""
def test_every_prompt_of_the_migration_uses_ask(self):
path = os.path.join(REPO, "script", "todo", "todo_upgrade.py")
with open(path) as handle:
tree = ast.parse(handle.read())
target = [
node
for node in ast.walk(tree)
if isinstance(node, ast.FunctionDef)
and node.name == "execute_odoo_upgrade"
]
self.assertEqual(len(target), 1, "fonction introuvable")
bare = [
node.lineno
for node in ast.walk(target[0])
if isinstance(node, ast.Call)
and getattr(node.func, "id", "") == "input"
]
self.assertEqual(bare, [], "invites hors du mode auto")
def test_the_gate_uses_it_too(self):
# `ask_gate` porte le retour en arrière : la laisser en input() nu
# aurait fait un mode auto qui s'arrête à chaque « b = revenir ».
source = inspect.getsource(TodoUpgrade.ask_gate)
self.assertIn("self.ask(", source)
self.assertNotIn("= input(", source)
if __name__ == "__main__":
unittest.main()

View file

@ -29,9 +29,13 @@ from script.todo.todo_upgrade import TodoUpgrade # noqa: E402
def prompt_call(source):
"""Le voisinage de l'appel à click.prompt, où se joue le défaut."""
start = source.index("click.prompt(")
return source[start : start + 240]
"""Le voisinage de l'invite de version, où se joue le défaut.
Ce n'est plus click.prompt : lui ne sait pas rendre la main après un
délai, et le mode auto se serait arrêté à la première question.
"""
start = source.index("Which version do you want to upgrade to?")
return source[start : start + 1400]
class TestTheSource(unittest.TestCase):
@ -44,12 +48,18 @@ class TestTheSource(unittest.TestCase):
def source(self):
return inspect.getsource(TodoUpgrade.execute_odoo_upgrade)
def test_click_receives_a_default(self):
# Sans lui, Entrée n'atteint aucun code : click redemande, muet.
def test_the_prompt_receives_a_default(self):
window = prompt_call(self.source())
self.assertIn("default=", window)
self.assertIn("default_index", window)
def test_it_goes_through_the_auto_aware_reader(self):
# `self.ask` porte le mode auto ; click.prompt bloquerait pour
# toujours sur une invite que personne ne vient servir.
window = prompt_call(self.source())
self.assertIn("self.ask(", window)
self.assertNotIn("click.prompt(", window)
def test_the_default_is_not_written_by_hand(self):
window = prompt_call(self.source())
self.assertNotIn('default="6"', window)
@ -63,10 +73,11 @@ class TestTheSource(unittest.TestCase):
self.assertIn("range(len(lst_odoo_version))", source)
def test_an_empty_catalogue_shows_no_default(self):
# show_default sur une chaîne vide afficherait « [] », qui se lit
# comme un choix possible.
# Afficher « [] » se lirait comme un choix possible : la marque
# n'est posée que s'il y a vraiment un défaut.
self.assertIn(
"show_default=bool(default_index)", prompt_call(self.source())
'marque = f" [{default_index}]" if default_index else ""',
prompt_call(self.source()),
)
def test_the_question_announces_the_version_not_the_rank(self):
@ -75,10 +86,12 @@ class TestTheSource(unittest.TestCase):
self.assertIn("Enter =", source)
self.assertIn("default_version", source)
def test_no_special_case_survives(self):
# click rendant « 6 », le chemin normal suffit. Un cas particulier
# de plus serait un second comportement à maintenir.
self.assertNotIn("if not str(status).strip()", self.source())
def test_an_empty_answer_still_takes_the_default(self):
# click rendait « 6 » lui-même ; `self.ask` ne le fait qu'en mode
# AUTO. Hors mode auto, Entrée rend une chaîne vide, et sans ce cas
# le défaut ne servirait que la moitié du temps.
window = prompt_call(self.source())
self.assertIn("if not str(status).strip()", window)
class TestTheRankItself(unittest.TestCase):