From 6c186f71e91043d0fbb99eab8e9ac3ec76db9579 Mon Sep 17 00:00:00 2001 From: Mathieu Benoit Date: Wed, 19 Aug 2026 03:53:56 -0400 Subject: [PATCH] =?UTF-8?q?[ADD]=20migration:=20=C2=AB=20t=20=C2=BB=20show?= =?UTF-8?q?s=20where=20the=20migration=20stands?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A migration crosses six bumps, runs hundreds of commands and lasts hours. The journal said what had been LAUNCHED; it never said what came of it. Three hours in, one reads two hundred command lines without knowing which one failed, nor what the smoke test concluded. « v » could not carry this: it already means « view the differences » in three prompts, and a letter meaning two things is worse than a letter meaning nothing. « t » was free, and it is the same everywhere — one shared string carries both shortcuts, so no prompt can drift. Looking is not answering: the same question comes back afterwards. What was missing was the data. Failures and tool verdicts are now recorded with the step they happened in, bounded so a progression file cannot grow without end. A tool rerun after a repair keeps its LAST verdict and the count of its runs — showing both without distinction would read a repair as a lasting failure. --- FR --- [ADD] migration : « t » montre où en est la migration Une migration traverse six paliers, lance des centaines de commandes et dure des heures. Le journal disait ce qui avait été LANCÉ, jamais ce que cela avait donné. Trois heures plus tard on relit deux cents lignes sans savoir laquelle a échoué, ni ce que le test de fumée a conclu. « v » ne pouvait pas porter cela : il veut déjà dire « voir les différences » dans trois invites, et une lettre qui signifie deux choses est pire qu'une lettre qui ne signifie rien. « t » était libre, et il est le même partout — une seule chaîne porte les deux raccourcis. Regarder n'est pas répondre : la même question revient ensuite. Ce qui manquait, c'étaient les données. Les échecs et les verdicts d'outils sont désormais retenus avec l'étape où ils se sont produits. Assisted-by: Claude Opus 5 --- script/todo/migration_status.py | 227 +++++++++++++++++ script/todo/migration_status_tui.py | 200 +++++++++++++++ script/todo/todo_i18n.py | 73 ++++++ script/todo/todo_upgrade.py | 103 +++++++- test/test_database_cleanup.py | 9 +- test/test_error_retry_loop.py | 4 + test/test_migration_status.py | 376 ++++++++++++++++++++++++++++ test/test_smoke_internal_ui.py | 2 +- 8 files changed, 980 insertions(+), 14 deletions(-) create mode 100755 script/todo/migration_status.py create mode 100644 script/todo/migration_status_tui.py create mode 100644 test/test_migration_status.py diff --git a/script/todo/migration_status.py b/script/todo/migration_status.py new file mode 100755 index 0000000..6cd506d --- /dev/null +++ b/script/todo/migration_status.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""Où en est cette migration, et qu'est-ce qui a mal tourné. + +Une migration traverse six paliers, lance des centaines de commandes et +dure des heures. Le journal existant dit ce qui a été LANCÉ ; il ne dit +jamais ce que cela a donné. Après trois heures on relit deux cents lignes +de commandes sans savoir laquelle a échoué, ni ce que le test de fumée a +conclu, ni à quelle étape on se trouve. + +Ce module assemble la réponse à partir du fichier de progression — le +même que la migration écrit après chaque geste — et la rend en texte. Le +plein écran, lui, n'est qu'une autre vue de CES données : deux rendus +séparés dériveraient l'un de l'autre sans que rien ne le signale. + +Rien n'est lu ailleurs que dans ce fichier : l'écran d'état ne doit jamais +toucher une base ni lancer un serveur. On l'ouvre en pleine migration. +""" + +import json +import os +import sys + +sys.path.append( + os.path.normpath(os.path.join(os.path.dirname(__file__), "..", "..")) +) + +try: + from script.todo.todo_i18n import t +except Exception: # pragma: no cover - repli si i18n indisponible + + def t(key: str) -> str: + return key + + +DEFAULT_PATH = ".venv.erplibre/odoo_database_migration_log.json" + +# Les codes de sortie que TOUS les outils de migration partagent. Les +# traduire ici plutôt qu'à l'affichage évite qu'un « 1 » passe pour une +# panne alors qu'il annonce des trouvailles. +VERDICT = { + 0: ("✅", "nothing to report"), + 1: ("⚠️", "findings to look at"), + 2: ("❌", "the tool itself failed"), +} + + +def read(path=DEFAULT_PATH): + """La progression, telle qu'écrite. {} si elle n'existe pas encore.""" + try: + with open(path, "r", encoding="utf-8") as handle: + return json.load(handle) + except (OSError, ValueError): + return {} + + +def journal_by_step(dct): + """Le journal découpé par étape, dans l'ordre. + + Les entrées commençant par « # » sont les en-têtes d'étape que la + migration y dépose elle-même ; tout ce qui suit appartient à l'étape + ouverte. On réutilise donc un marquage qui existe déjà plutôt que d'en + inventer un second, qui divergerait. + """ + lst_section = [] + courante = None + for entry in dct.get("command_executed") or []: + texte = str(entry) + if texte.startswith("#"): + courante = {"step": texte.lstrip("# ").strip(), "lst_cmd": []} + lst_section.append(courante) + continue + if courante is None: + courante = {"step": t("before the first step"), "lst_cmd": []} + lst_section.append(courante) + courante["lst_cmd"].append(texte) + return lst_section + + +def events(dct, kind=None): + """Ce qui a mal tourné, et ce que les outils ont conclu.""" + lst = [ + item + for item in (dct.get("lst_event") or []) + if kind is None or item.get("kind") == kind + ] + return lst + + +def verdict(status): + """(icône, phrase) pour un code de sortie d'outil.""" + icone, phrase = VERDICT.get(status, ("❔", "unknown result")) + return icone, t(phrase) + + +def tests_summary(dct): + """Le DERNIER verdict de chaque outil, et le compte des passages. + + Un outil relancé après correction a deux verdicts contradictoires dans + le journal, et c'est le dernier qui décrit la base telle qu'elle est. + Afficher les deux sans les distinguer ferait lire une réparation comme + un échec persistant. + """ + dernier = {} + for item in events(dct, kind="test"): + nom = item.get("name") or "?" + entree = dernier.setdefault(nom, {"name": nom, "runs": 0}) + entree["runs"] += 1 + entree["status"] = item.get("status") + entree["at"] = item.get("at") + entree["step"] = item.get("step") + return [dernier[nom] for nom in sorted(dernier)] + + +def failures(dct): + """Les commandes qui ont échoué, la plus récente d'abord.""" + return list(reversed(events(dct, kind="command"))) + + +def overview(dct): + """L'en-tête : de quelle migration parle-t-on.""" + return { + "file": os.path.basename(dct.get("migration_file") or "?"), + "database": dct.get("config_database_name") or "?", + "target": dct.get("target_odoo_version") or "?", + "started": dct.get("date_create") or "?", + "updated": dct.get("date_update") or "?", + "step": ( + (dct.get("lst_event") or [{}])[-1].get("step") + or _last_step(dct) + or "?" + ), + } + + +def _last_step(dct): + lst = journal_by_step(dct) + return lst[-1]["step"] if lst else None + + +def bumps(dct): + """Les paliers, et lesquels sont faits.""" + done = dct.get("state_4_upgrade_odoo_lst") or [] + lst = dct.get("lst_version_bump") or [] + if not lst: + # Le nom des bases de palier porte la version : c'est la seule + # source disponible quand la liste n'a pas été écrite. + lst = list(range(len(done))) + return [ + {"version": version, "done": bool(i < len(done) and done[i])} + for i, version in enumerate(lst) + ] + + +def render_text(dct, limit_cmd=12): + """Le rapport complet, en texte. C'est aussi le repli du plein écran.""" + if not dct: + return f"ℹ️ {t('No migration in progress.')}" + info = overview(dct) + lignes = [ + f"📍 {t('Migration state')}", + f" {t('database')} : {info['database']}", + f" {t('image')} : {info['file']}", + f" {t('started')} : {info['started']}", + f" {t('last written')} : {info['updated']}", + f" {t('current step')} : {info['step']}", + ] + + lst_test = tests_summary(dct) + lignes.append(f"\n🧪 {t('Test results')}") + if not lst_test: + lignes.append(f" {t('No tool has run yet.')}") + for item in lst_test: + icone, phrase = verdict(item.get("status")) + rejeu = ( + f" ({item['runs']} {t('runs')})" + if item.get("runs", 1) > 1 + else "" + ) + lignes.append(f" {icone} {item['name']:<24} {phrase}{rejeu}") + + lst_failure = failures(dct) + lignes.append(f"\n❌ {t('Commands that failed')} : {len(lst_failure)}") + for item in lst_failure[:10]: + lignes.append(f" · [{item.get('step') or '?'}] {item.get('name')}") + + lignes.append(f"\n🔷 {t('What was done, step by step')}") + for section in journal_by_step(dct): + lst_cmd = section["lst_cmd"] + lignes.append(f" {section['step']} ({len(lst_cmd)})") + for cmd in lst_cmd[:limit_cmd]: + lignes.append(f" {cmd[:120]}") + if len(lst_cmd) > limit_cmd: + reste = len(lst_cmd) - limit_cmd + lignes.append(f" … {reste} {t('more')}") + return "\n".join(lignes) + + +def main(argv=None): + import argparse + + parser = argparse.ArgumentParser( + description="Show what this migration has done, and what failed." + ) + parser.add_argument("-f", "--file", default=DEFAULT_PATH) + parser.add_argument( + "--text", + action="store_true", + help="print the report instead of opening the full screen", + ) + config = parser.parse_args(argv) + dct = read(config.file) + if not config.text: + try: + from script.todo.migration_status_tui import run_tui + except Exception: + run_tui = None + if run_tui and run_tui(dct): + return 0 + print(render_text(dct)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/script/todo/migration_status_tui.py b/script/todo/migration_status_tui.py new file mode 100644 index 0000000..0777a78 --- /dev/null +++ b/script/todo/migration_status_tui.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""L'état d'une migration, en plein écran. + +Le rapport texte dit tout, mais il dit tout D'UN COUP : sur une migration +de six paliers il fait plusieurs centaines de lignes, et ce qu'on cherche — +l'étape où ça a cassé, ce que le test de fumée a conclu — se trouve +quelque part au milieu. Un écran qui se parcourt règle exactement cela. + +Les données viennent de `migration_status`, comme le rapport texte. Deux +assemblages séparés dériveraient l'un de l'autre sans que rien ne le dise, +et l'on finirait par lire deux états contradictoires de la même migration. +""" + +import os +import sys + +sys.path.append( + os.path.normpath(os.path.join(os.path.dirname(__file__), "..", "..")) +) + +from script.todo import migration_status as status # noqa: E402 + +try: + from script.todo.todo_i18n import t +except Exception: # pragma: no cover - repli si i18n indisponible + + def t(key: str) -> str: + return key + + +CSS = """ +Screen { layout: vertical; } +#head { height: 4; padding: 0 1; background: $panel; color: $text; } +#body { height: 1fr; } +#left { width: 46; border-right: solid $accent; } +#pane { width: 1fr; padding: 0 1; } +""" + + +def head_text(dct): + """Ce qui identifie la migration. Toujours visible, jamais à chercher.""" + info = status.overview(dct) + lst_test = status.tests_summary(dct) + casse = [x for x in lst_test if x.get("status")] + return ( + f"{info['database']} · {info['file']}\n" + f"{t('current step')} : {info['step']}\n" + f"{t('last written')} : {info['updated']}" + f" · {len(status.failures(dct))} {t('failed')}" + f" · {len(casse)}/{len(lst_test)} {t('Test results')}" + ) + + +def rows(dct): + """Les lignes du panneau de gauche : tests d'abord, puis les étapes. + + Les tests en tête parce que c'est la question qu'on se pose en ouvrant + cet écran ; les étapes ensuite parce que c'est là qu'on cherche le + détail une fois qu'on sait QUOI chercher. + """ + lst = [] + for item in status.tests_summary(dct): + icone, _phrase = status.verdict(item.get("status")) + lst.append( + { + "kind": "test", + "label": f"{icone} {item['name']}", + "detail": str(item.get("runs", 1)), + "data": item, + } + ) + for section in status.journal_by_step(dct): + lst.append( + { + "kind": "step", + "label": section["step"], + "detail": str(len(section["lst_cmd"])), + "data": section, + } + ) + return lst + + +def pane_text(dct, row): + """Le détail de la ligne choisie.""" + if row is None: + return t("Nothing to show yet.") + if row["kind"] == "test": + item = row["data"] + icone, phrase = status.verdict(item.get("status")) + lignes = [ + f"{icone} {item['name']}", + f" {phrase} ({t('exit code')} {item.get('status')})", + f" {t('runs')} : {item.get('runs')}", + f" {t('current step')} : {item.get('step') or '?'}", + f" {item.get('at') or ''}", + ] + return "\n".join(lignes) + section = row["data"] + lignes = [f"{section['step']}", ""] + lst_failure = [ + item + for item in status.failures(dct) + if (item.get("step") or "") == section["step"] + ] + if lst_failure: + lignes.append(f"❌ {t('Commands that failed')} :") + for item in lst_failure: + lignes.append(f" {item.get('name')}") + lignes.append("") + if not section["lst_cmd"]: + lignes.append(t("No tool has run yet.")) + for cmd in section["lst_cmd"]: + lignes.append(f"· {cmd}") + return "\n".join(lignes) + + +def build_app(dct): + """Textual est importé ICI, pas au chargement du module. + + Le module reste importable — donc testable — sur une machine sans + Textual, et c'est aussi ce qui permet à l'appelant de retomber sur le + rapport texte plutôt que d'échouer. + """ + from textual.app import App, ComposeResult + from textual.containers import Horizontal, VerticalScroll + from textual.widgets import DataTable, Footer, Header, Static + + class StatusApp(App): + CSS = globals()["CSS"] + BINDINGS = [("q,escape", "quit", "Quit")] + + def __init__(self, dct): + super().__init__() + self.dct = dct + self.lst_row = rows(dct) + self.index = 0 + + def compose(self) -> ComposeResult: + yield Header() + yield Static("", id="head") + with Horizontal(id="body"): + yield DataTable(id="left", cursor_type="row") + with VerticalScroll(id="pane"): + yield Static("", id="content") + yield Footer() + + def on_mount(self): + self.title = t("Migration state") + table = self.query_one("#left", DataTable) + table.add_columns(t("Test results"), "#") + for row in self.lst_row: + table.add_row(row["label"][:38], row["detail"]) + self.query_one("#head", Static).update(head_text(self.dct)) + self._show() + + def _show(self): + row = self.lst_row[self.index] if self.lst_row else None + self.query_one("#content", Static).update(pane_text(self.dct, row)) + + def on_data_table_row_highlighted(self, event): + if event.data_table.id == "left" and self.lst_row: + self.index = event.cursor_row + self._show() + + return StatusApp(dct) + + +def run_tui(dct, run_app=True): + """Ouvrir l'écran. False si l'on n'a pas pu — et alors on DIT pourquoi. + + Se taire ferait réafficher le rapport texte à la place de l'écran + demandé, sans rien qui distingue les deux. + """ + if not dct: + return False + if not sys.stdout.isatty(): + print(f"ℹ️ {t('Not a terminal: showing the text report instead.')}") + return False + try: + from script.todo import textual_setup + except Exception: + textual_setup = None + if textual_setup and not textual_setup.ensure(): + return False + try: + app = build_app(dct) + except ImportError: + print( + f"ℹ️ {t('Textual is missing from this interpreter:')}" + f" {sys.executable}" + ) + return False + if not run_app: + return app + app.run() + return True diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index c6abf99..934ff08 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -5530,6 +5530,79 @@ TRANSLATIONS = { "fr": "s'y connecter.", "en": "sign in with.", }, + "Migration state": { + "fr": "État de la migration", + "en": "Migration state", + }, + "started": { + "fr": "démarrée", + "en": "started", + }, + "last written": { + "fr": "dernière écriture", + "en": "last written", + }, + "current step": { + "fr": "étape courante", + "en": "current step", + }, + "Test results": { + "fr": "Résultat des tests", + "en": "Test results", + }, + "No tool has run yet.": { + "fr": "Aucun outil n'a encore tourné.", + "en": "No tool has run yet.", + }, + "Commands that failed": { + "fr": "Commandes en échec", + "en": "Commands that failed", + }, + "What was done, step by step": { + "fr": "Ce qui a été fait, étape par étape", + "en": "What was done, step by step", + }, + "runs": { + "fr": "passages", + "en": "runs", + }, + "nothing to report": { + "fr": "rien à signaler", + "en": "nothing to report", + }, + "findings to look at": { + "fr": "des trouvailles à regarder", + "en": "findings to look at", + }, + "the tool itself failed": { + "fr": "l'outil lui-même a échoué", + "en": "the tool itself failed", + }, + "unknown result": { + "fr": "résultat inconnu", + "en": "unknown result", + }, + "before the first step": { + "fr": "avant la première étape", + "en": "before the first step", + }, + "No migration in progress.": { + "fr": "Aucune migration en cours.", + "en": "No migration in progress.", + }, + "show the migration state": { + "fr": "afficher l'état de la migration", + "en": "show the migration state", + }, + "Nothing to show yet.": { + "fr": "Rien à montrer pour l'instant.", + "en": "Nothing to show yet.", + }, + "(b = go back a step, t = show the migration state)": { + "fr": "(b = revenir à une étape précédente," + " t = afficher l'état de la migration)", + "en": "(b = go back a step, t = show the migration state)", + }, "Clean the database before testing the pages?": { "fr": "Nettoyer la base avant de tester les pages ?", "en": "Clean the database before testing the pages?", diff --git a/script/todo/todo_upgrade.py b/script/todo/todo_upgrade.py index a08e0d4..ee71f70 100755 --- a/script/todo/todo_upgrade.py +++ b/script/todo/todo_upgrade.py @@ -79,6 +79,10 @@ MIGRATION_STEP = [ GLOBAL_PROGRESSION_KEY = frozenset( { "command_executed", + # Le registre des échecs et des verdicts d'outils, comme le journal + # des commandes : un retour en arrière ne doit PAS l'effacer. C'est + # justement ce qu'on vient consulter après être revenu. + "lst_event", "config_database_name", "config_migrate_repo", "date_create", @@ -820,13 +824,37 @@ class TodoUpgrade: """ while True: answer = self.ask(prompt, default=default) - if (answer or "").strip().lower() != "b": + reponse = (answer or "").strip().lower() + if reponse == "t": + # Regarder n'est pas répondre : on repose la MÊME question + # après. Sans cela, ouvrir l'état vaudrait « continuer », ce + # qui est exactement ce qu'on ne voulait pas faire. + self.show_migration_status() + continue + if reponse != "b": return answer if self.rewind_to_chosen_step(): raise MigrationRewind() # Renoncer au retour en arrière ne doit pas arrêter la migration : # on revient à la même invite, exactement là où l'on était. + def show_migration_status(self): + """Ouvrir l'état de la migration, en plein écran. + + On ÉCRIT avant de lire : l'écran lit le fichier de progression, et + ce qui vient de se passer n'y serait pas encore. + + Pas par `run_on_terminal` : celui-ci consigne ce qu'il lance dans + le journal, et le journal est justement ce que cet écran montre. + Regarder l'état polluerait alors l'état. + """ + self.write_config() + subprocess.call( + f"{PYTHON_BIN} ./script/todo/migration_status.py", + shell=True, + executable="/bin/bash", + ) + def rewind_to_chosen_step(self): """Demander l'étape et rembobiner jusqu'à elle. Écrit la progression.""" ctx = self.resume_context(self.dct_progression) @@ -1532,7 +1560,7 @@ class TodoUpgrade: self.ask_gate( "💬 " + t("Neutralize database, press to continue") - + f" {t('(b = go back to a previous step)')} : " + + f" {t('(b = go back a step, t = show the migration state)')} : " ) .strip() .lower() @@ -2527,7 +2555,7 @@ class TodoUpgrade: ) status = self.ask_gate( f"💬 {t('Press to continue')} {msg}" - f" {t('(b = go back to a previous step)')} : " + f" {t('(b = go back a step, t = show the migration state)')} : " ).strip() # The technique change at version 14 if next_version <= 13: @@ -2860,6 +2888,9 @@ class TodoUpgrade: une phrase. """ prefix, sep, label = msg.partition(" - ") + # Retenu pour l'écran d'état : un événement sans étape oblige à + # relire tout le journal pour savoir OÙ il s'est produit. + self.current_step = msg print(f"🔷 {prefix}{sep}{t(label)}" if sep else f"🔷 {t(msg)}") def installed_theme(self, database_name): @@ -2903,7 +2934,7 @@ class TodoUpgrade: answer = ( self.ask_gate( f"💬 {t('Uninstall them properly before migrating?')}" - f" (Y/n, {t('(b = go back to a previous step)')}) : ", + f" (Y/n, {t('(b = go back a step, t = show the migration state)')}) : ", default="y", ) .strip() @@ -2989,9 +3020,10 @@ class TodoUpgrade: print(f"⚠️ {t('Unknown choice, nothing was reset.')}") return False args = " ".join(f"--reset {key}" for key in lst_chosen) - status = self.run_on_terminal( + status = self.run_tool( + "reset_stale_cow_views", f"{PYTHON_BIN} ./{os.path.join(PATH_MIGRATION_GLOBAL, 'reset_stale_cow_views.py')}" - f" -d {database_name} {args} --apply" + f" -d {database_name} {args} --apply", ) return status == 0 @@ -3009,7 +3041,7 @@ class TodoUpgrade: answer = ( self.ask_gate( f"💬 {t('Clean the database before testing the pages?')}" - f" (Y/n, {t('(b = go back to a previous step)')}) : ", + f" (Y/n, {t('(b = go back a step, t = show the migration state)')}) : ", default="y", ) .strip() @@ -3017,10 +3049,11 @@ class TodoUpgrade: ) if answer != "y": return - self.run_on_terminal( + self.run_tool( + "database_cleanup", f"{PYTHON_BIN}" " ./script/odoo/migration/database_cleanup.py" - f" -d {database_name}" + f" -d {database_name}", ) def prompt_smoke_public_url(self, database_name, baseline=False): @@ -3050,7 +3083,7 @@ class TodoUpgrade: answer = ( self.ask_gate( f"💬 {t('Request every public URL of this database now?')}" - f" (Y/n, {t('(b = go back to a previous step)')}) : ", + f" (Y/n, {t('(b = go back a step, t = show the migration state)')}) : ", default="y", ) .strip() @@ -3074,11 +3107,12 @@ class TodoUpgrade: f" {t('was not neutralized, so there is no test user to')}" f" {t('sign in with.')}" ) - self.run_on_terminal( + self.run_tool( + "smoke_public_url", f"{PYTHON_BIN}" " ./script/odoo/migration/smoke_public_url.py" f" -d {database_name}" - + (" --internal-required" if neutralise else "") + + (" --internal-required" if neutralise else ""), ) def show_cow_drift(self, database_name, next_version, mode="diff"): @@ -3542,6 +3576,10 @@ class TodoUpgrade: # failure, never as a success (defence in depth: exec_command_live now # always sets one, but a silent None must not skip this prompt). if (status is None or status) and wait_at_error: + # AVANT l'invite : si l'on répond ctrl+c, l'échec doit tout de + # même figurer dans l'état — c'est précisément celui qu'on + # cherchera en revenant. + self.record_event("command", cmd, status if status else 1) database_name = self.database_from_command(cmd) # « 3 » par défaut, car le motif d'échec le plus fréquent ici est # une copie COW en retard : la réparer est presque toujours ce @@ -3815,6 +3853,47 @@ class TodoUpgrade: return parts[1], remote, remote_branch return None, None, None + MAX_EVENT = 200 + + def record_event(self, kind, name, status, detail=""): + """Garder ce qui s'est MAL passé, et ce que les outils ont conclu. + + Le journal existant (`command_executed`) dit ce qui a été LANCÉ, + jamais ce que cela a donné. Après six paliers on lit deux cents + lignes de commandes sans savoir laquelle a échoué, ni ce que le + test de fumée a trouvé. C'est cette question-là que l'écran d'état + doit pouvoir répondre. + + Borné : une migration lance des centaines de commandes, et un + fichier de progression qui enfle sans limite finit par coûter plus + cher à écrire qu'à lire. + """ + lst = list(self.dct_progression.get("lst_event") or []) + lst.append( + { + "at": str(datetime.datetime.now()), + "step": getattr(self, "current_step", "") or "", + "kind": kind, + "name": name, + "status": status, + "detail": str(detail)[:500], + } + ) + self.dct_progression["lst_event"] = lst[-self.MAX_EVENT :] + self.write_config() + + def run_tool(self, name, cmd): + """Lancer un outil de migration et RETENIR sa conclusion. + + Les outils partagent une convention de code de sortie — 0 rien à + signaler, 1 des trouvailles, 2 l'outil a échoué — et c'est + exactement ce qu'on veut relire plus tard. Sans cela, « le test de + fumée est-il passé ? » n'a pas de réponse une heure après. + """ + status = self.run_on_terminal(cmd) + self.record_event("test", name, status, cmd) + return status + def add_comment_progression(self, comment): comment_to_add = f"# {comment}" self.lst_command_executed.append(comment_to_add) diff --git a/test/test_database_cleanup.py b/test/test_database_cleanup.py index 29ca709..b6767a1 100755 --- a/test/test_database_cleanup.py +++ b/test/test_database_cleanup.py @@ -771,8 +771,15 @@ class TestTheMigrationRunsItBeforeTheSmokeTest(unittest.TestCase): from script.todo.todo_upgrade import TodoUpgrade source = inspect.getsource(TodoUpgrade.prompt_database_cleanup) - self.assertIn("run_on_terminal", source) + # `run_tool` retient le verdict PUIS délègue à `run_on_terminal` : + # les deux propriétés comptent, et vérifier la seconde à la source + # évite qu'un raccourci futur reprenne l'exécuteur qui capture. + self.assertIn("run_tool(", source) self.assertNotIn("todo_upgrade_execute", source) + self.assertIn( + "self.run_on_terminal(", + inspect.getsource(TodoUpgrade.run_tool), + ) def test_saying_no_cleans_nothing(self): # Le défaut ne retire pas le choix : il ne fait qu'en proposer un. diff --git a/test/test_error_retry_loop.py b/test/test_error_retry_loop.py index 296c012..d2554be 100644 --- a/test/test_error_retry_loop.py +++ b/test/test_error_retry_loop.py @@ -208,6 +208,8 @@ class TestTheResetReportsWhatItDid(unittest.TestCase): def test_a_reset_that_ran_is_True(self): obj = TodoUpgrade.__new__(TodoUpgrade) + obj.dct_progression = {} + obj.write_config = lambda: None obj.stale_cow_keys = lambda db: ["web.layout"] obj.ask_gate = lambda prompt, default="": default obj.run_on_terminal = lambda cmd: 0 @@ -218,6 +220,8 @@ class TestTheResetReportsWhatItDid(unittest.TestCase): # Rejouer derrière une réinitialisation qui a échoué, c'est brûler # une tentative sur un état inchangé. obj = TodoUpgrade.__new__(TodoUpgrade) + obj.dct_progression = {} + obj.write_config = lambda: None obj.stale_cow_keys = lambda db: ["web.layout"] obj.ask_gate = lambda prompt, default="": default obj.run_on_terminal = lambda cmd: 2 diff --git a/test/test_migration_status.py b/test/test_migration_status.py new file mode 100644 index 0000000..d335719 --- /dev/null +++ b/test/test_migration_status.py @@ -0,0 +1,376 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""« Où en est-on, et qu'est-ce qui a cassé ? » + +Une migration traverse six paliers, lance des centaines de commandes et +dure des heures. Le journal existant dit ce qui a été LANCÉ ; il ne dit +jamais ce que cela a donné. Trois heures plus tard on relit deux cents +lignes de commandes sans savoir laquelle a échoué, ni ce que le test de +fumée a conclu. + +Deux choses se vérifient ici, et la seconde est la moins évidente : + +- un outil relancé APRÈS correction a deux verdicts contradictoires dans + le journal, et c'est le dernier qui décrit la base telle qu'elle est. + Les afficher tous les deux sans les distinguer ferait lire une + réparation comme un échec persistant ; +- l'écran ne doit RIEN toucher. On l'ouvre en pleine migration, souvent + pendant qu'un serveur tourne. +""" + +import io +import os +import unittest +from contextlib import redirect_stdout + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +from script.todo import migration_status as status # noqa: E402 +from script.todo import migration_status_tui as tui # noqa: E402 +from script.todo import todo_i18n # noqa: E402 +from script.todo.todo_upgrade import TodoUpgrade # noqa: E402 + + +def progression(**override): + dct = { + "migration_file": "/a/b/technolibre_2026.zip", + "config_database_name": "test_neutralize", + "date_create": "2026-08-18 03:25:25", + "date_update": "2026-08-18 06:38:55", + "command_executed": [ + "# 0 - Inspect zip", + "make switch_odoo_12", + "# 2 - Succeed update all addons", + "./script/addons/update_addons_all.sh test", + "./script/odoo/migration/check_cow_views.py -d test", + ], + "lst_event": [ + { + "at": "05:01", + "step": "2 - Succeed update all addons", + "kind": "command", + "name": "update_addons_all.sh test", + "status": 1, + "detail": "", + }, + { + "at": "05:20", + "step": "2 - Succeed update all addons", + "kind": "test", + "name": "smoke_public_url", + "status": 2, + "detail": "", + }, + { + "at": "05:44", + "step": "2 - Succeed update all addons", + "kind": "test", + "name": "smoke_public_url", + "status": 0, + "detail": "", + }, + ], + } + dct.update(override) + return dct + + +class Base(unittest.TestCase): + def setUp(self): + self.addCleanup( + setattr, todo_i18n, "_current_lang", todo_i18n._current_lang + ) + todo_i18n._current_lang = "en" + + +class TestCuttingTheJournalByStep(Base): + def test_the_step_headers_are_the_cut(self): + # On réutilise le marquage « # » que la migration dépose déjà. + # En inventer un second qui divergerait serait pire que rien. + lst = status.journal_by_step(progression()) + self.assertEqual( + [s["step"] for s in lst], + ["0 - Inspect zip", "2 - Succeed update all addons"], + ) + + def test_the_commands_land_under_their_step(self): + lst = status.journal_by_step(progression()) + self.assertEqual(len(lst[1]["lst_cmd"]), 2) + + def test_a_command_before_any_header_is_not_lost(self): + dct = progression(command_executed=["make quelque_chose"]) + lst = status.journal_by_step(dct) + self.assertEqual(len(lst), 1) + self.assertEqual(lst[0]["lst_cmd"], ["make quelque_chose"]) + + def test_an_empty_progression_yields_nothing(self): + self.assertEqual(status.journal_by_step({}), []) + + +class TestARepairMustNotReadAsAFailure(Base): + """LE point délicat du résumé. + + `smoke_public_url` a rendu 2 puis 0 : il a échoué, on a réparé, il est + repassé. Montrer les deux verdicts côte à côte ferait conclure que la + base est toujours cassée. + """ + + def test_the_LAST_verdict_wins(self): + lst = status.tests_summary(progression()) + smoke = [x for x in lst if x["name"] == "smoke_public_url"][0] + self.assertEqual(smoke["status"], 0) + + def test_but_the_earlier_runs_are_still_counted(self): + # Les taire ferait croire à un premier essai réussi, et l'on + # perdrait la trace de ce qui a demandé une réparation. + lst = status.tests_summary(progression()) + smoke = [x for x in lst if x["name"] == "smoke_public_url"][0] + self.assertEqual(smoke["runs"], 2) + + def test_the_report_shows_both_facts(self): + text = status.render_text(progression()) + self.assertIn("smoke_public_url", text) + self.assertIn("2 runs", text) + self.assertIn("nothing to report", text) + + +class TestTheExitCodeConvention(Base): + """0 rien, 1 des trouvailles, 2 l'outil a échoué — partout la même.""" + + def test_zero_is_not_alarming(self): + self.assertEqual(status.verdict(0)[1], "nothing to report") + + def test_one_is_findings_not_a_crash(self): + # Le dire « échec » inquiéterait pour rien : 1 veut dire qu'il y a + # quelque chose à regarder, ce qui est le but de l'outil. + self.assertEqual(status.verdict(1)[1], "findings to look at") + + def test_two_is_the_tool_itself(self): + self.assertEqual(status.verdict(2)[1], "the tool itself failed") + + def test_anything_else_is_admitted_as_unknown(self): + self.assertEqual(status.verdict(77)[1], "unknown result") + + +class TestFailedCommands(Base): + def test_they_are_listed_newest_first(self): + dct = progression() + dct["lst_event"].append( + { + "at": "06:00", + "step": "3", + "kind": "command", + "name": "récente", + "status": 1, + "detail": "", + } + ) + self.assertEqual(status.failures(dct)[0]["name"], "récente") + + def test_they_carry_the_step_they_happened_in(self): + # Sans l'étape, il faut relire tout le journal pour savoir OÙ. + self.assertIn( + "2 - Succeed update all addons", + status.failures(progression())[0]["step"], + ) + + def test_tests_are_not_mixed_in_with_them(self): + self.assertEqual(len(status.failures(progression())), 1) + + +class TestItTouchesNothing(Base): + """On l'ouvre en pleine migration, souvent serveur allumé.""" + + def test_reading_a_missing_file_is_not_an_error(self): + self.assertEqual(status.read("/nexiste/pas.json"), {}) + + def test_reading_a_broken_file_is_not_an_error(self): + import tempfile + + chemin = os.path.join(tempfile.gettempdir(), "erplibre_casse.json") + with open(chemin, "w") as handle: + handle.write("{ pas du json") + self.addCleanup(os.remove, chemin) + self.assertEqual(status.read(chemin), {}) + + def test_nothing_in_the_module_writes_or_connects(self): + import inspect + + source = inspect.getsource(status) + for interdit in ("psql", "subprocess", "odoo_bin", 'open(.*, "w")'): + self.assertNotIn(interdit, source, interdit) + + def test_an_empty_progression_says_so_rather_than_crashing(self): + self.assertIn("No migration in progress", status.render_text({})) + + +class TestTheFullScreenShowsTheSameThing(Base): + def test_it_reads_the_same_assembly_as_the_text(self): + # Deux assemblages sépareraient les deux vues, et l'on finirait + # par lire deux états contradictoires de la même migration. + import inspect + + self.assertIn("import migration_status", inspect.getsource(tui)) + + def test_the_tests_come_before_the_steps(self): + # C'est la question qu'on se pose en ouvrant cet écran. + lst = tui.rows(progression()) + self.assertEqual(lst[0]["kind"], "test") + self.assertTrue(any(x["kind"] == "step" for x in lst)) + + def test_the_head_never_hides_what_failed(self): + texte = tui.head_text(progression()) + self.assertIn("test_neutralize", texte) + self.assertIn("1", texte) + + def test_a_step_pane_names_its_own_failures(self): + lst = tui.rows(progression()) + etape = [x for x in lst if x["kind"] == "step"][1] + self.assertIn( + "update_addons_all.sh test", tui.pane_text(progression(), etape) + ) + + def test_a_test_pane_gives_the_exit_code(self): + lst = tui.rows(progression()) + texte = tui.pane_text(progression(), lst[0]) + self.assertIn("smoke_public_url", texte) + + def test_a_pipe_is_explained_rather_than_silent(self): + out = io.StringIO() + with redirect_stdout(out): + self.assertFalse(tui.run_tui(progression())) + self.assertTrue(out.getvalue().strip()) + + def test_nothing_to_show_stays_silent(self): + out = io.StringIO() + with redirect_stdout(out): + self.assertFalse(tui.run_tui({})) + self.assertEqual(out.getvalue(), "") + + +class TestTheLetterIsFreeAndUniform(Base): + """« v » était pris : il veut dire « voir les différences » ailleurs.""" + + def test_v_is_taken_by_the_diff_prompts(self): + import inspect + + source = inspect.getsource(TodoUpgrade.prompt_cow_prediction) + self.assertIn('"v"', source) + + def test_the_state_letter_is_t_everywhere(self): + import inspect + + source = inspect.getsource(TodoUpgrade.ask_gate) + self.assertIn('reponse == "t"', source) + + def test_every_gate_prompt_announces_it(self): + chemin = os.path.join(REPO, "script", "todo", "todo_upgrade.py") + with open(chemin) as handle: + texte = handle.read() + # Une seule chaîne porte les deux raccourcis : les annoncer + # séparément les laisserait diverger d'une invite à l'autre. + self.assertGreaterEqual(texte.count("t = show the migration state"), 1) + self.assertNotIn("(b = go back to a previous step)", texte) + + def test_t_collides_with_no_other_answer(self): + import inspect + + for methode in ( + "prompt_uninstall_theme", + "prompt_database_cleanup", + "prompt_smoke_public_url", + "prompt_reset_stale_cow_views", + ): + source = inspect.getsource(getattr(TodoUpgrade, methode)) + self.assertNotIn('== "t"', source, methode) + + +class TestLookingIsNotAnswering(Base): + def test_the_state_reopens_the_same_question(self): + obj = TodoUpgrade.__new__(TodoUpgrade) + vu = [] + obj.show_migration_status = lambda: vu.append("ouvert") + reponses = iter(["t", "t", "y"]) + obj.ask = lambda prompt, default="": next(reponses) + self.assertEqual(obj.ask_gate("q : "), "y") + self.assertEqual(len(vu), 2) + + def test_it_writes_before_it_reads(self): + # L'écran lit le FICHIER de progression : ce qui vient de se passer + # n'y serait pas encore. + import inspect + + source = inspect.getsource(TodoUpgrade.show_migration_status) + self.assertLess( + source.index("write_config"), source.index("subprocess.call") + ) + + def test_looking_does_not_pollute_the_journal(self): + # `run_on_terminal` consigne ce qu'il lance, et le journal est + # justement ce que cet écran montre. + import inspect + + source = inspect.getsource(TodoUpgrade.show_migration_status) + # L'APPEL, pas le mot : le commentaire du correctif nomme lui-même + # ce qu'il faut éviter, et le chercher à l'aveugle se déclenchait + # sur la prose plutôt que sur le code. + self.assertNotIn("self.run_on_terminal(", source) + + +class TestWhatGetsRecorded(Base): + def upgrade(self): + obj = TodoUpgrade.__new__(TodoUpgrade) + obj.dct_progression = {} + obj.lst_command_executed = [] + obj.write_config = lambda: None + obj.current_step = "4.2.I - Migrate database" + return obj + + def test_a_tool_run_keeps_its_verdict(self): + obj = self.upgrade() + obj.run_on_terminal = lambda cmd: 1 + self.assertEqual(obj.run_tool("smoke_public_url", "cmd"), 1) + event = obj.dct_progression["lst_event"][0] + self.assertEqual(event["kind"], "test") + self.assertEqual(event["status"], 1) + self.assertEqual(event["step"], "4.2.I - Migrate database") + + def test_the_list_is_bounded(self): + # Une migration lance des centaines de commandes ; un fichier qui + # enfle sans limite coûte plus cher à écrire qu'à lire. + obj = self.upgrade() + for index in range(TodoUpgrade.MAX_EVENT + 25): + obj.record_event("test", f"outil{index}", 0) + self.assertEqual( + len(obj.dct_progression["lst_event"]), TodoUpgrade.MAX_EVENT + ) + + def test_the_newest_survive_the_trimming(self): + obj = self.upgrade() + for index in range(TodoUpgrade.MAX_EVENT + 3): + obj.record_event("test", f"outil{index}", 0) + dernier = obj.dct_progression["lst_event"][-1] + self.assertEqual(dernier["name"], f"outil{TodoUpgrade.MAX_EVENT + 2}") + + def test_the_step_is_captured_when_it_is_printed(self): + import inspect + + source = inspect.getsource(TodoUpgrade.print_step) + self.assertIn("self.current_step = msg", source) + + def test_a_failing_command_is_recorded_before_the_prompt(self): + # Si l'on répond ctrl+c, l'échec doit tout de même figurer dans + # l'état : c'est précisément celui qu'on cherchera en revenant. + import inspect + + source = inspect.getsource(TodoUpgrade.todo_upgrade_execute) + self.assertLess( + source.index('record_event("command"'), + source.index("Error detected"), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_smoke_internal_ui.py b/test/test_smoke_internal_ui.py index 3c41c85..d12c85a 100644 --- a/test/test_smoke_internal_ui.py +++ b/test/test_smoke_internal_ui.py @@ -500,7 +500,7 @@ class TestASkipThatHidesAFailure(unittest.TestCase): source = inspect.getsource(TodoUpgrade.prompt_smoke_public_url) self.assertLess( source.index("Public pages only:"), - source.index("run_on_terminal"), + source.index("run_tool("), )