diff --git a/script/todo/migration_form.py b/script/todo/migration_form.py index 98a150c..d320208 100644 --- a/script/todo/migration_form.py +++ b/script/todo/migration_form.py @@ -68,6 +68,7 @@ def run_resume_tui(ctx, run_app: bool = True): """ BINDINGS = [ ("c", "cont", t("Continue where it stopped")), + ("b", "back_step", t("Go back to a step")), ("n", "new", t("New migration, erase everything")), ("r", "keep_zip", t("Keep the zip only")), ("q", "quit_nothing", t("Quit without doing anything")), @@ -99,10 +100,12 @@ def run_resume_tui(ctx, run_app: bool = True): id="a_cont", ) yield Button(t("New migration"), id="a_new") + yield Button(t("Go back to a step"), id="a_back") yield Button(t("Keep the zip only"), id="a_keep") yield Button(t("Quit"), id="a_quit") yield Static( - f" {t('Enter on a step or a version = replay from there')}", + f" {t('Enter on a step or a version = replay from there')}" + f" · b = {t('Go back to a step')}", id="hint", ) yield Footer() @@ -156,6 +159,9 @@ def run_resume_tui(ctx, run_app: bool = True): self._answer(f"4.{event.option.id}") def on_button_pressed(self, event) -> None: + if event.button.id == "a_back": + self.action_back_step() + return mapping = { "a_cont": "c", "a_new": "n", @@ -166,6 +172,18 @@ def run_resume_tui(ctx, run_app: bool = True): if value: self._answer(value) + def action_back_step(self) -> None: + """Amener au tableau des étapes, où le choix se fait déjà. + + Le mécanisme existait — Entrée sur une ligne — mais vivait dans une + ligne d'astuce sous quatre boutons nommés. Une capacité qu'il faut + deviner n'est pas offerte ; celle-ci porte donc une touche, un + bouton et un message, comme les autres. + """ + table = self.query_one("#steps", DataTable) + table.focus() + self.notify(t("Choose a step, Enter replays from there.")) + def action_cont(self) -> None: self._answer("c") diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index 7f45853..f920d86 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -3920,6 +3920,14 @@ TRANSLATIONS = { "fr": "Relancez la migration pour reprendre à partir de là.", "en": "Relaunch the migration to resume from there.", }, + "Go back to a step": { + "fr": "Revenir à une étape", + "en": "Go back to a step", + }, + "Choose a step, Enter replays from there.": { + "fr": "Choisissez une étape, Entrée rejoue à partir de là.", + "en": "Choose a step, Enter replays from there.", + }, # --- script/analyse : vues personnalisées --- "Customised views": { "fr": "Vues personnalisées", diff --git a/script/todo/todo_upgrade.py b/script/todo/todo_upgrade.py index 1437fc0..b6bfbba 100755 --- a/script/todo/todo_upgrade.py +++ b/script/todo/todo_upgrade.py @@ -59,10 +59,11 @@ LOCAL_MANIFEST = os.path.join( # be dropped depends on the database, so that choice is never versioned. PATH_MIGRATION_GLOBAL = os.path.join("script", "odoo", "migration") PATH_MIGRATION_PRIVATE = os.path.join("private", "odoo", "migration") -# Steps of the migration, in order. Each one owns the progression keys prefixed -# with « state_ ». Rewinding to a step drops its keys and every later -# one, so the run replays from there. Labels go through t(): the key IS the -# English string, as everywhere else in this project. +# Steps of the migration, in order. What each one owns is declared just below, +# by GLOBAL_PROGRESSION_KEY and STEP_OWNED_KEY — not by the prefix alone, which +# lies on some keys and is missing on others. Rewinding to a step drops +# everything it and the later steps own, so the run really replays from there. +# Labels go through t(): the key IS the English string, as everywhere else. MIGRATION_STEP = [ (0, "Prepare the environment"), (1, "Restore and neutralize the database"), @@ -72,6 +73,71 @@ MIGRATION_STEP = [ ] +# Ce que le journal garde quoi qu'il arrive : des décisions et de la +# métadonnée, jamais du progrès. Rembobiner ne doit pas faire oublier QUELLE +# base on migre ni vers quelle version — on repartirait sur autre chose. +GLOBAL_PROGRESSION_KEY = frozenset( + { + "command_executed", + "config_database_name", + "config_migrate_repo", + "date_create", + "date_update", + "migration_file", + "target_odoo_version", + } +) + +# Clés que leur nom ne rattache à aucune étape, et l'étape qui les produit. +# +# Tout le reste s'auto-décrit : `state__` et `config_state__` nomment +# leur étape. Ces huit-là sont les sorties de la recherche de modules de +# l'étape 0 ; `dct_module_exist` est ensuite lu par l'étape 4. +# +# `state_1_update_all` est l'exception inverse : son nom dit 1, son travail +# est celui de l'étape 2 — c'est la mise à jour précoce, offerte avant la +# neutralisation quand la base vient d'une vieille version. +STEP_OWNED_KEY = { + 0: ( + "dct_module_exist", + "dct_module_per_version", + "len_dct_module_exist", + "len_lst_module_duplicate", + "len_lst_module_missing", + "lst_module_duplicate", + "lst_module_missing", + "lst_module_per_version_origin", + ), + 2: ("state_1_update_all",), +} + +# Réponses de l'étape 4 indexées par bump de version, une entrée par palier — +# exactement comme les drapeaux `state_4_*_odoo_lst`. Leur nom ne le dit pas, +# d'où cette liste : rejouer un palier sans les vider réutilisait en silence +# les modules choisis au passage précédent. +STEP_4_PER_BUMP_KEY = ( + "config_state_4_install_module", + "config_state_4_module_to_migrate_code", + "config_state_4_uninstall_module", +) + +STEP_PREFIX_RE = re.compile(r"^(?:config_)?state_(\d+)_") + + +def flag_step(key): + """Étape dont relève une clé du journal, ou None si elle est globale. + + Le préfixe ne suffit pas : une clé peut le porter à tort + (`state_1_update_all`) ou ne rien porter du tout (les listes de modules). + La table tranche d'abord, le nom ensuite. + """ + for step, lst_key in STEP_OWNED_KEY.items(): + if key in lst_key: + return step + match = STEP_PREFIX_RE.match(key) + return int(match.group(1)) if match else None + + class MigrationRewind(Exception): """L'utilisateur a demandé de revenir à une étape antérieure. @@ -666,18 +732,32 @@ class TodoUpgrade: the point — the intermediate database of a failed bump is half migrated, so it must be dropped and rebuilt from the previous version rather than upgraded again. + + The answers given for a bump — STEP_4_PER_BUMP_KEY, indexed exactly + like the flags — are trimmed too. They were not, so replaying a bump + reused the modules chosen the previous time and asked nothing. """ dct_kept = dict(old_dct_progression) for key, value in old_dct_progression.items(): - if ( + if not isinstance(value, list): + continue + per_bump = ( key.startswith("state_4_") - and isinstance(value, list) and key.endswith(("_odoo_lst", "_module")) - ): - dct_kept[key] = [ - item if i < index else False - for i, item in enumerate(value) - ] + ) or key in STEP_4_PER_BUMP_KEY + if not per_bump: + continue + # Vider en gardant le type : `config_state_4_module_to_migrate_code` + # porte des listes, sur lesquelles le code fait .append() sans les + # avoir revues. Un False à leur place plante le palier rejoué. + dct_kept[key] = [ + ( + item + if i < index + else ([] if isinstance(item, list) else False) + ) + for i, item in enumerate(value) + ] return dct_kept def ask_gate(self, prompt): @@ -723,18 +803,27 @@ class TodoUpgrade: @staticmethod def rewind_progression(old_dct_progression, step): - """Drop the progression of `step` and of every later step. + """Drop everything `step` and the later steps own, not just their flags. - Configuration answers (config_*), the zip and the target version are - kept: they are decisions, not progress. Only « state_* » is rewound. + Rewinding used to keep every key that was not « state_* ». The chosen + step replayed, then the ones after it ran on the previous course's + leftovers: the per-bump answers of step 4 — `config_state_4_*`, one + entry per version bump — were still there, so the questions were not + asked again and the work was taken for decided. + + So ownership is declared rather than guessed: GLOBAL_PROGRESSION_KEY + survives everything, STEP_OWNED_KEY and the « state__ » prefixes + say the rest. A key belonging to no one is progress by default and + goes: a replay that keeps an unknown leftover is the very defect + above, and `test_todo_upgrade_steps` fails on any unclassified key. """ dct_kept = {} for key, value in old_dct_progression.items(): - if not key.startswith("state_"): + if key in GLOBAL_PROGRESSION_KEY: dct_kept[key] = value continue - index = key[len("state_") :].split("_", 1)[0] - if index.isdigit() and int(index) < step: + index = flag_step(key) + if index is not None and index < step: dct_kept[key] = value # The module search fills an in-memory dict the later steps rely on; # it must run again even when step 0 itself is kept. diff --git a/test/test_migration_form_back.py b/test/test_migration_form_back.py new file mode 100644 index 0000000..0e9e919 --- /dev/null +++ b/test/test_migration_form_back.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""L'écran de reprise doit OFFRIR le retour à une étape, pas le cacher. + +Le mécanisme existait — Entrée sur une ligne du tableau — mais vivait dans une +ligne d'astuce sous quatre boutons nommés. Une capacité qu'il faut deviner +n'est pas offerte : elle a donc une touche, un bouton et un message, comme les +quatre autres actions. +""" + +import json +import unittest + +from script.todo import todo_i18n +from script.todo.migration_form import run_resume_tui +from script.todo.todo_upgrade import TodoUpgrade + + +def context(): + """Le contexte d'une migration arrêtée avant les montées de version.""" + progression = { + "migration_file": "./image_db/technolibre.zip", + "config_database_name": "technolibre_migration_01_neutralize", + "target_odoo_version": "18.0", + "date_create": "2026-08-12 05:04:22", + "state_0_install_odoo": True, + "state_1_restore_database": True, + "state_2_update_all": True, + "state_3_clean_database": True, + } + return TodoUpgrade.resume_context(TodoUpgrade, progression) + + +class TestBackIsOffered(unittest.TestCase): + def setUp(self): + self.addCleanup( + setattr, todo_i18n, "_current_lang", todo_i18n._current_lang + ) + todo_i18n._current_lang = "en" + self.ctx = context() + + def bindings(self, app): + out = {} + for item in app.BINDINGS: + if isinstance(item, tuple): + out[item[0]] = item[1] + else: + out[item.key] = item.action + return out + + def test_b_is_a_binding_like_the_others(self): + app = run_resume_tui(self.ctx, run_app=False) + self.assertEqual(self.bindings(app).get("b"), "back_step") + + def test_the_action_exists(self): + self.assertTrue( + hasattr( + run_resume_tui(self.ctx, run_app=False), "action_back_step" + ) + ) + + +class TestBackWorks(unittest.IsolatedAsyncioTestCase): + """Piloté dans un terminal simulé : la touche mène bien au choix.""" + + def setUp(self): + self.addCleanup( + setattr, todo_i18n, "_current_lang", todo_i18n._current_lang + ) + todo_i18n._current_lang = "en" + self.ctx = context() + + async def test_b_focuses_the_steps_and_enter_replays_from_there(self): + from textual.widgets import DataTable + + app = run_resume_tui(self.ctx, run_app=False) + async with app.run_test() as pilot: + await pilot.press("b") + await pilot.pause() + table = app.query_one("#steps", DataTable) + self.assertTrue(table.has_focus) + # Le curseur part sur la première étape inachevée ; on remonte de + # deux pour viser « mettre à jour tous les modules ». + await pilot.press("up") + await pilot.press("up") + await pilot.pause() + expected = str(self.ctx["steps"][table.cursor_row]["step"]) + await pilot.press("enter") + await pilot.pause() + self.assertEqual(app._result["answer"], expected) + + async def test_there_is_a_button_too(self): + # C'est l'absence VISIBLE qui a été signalée : une touche seule ne se + # voit pas parmi quatre boutons. Les widgets ne se lisent qu'une fois + # l'écran monté, d'où le pilote. + from textual.widgets import Button + + app = run_resume_tui(self.ctx, run_app=False) + async with app.run_test(): + labels = [b.label.plain for b in app.query(Button)] + self.assertIn("Go back to a step", labels) + + async def test_the_button_does_the_same(self): + from textual.widgets import Button, DataTable + + app = run_resume_tui(self.ctx, run_app=False) + async with app.run_test() as pilot: + button = next(b for b in app.query(Button) if b.id == "a_back") + await pilot.click(button) + await pilot.pause() + self.assertTrue(app.query_one("#steps", DataTable).has_focus) + # Et surtout : le bouton ne répond PAS à la place de l'utilisateur. + self.assertIsNone(app._result["answer"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_todo_upgrade_steps.py b/test/test_todo_upgrade_steps.py index 58db583..5f6efc2 100644 --- a/test/test_todo_upgrade_steps.py +++ b/test/test_todo_upgrade_steps.py @@ -100,6 +100,259 @@ class TestNeedsUpdateAll(unittest.TestCase): ) +class TestRewindReallyReplays(StatusCase): + """Rembobiner doit REJOUER l'étape, pas seulement effacer sa trace. + + Un journal venant d'une vieille base porte `state_1_update_all` : la mise + à jour précoce, offerte avant la neutralisation. Le travail est celui de + l'étape 2, seul le nom dit 1. Rembobiner à l'étape 2 gardait donc ce + drapeau — son préfixe le rangeait dans l'étape 1 — et l'étape sautait + alors qu'on venait de demander à la rejouer. + + Effacer les clés ne suffit pas à le prouver : il faut interroger les + gardes qui décident du travail. + """ + + def journal(self): + """La forme d'un vrai journal ayant fait la mise à jour précoce.""" + return { + "config_database_name": "db", + "state_0_install_odoo": True, + "state_0_search_missing_module": True, + "state_1_restore_database": True, + "state_1_update_all": True, + "state_1_neutralize_database": True, + "state_3_clean_database": True, + "state_4_upgrade_odoo_lst": [True, False], + } + + def rewind(self, step): + import contextlib + import io + + with contextlib.redirect_stdout(io.StringIO()): + return TodoUpgrade.rewind_progression(self.journal(), step) + + def test_going_back_to_step_2_runs_the_update_again(self): + # LE défaut signalé : « il continue en ignorant l'étape que j'ai + # choisie ». Trois heures de mise à jour silencieusement sautées. + self.assertTrue( + TodoUpgrade.needs_update_all(self.rewind(2)), + "rembobiner à l'étape 2 doit rejouer la mise à jour", + ) + + def test_the_earlier_steps_replay_it_too(self): + for step in (0, 1): + self.assertTrue( + TodoUpgrade.needs_update_all(self.rewind(step)), + f"étape {step}", + ) + + def test_going_back_to_a_later_step_leaves_it_done(self): + # Le symétrique : reculer à l'étape 3 ne doit PAS relancer l'étape 2. + for step in (3, 4): + self.assertFalse( + TodoUpgrade.needs_update_all(self.rewind(step)), + f"étape {step}", + ) + + def test_no_work_guard_of_the_chosen_step_or_later_survives(self): + # Les gardes sont lues dans la SOURCE : une garde ajoutée plus tard + # sera couverte sans que ce test soit retouché. + import ast + + from script.todo import todo_upgrade + + guards = set() + with open(todo_upgrade.__file__) as handle: + tree = ast.parse(handle.read()) + for node in ast.walk(tree): + if not ( + isinstance(node, ast.If) and isinstance(node.test, ast.UnaryOp) + ): + continue + for sub in ast.walk(node.test): + if ( + isinstance(sub, ast.Call) + and isinstance(sub.func, ast.Attribute) + and sub.func.attr == "get" + and sub.args + and isinstance(sub.args[0], ast.Constant) + and str(sub.args[0].value).startswith("state_") + ): + guards.add(str(sub.args[0].value)) + self.assertTrue( + guards, "aucune garde trouvée : le test ne prouve rien" + ) + for step, _ in MIGRATION_STEP: + kept = self.rewind(step) + still_closed = [ + key + for key in sorted(guards) + if todo_upgrade.flag_step(key) >= step and kept.get(key) + ] + self.assertEqual(still_closed, [], f"étape {step}") + + +class TestEveryKeyBelongsToSomeone(unittest.TestCase): + """Aucune clé du journal ne doit échapper au classement. + + Rembobiner ne gardait que ce qui ne commençait pas par « state_ ». Vingt- + cinq clés passaient donc au travers, dont les réponses par palier de + l'étape 4 : l'étape choisie se rejouait, les suivantes repartaient sur les + décisions de la course précédente et ne demandaient plus rien. + + Ce test lit les clés dans la SOURCE. Une clé ajoutée demain sans être + classée le fera échouer, au lieu de survivre en silence au rembobinage. + """ + + def keys_in_source(self): + import ast + + from script.todo import todo_upgrade + + with open(todo_upgrade.__file__) as handle: + tree = ast.parse(handle.read()) + keys = set() + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + for target in node.targets: + if ( + isinstance(target, ast.Subscript) + and isinstance(target.value, ast.Attribute) + and target.value.attr == "dct_progression" + and isinstance(target.slice, ast.Constant) + ): + keys.add(str(target.slice.value)) + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "get" + and node.args + and isinstance(node.args[0], ast.Constant) + ): + owner = node.func.value + name = getattr(owner, "attr", getattr(owner, "id", "")) + if "progression" in str(name): + keys.add(str(node.args[0].value)) + return keys + + def test_no_key_is_left_unclassified(self): + from script.todo import todo_upgrade + + keys = self.keys_in_source() + self.assertGreater(len(keys), 30, "l'extraction n'a rien trouvé") + orphans = sorted( + key + for key in keys + if key not in todo_upgrade.GLOBAL_PROGRESSION_KEY + and todo_upgrade.flag_step(key) is None + ) + self.assertEqual( + orphans, [], "à classer : global, ou à quelle étape ?" + ) + + def test_a_key_is_never_both_global_and_owned(self): + from script.todo import todo_upgrade + + for key in sorted(todo_upgrade.GLOBAL_PROGRESSION_KEY): + self.assertIsNone(todo_upgrade.flag_step(key), key) + + +class TestRewindDropsTheAnswersToo(StatusCase): + """« Il faut aussi revert les variables mises à jour entre les étapes. »""" + + def journal(self): + return { + "config_database_name": "db", + "target_odoo_version": "18.0", + "migration_file": "a.zip", + "config_state_1_install_module": ["sale"], + "config_state_4_install_module": [False, ["stock"]], + "config_state_4_uninstall_module": [False, False], + "config_state_4_module_to_migrate_code": [[], ["m"]], + "dct_module_exist": {"sale": "1.0"}, + "state_0_install_odoo": True, + "state_2_update_all": True, + } + + def rewind(self, step): + import contextlib + import io + + with contextlib.redirect_stdout(io.StringIO()): + return TodoUpgrade.rewind_progression(self.journal(), step) + + def test_the_step_4_answers_go_when_step_4_is_replayed(self): + # LE défaut signalé : l'étape rejouée marchait, les suivantes non. + kept = self.rewind(2) + for key in ( + "config_state_4_install_module", + "config_state_4_uninstall_module", + "config_state_4_module_to_migrate_code", + ): + self.assertNotIn(key, kept, key) + + def test_the_step_1_answers_stay_when_step_1_is_kept(self): + # Le symétrique : rembobiner à 2 ne doit pas re-demander l'étape 1. + self.assertIn("config_state_1_install_module", self.rewind(2)) + self.assertNotIn("config_state_1_install_module", self.rewind(1)) + + def test_the_step_0_module_search_output_goes_with_step_0(self): + # `dct_module_exist` ne nomme aucune étape mais est produit par la 0, + # et relu par la 4 : le garder après un rembobinage à 0 ferait juger + # la 4 sur un inventaire périmé. + self.assertNotIn("dct_module_exist", self.rewind(0)) + self.assertIn("dct_module_exist", self.rewind(1)) + + def test_the_decisions_survive_every_rewind(self): + # Sans elles on ne saurait plus QUELLE base on migre : le rembobinage + # deviendrait une autre migration. + for step, _ in MIGRATION_STEP: + kept = self.rewind(step) + for key in ("config_database_name", "target_odoo_version"): + self.assertIn(key, kept, f"{key} à l'étape {step}") + + +class TestRewindOneVersionBump(unittest.TestCase): + """Rejouer un palier de l'étape 4 doit aussi rendre ses réponses.""" + + def journal(self): + return { + "state_4_clone_odoo_lst": [True, True, True], + "state_4_uninstall_module": [True, True, False], + "config_state_4_install_module": [["sale"], ["stock"], False], + "config_state_4_module_to_migrate_code": [["m1"], ["m2"], []], + "state_4_lst_module_missing": ["a", "b"], + "state_4_len_lst_module_missing": 40, + } + + def test_the_answers_of_the_replayed_bumps_are_cleared(self): + kept = TodoUpgrade.rewind_version_bump(self.journal(), 1) + self.assertEqual( + kept["config_state_4_install_module"][1:], [[], False] + ) + self.assertEqual(kept["config_state_4_install_module"][0], ["sale"]) + + def test_a_list_entry_stays_a_list(self): + # Le code fait .append() sur cette entrée sans l'avoir revue : un + # False à sa place ferait planter le palier qu'on vient de rejouer. + kept = TodoUpgrade.rewind_version_bump(self.journal(), 1) + for entry in kept["config_state_4_module_to_migrate_code"]: + self.assertIsInstance(entry, list) + + def test_a_flat_list_is_not_a_per_bump_list(self): + # `state_4_lst_module_missing` est une liste de modules, pas un + # tableau par palier : la tronquer perdrait l'inventaire. + kept = TodoUpgrade.rewind_version_bump(self.journal(), 1) + self.assertEqual(kept["state_4_lst_module_missing"], ["a", "b"]) + self.assertEqual(kept["state_4_len_lst_module_missing"], 40) + + def test_the_earlier_bumps_are_untouched(self): + kept = TodoUpgrade.rewind_version_bump(self.journal(), 2) + self.assertEqual(kept["state_4_clone_odoo_lst"], [True, True, False]) + + class TestBackGate(unittest.TestCase): """Revenir à une étape précédente depuis une invite en cours de route.