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("), )