diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index 3284524..70b1f72 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -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", diff --git a/script/todo/todo_upgrade.py b/script/todo/todo_upgrade.py index b1149d3..7aec353 100755 --- a/script/todo/todo_upgrade.py +++ b/script/todo/todo_upgrade.py @@ -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) diff --git a/test/test_auto_execute.py b/test/test_auto_execute.py new file mode 100755 index 0000000..13cdf17 --- /dev/null +++ b/test/test_auto_execute.py @@ -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() diff --git a/test/test_target_version_default.py b/test/test_target_version_default.py index 5be81d1..0abedc4 100755 --- a/test/test_target_version_default.py +++ b/test/test_target_version_default.py @@ -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):