diff --git a/script/analyse/check_migration_quality.py b/script/analyse/check_migration_quality.py index 1d750bb..3d7b51b 100755 --- a/script/analyse/check_migration_quality.py +++ b/script/analyse/check_migration_quality.py @@ -244,6 +244,28 @@ def event_excerpt(lignes, event, avant=18): return lignes[max(0, rang - avant) : fin + 1], rang +def excerpt_has_output(extrait, rang_relatif): + """La sortie de l'outil est-elle DANS l'extrait, ou seulement le cadre ? + + Avant que le pilote ne capture, le journal ne portait que « $ … » + suivi de « -> code », sans une ligne entre les deux. Le dire était + juste ; le dire encore quand la sortie est là serait un mensonge, et + le panneau enverrait chercher ailleurs ce qu'il a sous les yeux. + """ + for ligne in extrait[rang_relatif + 1 :]: + depouille = ligne.strip() + if not depouille: + continue + if MARQUEUR_COMMANDE in ligne: + continue + if depouille.startswith("-> ") or depouille.startswith("[test] "): + continue + if "] -> " in ligne or "] [test] " in ligne: + continue + return True + return False + + def event_database(event): """La base sur laquelle ce verdict portait, lue dans sa commande.""" for mot in event.get("detail", "").split(): diff --git a/script/analyse/check_migration_quality_tui.py b/script/analyse/check_migration_quality_tui.py index 691b42e..e0fcf57 100644 --- a/script/analyse/check_migration_quality_tui.py +++ b/script/analyse/check_migration_quality_tui.py @@ -372,7 +372,7 @@ def log_excerpt_lines(row, colour=False): brut = handle.read().splitlines() except OSError: return [] - extrait, _rang = quality.event_excerpt(brut, row.get("event") or {}) + extrait, rang = quality.event_excerpt(brut, row.get("event") or {}) lignes = [ status.paint(f" ── {t('step log')} ──", "step", colour), status.paint(f" {chemin}", "dim", colour), @@ -386,28 +386,23 @@ def log_excerpt_lines(row, colour=False): for ligne in extrait: lignes.append(f" {ligne[:150]}") lignes.append("") - lignes.append( - status.paint( - f" {t('the tool output is not in there: it goes to the')}", - "dim", - colour, - ) - ) - lignes.append( - status.paint( - f" {t('terminal, which a pipe would make full-screen tools')}", - "dim", - colour, - ) - ) - lignes.append( - status.paint( - f" {t('give up. What precedes is what Odoo was writing.')}", - "dim", - colour, - ) - ) - lignes.append("") + # Ne le dire que si c'est vrai : depuis que le pilote capture, la + # sortie EST là, et répéter qu'elle manque enverrait la chercher + # ailleurs alors qu'on l'a sous les yeux. + rang_relatif = None + if rang is not None: + debut = max(0, rang - 18) + rang_relatif = rang - debut + if rang_relatif is None or not quality.excerpt_has_output( + extrait, rang_relatif + ): + for phrase in ( + "the tool output is not in there: it goes to the", + "terminal, which a pipe would make full-screen tools", + "give up. What precedes is what Odoo was writing.", + ): + lignes.append(status.paint(f" {t(phrase)}", "dim", colour)) + lignes.append("") return lignes diff --git a/script/analyse/check_migration_residue.py b/script/analyse/check_migration_residue.py index ebda974..0de9d8b 100755 --- a/script/analyse/check_migration_residue.py +++ b/script/analyse/check_migration_residue.py @@ -287,7 +287,7 @@ def verdicts(database, path=None): def extrait_du_journal(dct, event, colour=True, avant=6): - """Ce que le journal d'étape montre autour de ce verdict. + """(lignes, la sortie de l'outil y est-elle) autour de ce verdict. La commande seule ne dit pas POURQUOI. Le journal, lui, garde ce qu'Odoo écrivait au moment du test — et c'est tout ce qu'on a : la @@ -300,19 +300,23 @@ def extrait_du_journal(dct, event, colour=True, avant=6): """ chemin = status.step_log_path(dct, event.get("step")) if not chemin or avant <= 0: - return [] + return [], False try: with open(chemin, "r", encoding="utf-8", errors="replace") as handle: brut = handle.read().splitlines() except OSError: - return [] - extrait, _rang = quality.event_excerpt(brut, event, avant=avant) + return [], False + extrait, rang = quality.event_excerpt(brut, event, avant=avant) if not extrait: - return [] + return [], False lignes = [paint(f" {chemin}", "dim", colour)] for ligne in extrait: lignes.append(paint(f" {ligne[:150]}", "dim", colour)) - return lignes + debut = max(0, rang - avant) if rang is not None else 0 + avec_sortie = rang is not None and quality.excerpt_has_output( + extrait, rang - debut + ) + return lignes, avec_sortie def verdicts_block(database, colour=True, path=None, lignes_avant=6): @@ -341,6 +345,7 @@ def verdicts_block(database, colour=True, path=None, lignes_avant=6): ) ) dct = quality.read_progression(chemin) + sortie_presente = False for event in ratés: version = quality.version_of(quality.event_database(event), dct) palier = str(version) if version else quality.event_step(event) @@ -354,7 +359,11 @@ def verdicts_block(database, colour=True, path=None, lignes_avant=6): lignes.append( paint(f" {event['detail'][:120]}", "dim", colour) ) - lignes.extend(extrait_du_journal(dct, event, colour, lignes_avant)) + bloc, avec_sortie = extrait_du_journal( + dct, event, colour, lignes_avant + ) + sortie_presente = sortie_presente or avec_sortie + lignes.extend(bloc) lignes.append("") lignes.append( paint( @@ -364,7 +373,7 @@ def verdicts_block(database, colour=True, path=None, lignes_avant=6): colour, ) ) - if ratés and lignes_avant > 0: + if ratés and lignes_avant > 0 and not sortie_presente: lignes.append( paint( f" {t('the tool output is not in the step log: it goes')}" diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index 8d6ef65..f1a9f1b 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -6979,6 +6979,10 @@ TRANSLATIONS = { }, # --- Écran de qualité : Verdicts, Validation, Revue --- # --- Journal d'étape et bascule de version --- + "capture unavailable": { + "fr": "capture indisponible", + "en": "capture unavailable", + }, "last run": { "fr": "dernière exécution", "en": "last run", diff --git a/script/todo/todo_upgrade.py b/script/todo/todo_upgrade.py index 908099a..e9ed49e 100755 --- a/script/todo/todo_upgrade.py +++ b/script/todo/todo_upgrade.py @@ -1813,7 +1813,7 @@ class TodoUpgrade: # la décision se prend sur le code de retour — et la capture faisait # annoncer « Command returned error code: 1 » sur un rapport qui va # bien. 1 veut dire « des copies casseront », pas « l'outil a raté ». - status = self.run_on_terminal( + status = self.run_captured( f"{PYTHON_BIN} ./script/odoo/migration/check_cow_views.py" f" -d {database_name} -t odoo{start_version + 1}.0" ) @@ -1837,7 +1837,9 @@ class TodoUpgrade: # Appliquer depuis là lève « KeyError: 'web_editor.assets' » — c'est # arrivé sur une vraie migration, et l'échec est passé pour un # succès. La correction est proposée après le palier, plus bas. - self.run_on_terminal( + # « --report-only » éteint l'invite de l'outil : rien ici + # n'attend de réponse, tout est bon à garder. + self.run_captured( f"{PYTHON_BIN} ./script/odoo/migration/check_stale_scss.py" f" -d {database_name} -t odoo{start_version + 1}.0 --report-only" ) @@ -2761,7 +2763,7 @@ class TodoUpgrade: # L'outil ne fait rien s'il n'y a pas de DMS, et rien non # plus s'il a déjà réparé : le rejouer est sans effet. if next_version == 13: - self.run_on_terminal( + self.run_captured( f"{PYTHON_BIN}" " ./script/odoo/migration/dms_access_repair.py" f" -d {database_name_upgrade} --apply" @@ -2798,10 +2800,11 @@ class TodoUpgrade: # c'est vrai aussi. Pourtant plus personne n'atteint les # données. Mesuré sur DMS au palier 13 : 69 fichiers et # 23 Mo intacts, zéro visible, pour tous les utilisateurs. - self.run_on_terminal( + self.run_tool( + "check_hidden_models", f"{PYTHON_BIN}" " ./script/odoo/migration/check_hidden_models.py" - f" -d {database_name_upgrade}" + f" -d {database_name_upgrade}", ) print(f"[y] {t('Open the server with Selenium')}") @@ -2961,7 +2964,7 @@ class TodoUpgrade: # Le thème parti, la commande mérite un nouvel essai — # et `repare` autorise le rejeu automatique borné. for theme in themes: - self.run_on_terminal( + self.run_captured( f"./script/addons/uninstall_addons_theme.sh" f" {database_name} {theme}" ) @@ -3348,13 +3351,14 @@ class TodoUpgrade: print(f"ℹ -> {t('Kept. Nothing was uninstalled.')}") return for theme in lst_theme: - # Sur le VRAI terminal : ce script finit par theme_leftover.py, - # qui pose une question. L'exécuteur capture la sortie par un - # tube — Python la met alors en tampon par blocs — et l'invite - # restait invisible pendant que le processus attendait une - # réponse. On tape Entrée à l'aveugle, plusieurs fois, et les - # frappes en trop vont à la question suivante. - self.run_on_terminal( + # Ce script finit par theme_leftover.py, qui pose une + # question. Un TUBE la rendrait invisible — Python met alors + # sa sortie en tampon par blocs — et l'on taperait Entrée à + # l'aveugle, les frappes en trop allant à la question + # suivante. Le pseudo-terminal de `run_captured` n'a pas ce + # défaut : l'invite s'affiche comme sur un terminal, et l'on + # garde tout de même le journal de la désinstallation. + self.run_captured( f"./script/addons/uninstall_addons_theme.sh" f" {database_name} {theme}" ) @@ -3536,13 +3540,106 @@ class TodoUpgrade: cmd += " --tui" self.run_on_terminal(cmd) - def run_on_terminal(self, cmd): - """Lance une commande en lui laissant le VRAI terminal. + # Les séquences ANSI d'un outil qui se croit — à juste titre — sur un + # terminal. Elles n'ont aucun sens dans un fichier qu'on relira, et + # une seule d'entre elles suffit à rendre l'extrait illisible. + RE_ANSI = re.compile(r"\x1b\[[0-9;?]*[a-zA-Z]|\x1b\][^\x07]*\x07") - `todo_upgrade_execute` capture la sortie par un tube. Un plein écran - y voit un stdout qui n'est pas un terminal, renonce, et retombe sur - son rapport texte : « w » réaffichait mot pour mot ce que « v » - venait de montrer, sans rien signaler. + def run_captured(self, cmd): + """Lancer en gardant la sortie, SANS que la commande s'en aperçoive. + + Un tube aurait suffi à capturer — et aurait changé le programme. + `smoke_public_url` demande `can_ask()`, qui exige stdin ET stdout + sur un terminal (smoke_public_url.py:63-72) : derrière un tube il + cesse d'offrir la réparation des vues COW, en silence. C'est + exactement la classe de régression que `run_on_terminal` évitait + en ne capturant rien du tout. + + Un pseudo-terminal lève le dilemme : l'enfant voit un vrai + terminal — mesuré, `isatty()` rend True des deux côtés — la + réponse tapée lui parvient, son code de sortie est intact, et la + sortie passe tout de même entre nos mains avant d'aller à l'écran. + + Ce qui est écrit dans le journal est nettoyé : les séquences ANSI + retirées, les secrets caviardés par le MÊME filtre que + `exec_command_live`, et les fins de ligne du terminal ramenées à + « \n ». Le terminal, lui, reçoit les octets tels quels. + """ + import pty + + self.lst_command_executed.append(cmd) + self.dct_progression["command_executed"] = self.lst_command_executed + self.write_config() + print(f"\n🏠 ⬇ {t('Execute command')} :\n") + print(cmd) + self.note_step_log(f"$ {cmd}") + + handle = getattr(self, "step_log", None) + if not handle: + # Sans journal ouvert il n'y a rien à gagner et un pty à + # payer : on retombe sur le chemin ordinaire. + status = subprocess.call(cmd, shell=True, executable="/bin/bash") + self.note_step_log(f" -> {status}") + return status + + reste = bytearray() + + def ecrire(morceau, dernier=False): + """Journaliser par LIGNES ENTIÈRES. + + Une séquence ANSI coupée entre deux lectures laisserait ses + débris dans le fichier ; on n'écrit donc qu'une fois la ligne + complète, et le reliquat part à la fermeture. + """ + reste.extend(morceau) + *lignes, tail = bytes(reste).split(b"\n") + del reste[:] + if dernier and tail: + lignes.append(tail) + else: + reste.extend(tail) + for ligne in lignes: + texte = ligne.decode("utf-8", "replace").replace("\r", "") + try: + handle.write( + execute.redact_secrets(self.RE_ANSI.sub("", texte)) + + "\n" + ) + except Exception: + return + + def lire(fd): + morceau = os.read(fd, 4096) + if morceau: + ecrire(morceau) + return morceau + + try: + brut = pty.spawn(["/bin/bash", "-c", cmd], lire) + status = os.waitstatus_to_exitcode(brut) + except Exception as exc: # noqa: BLE001 - un pty peut manquer + # Jamais au prix de l'exécution : mieux vaut la commande sans + # son journal que pas de commande du tout. + print(f"⚠ {t('capture unavailable')} : {exc}") + status = subprocess.call(cmd, shell=True, executable="/bin/bash") + else: + ecrire(b"", dernier=True) + self.note_step_log(f" -> {status}") + return status + + def run_on_terminal(self, cmd): + """Le terminal, tel quel, sans rien garder — pour les PLEINS ÉCRANS. + + C'est devenu l'exception : `run_captured` garde la sortie sans que + la commande s'en aperçoive, et presque tout passe par lui. Deux + appels restent ici, et pour une raison mesurée. + + `pty.spawn` crée son terminal en 0×0, et l'enfant lit sa taille + AVANT que le premier octet ne nous parvienne — donc avant qu'on + puisse la corriger. Vérifié : un programme lancé ainsi voit + « columns=0, lines=0 ». Sans conséquence pour un outil qui écrit + des lignes — aucun de ceux qu'on capture ne lit la largeur — mais + une application Textual s'y disposerait sur rien. Rien ne relit cette sortie — elle est REGARDÉE. Et le code de retour d'un afficheur ne veut pas dire « erreur » : 1 signifie « il y a des @@ -4524,7 +4621,7 @@ class TodoUpgrade: 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) + status = self.run_captured(cmd) self.record_event("test", name, status, cmd) return status diff --git a/test/test_check_migration_quality.py b/test/test_check_migration_quality.py index fce2a73..7074e5c 100644 --- a/test/test_check_migration_quality.py +++ b/test/test_check_migration_quality.py @@ -2042,6 +2042,60 @@ class TestTheStepLogInThePanel(Base): self.assertIn(chemin, texte) self.assertIn("40", texte) + def test_it_stays_quiet_once_the_output_is_actually_there(self): + # Le pilote capture désormais. Répéter que la sortie manque + # enverrait la chercher ailleurs alors qu'on l'a sous les yeux. + chemin = self.journal( + [ + "[2026-08-26 03:19:44.166204] $ ./script/odoo/migration/" + "smoke_public_url.py -d test_neutralize_upgrade_14", + "⧖ Démarrage d'Odoo…", + "❌ 3 URL sur 37 rendent 500", + "[2026-08-26 03:19:59.847489] [test] smoke_public_url -> 1", + ] + ) + texte = qtui.extra_pane(self.ligne_verdict(chemin)) + self.assertIn("3 URL sur 37", texte) + self.assertNotIn( + qtui.t("the tool output is not in there: it goes to the"), texte + ) + + def test_the_frame_alone_is_not_mistaken_for_output(self): + # « $ … » puis « -> 1 » : c'est le cadre, pas la sortie. + extrait = [ + "[2026-08-26 03:19:44.166204] $ ./script/x.py -d base", + "[2026-08-26 03:19:59.846406] -> 1", + "[2026-08-26 03:19:59.847489] [test] smoke_public_url -> 1", + ] + self.assertFalse(quality.excerpt_has_output(extrait, 0)) + + def test_one_line_of_output_is_enough(self): + extrait = [ + "[2026-08-26 03:19:44.166204] $ ./script/x.py -d base", + "une seule ligne dite par l'outil", + "[2026-08-26 03:19:59.846406] -> 1", + ] + self.assertTrue(quality.excerpt_has_output(extrait, 0)) + + def test_blank_lines_are_not_output(self): + extrait = [ + "[2026-08-26 03:19:44.166204] $ ./script/x.py -d base", + "", + " ", + "[2026-08-26 03:19:59.846406] -> 1", + ] + self.assertFalse(quality.excerpt_has_output(extrait, 0)) + + def test_what_came_before_the_command_is_not_its_output(self): + # Le contexte d'Odoo précède la commande ; le compter dirait + # « la sortie est là » sur tous les anciens journaux. + extrait = [ + "odoo: ce qui tournait avant", + "[2026-08-26 03:19:44.166204] $ ./script/x.py -d base", + "[2026-08-26 03:19:59.846406] -> 1", + ] + self.assertFalse(quality.excerpt_has_output(extrait, 1)) + def test_no_log_at_all_says_so(self): ligne = self.ligne_verdict(None) ligne["log"] = None diff --git a/test/test_check_stale_scss.py b/test/test_check_stale_scss.py index f301e5a..f0d70d9 100755 --- a/test/test_check_stale_scss.py +++ b/test/test_check_stale_scss.py @@ -285,22 +285,38 @@ class TestTheMigrationRunsIt(unittest.TestCase): source.index("4 - Upgrade version with OpenUpgrade"), ) - def test_it_gets_a_real_terminal(self): - # L'outil pose lui-même ses questions et peut ouvrir un plein écran : - # un tube les rendrait toutes injoignables, sans rien signaler. + def test_no_call_goes_through_the_piped_executor(self): + # L'outil pose lui-même ses questions : un TUBE les rendrait + # injoignables, sans rien signaler — Python met sa sortie en + # tampon par blocs et l'invite reste invisible. + # + # Les DEUX voies terminal conviennent : `run_on_terminal` laisse + # le vrai terminal, `run_captured` en fabrique un et garde une + # copie. Seul `todo_upgrade_execute` bufferise. import inspect from script.todo.todo_upgrade import TodoUpgrade source = inspect.getsource(TodoUpgrade.execute_odoo_upgrade) - # On ne compare pas à une mise en forme — black la change — mais à - # l'appel le plus proche EN AMONT : c'est lui qui exécute. - avant = source[: source.index("check_stale_scss.py")] - self.assertGreater( - avant.rfind("run_on_terminal("), - avant.rfind("todo_upgrade_execute("), - "l'outil repasse par l'exécuteur à tube", - ) + depart = 0 + vus = 0 + while True: + rang = source.find("check_stale_scss.py", depart) + if rang < 0: + break + vus += 1 + avant = source[:rang] + # L'appel le plus proche EN AMONT : c'est lui qui exécute. + terminal = max( + avant.rfind("run_on_terminal("), avant.rfind("run_captured(") + ) + self.assertGreater( + terminal, + avant.rfind("todo_upgrade_execute("), + "l'outil repasse par l'exécuteur à tube", + ) + depart = rang + 1 + self.assertEqual(2, vus) class TestTheFixCannotRunTooEarly(unittest.TestCase): diff --git a/test/test_database_cleanup.py b/test/test_database_cleanup.py index b6767a1..a97003e 100755 --- a/test/test_database_cleanup.py +++ b/test/test_database_cleanup.py @@ -771,15 +771,14 @@ class TestTheMigrationRunsItBeforeTheSmokeTest(unittest.TestCase): from script.todo.todo_upgrade import TodoUpgrade source = inspect.getsource(TodoUpgrade.prompt_database_cleanup) - # `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), - ) + capture = inspect.getsource(TodoUpgrade.run_captured) + self.assertIn("run_captured", inspect.getsource(TodoUpgrade.run_tool)) + # Un pseudo-terminal, jamais un tube : derrière un tube + # `can_ask()` rend False et l'outil cesse de poser sa question. + self.assertIn("pty.spawn", capture) + self.assertNotIn("subprocess.PIPE", capture) def test_saying_no_cleans_nothing(self): # Le défaut ne retire pas le choix : il ne fait qu'en proposer un. @@ -798,7 +797,7 @@ class TestTheMigrationRunsItBeforeTheSmokeTest(unittest.TestCase): upgrade.lst_command_executed = [] upgrade.write_config = lambda: None lst_cmd = [] - upgrade.run_on_terminal = lambda cmd: lst_cmd.append(cmd) or 0 + upgrade.run_captured = lambda cmd: lst_cmd.append(cmd) or 0 upgrade.ask_gate = lambda prompt, default="": "n" or default with contextlib.redirect_stdout(io.StringIO()): upgrade.prompt_database_cleanup("db") @@ -818,7 +817,7 @@ class TestTheMigrationRunsItBeforeTheSmokeTest(unittest.TestCase): upgrade.lst_command_executed = [] upgrade.write_config = lambda: None lst_cmd = [] - upgrade.run_on_terminal = lambda cmd: lst_cmd.append(cmd) or 0 + upgrade.run_captured = lambda cmd: lst_cmd.append(cmd) or 0 upgrade.ask_gate = lambda prompt, default="": "" or default with contextlib.redirect_stdout(io.StringIO()): upgrade.prompt_database_cleanup("db") @@ -833,7 +832,7 @@ class TestTheMigrationRunsItBeforeTheSmokeTest(unittest.TestCase): upgrade.lst_command_executed = [] upgrade.write_config = lambda: None lst_cmd = [] - upgrade.run_on_terminal = lambda cmd: lst_cmd.append(cmd) or 0 + upgrade.run_captured = lambda cmd: lst_cmd.append(cmd) or 0 upgrade.ask_gate = lambda prompt, default="": "y" or default upgrade.prompt_database_cleanup("db_upgrade_18") self.assertEqual(len(lst_cmd), 1) diff --git a/test/test_error_retry_loop.py b/test/test_error_retry_loop.py index 4729f13..3613bea 100644 --- a/test/test_error_retry_loop.py +++ b/test/test_error_retry_loop.py @@ -85,7 +85,7 @@ class Harness(unittest.TestCase): obj.prompt_reset_stale_cow_views = faux_reset obj.check_stale_cow_views = lambda db: None - obj.run_on_terminal = lambda cmd: 0 + obj.run_captured = lambda cmd: 0 reponses = iter(lst_answer or []) def faux_ask(prompt, default=""): @@ -233,7 +233,7 @@ class TestTheResetReportsWhatItDid(unittest.TestCase): 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 + obj.run_captured = lambda cmd: 0 with redirect_stdout(io.StringIO()): self.assertTrue(obj.prompt_reset_stale_cow_views("db")) @@ -245,7 +245,7 @@ class TestTheResetReportsWhatItDid(unittest.TestCase): 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 + obj.run_captured = lambda cmd: 2 with redirect_stdout(io.StringIO()): self.assertFalse(obj.prompt_reset_stale_cow_views("db")) diff --git a/test/test_migration_status.py b/test/test_migration_status.py index 62c486c..cb963e7 100644 --- a/test/test_migration_status.py +++ b/test/test_migration_status.py @@ -1559,7 +1559,7 @@ class TestWhatGetsRecorded(Base): def test_a_tool_run_keeps_its_verdict(self): obj = self.upgrade() - obj.run_on_terminal = lambda cmd: 1 + obj.run_captured = 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") diff --git a/test/test_monitoring.py b/test/test_monitoring.py index cd952ff..ab3dacc 100644 --- a/test/test_monitoring.py +++ b/test/test_monitoring.py @@ -556,6 +556,27 @@ class TestTheStepLogInTheReport(unittest.TestCase): ) self.assertNotIn("[test]", self.bloc(lignes_avant=0)) + def test_the_report_stops_saying_it_once_the_output_is_captured(self): + # Depuis que le pilote capture par pseudo-terminal, la sortie est + # dans le journal : le dire encore serait faux. + self.ecrire_progression([verdict()]) + self.ecrire_journal( + "4.1.I - Migrate database", + [ + "[2026-08-26 03:19:44.166204] $ .venv.erplibre/bin/python3" + " ./script/odoo/migration/smoke_public_url.py" + " -d test_neutralize_upgrade_14 --internal-required", + "❌ 3 URL sur 37 rendent 500", + "[2026-08-26 03:19:59.847489] [test] smoke_public_url -> 1", + ], + ) + texte = self.bloc() + self.assertIn("3 URL sur 37", texte) + self.assertNotIn( + residue.t("the tool output is not in the step log: it goes"), + texte, + ) + def test_it_says_the_tool_output_is_elsewhere(self): # Sans cela on cherche dans le journal une sortie qui n'y a # jamais été écrite. diff --git a/test/test_reset_stale_cow_prompt.py b/test/test_reset_stale_cow_prompt.py index c12fd8f..1639cde 100755 --- a/test/test_reset_stale_cow_prompt.py +++ b/test/test_reset_stale_cow_prompt.py @@ -44,7 +44,7 @@ class PromptCase(unittest.TestCase): upgrade.write_config = lambda: None upgrade.stale_cow_keys = lambda db: lst_key lst_cmd = [] - upgrade.run_on_terminal = lambda cmd: lst_cmd.append(cmd) or 0 + upgrade.run_captured = lambda cmd: lst_cmd.append(cmd) or 0 # Le doublon HONORE le défaut, comme le vrai `ask_gate` : sinon # les tests de défaut ne testeraient que le doublon. upgrade.ask_gate = lambda prompt, default="": answer or default diff --git a/test/test_smoke_public_url.py b/test/test_smoke_public_url.py index 3c0977c..3eea61b 100755 --- a/test/test_smoke_public_url.py +++ b/test/test_smoke_public_url.py @@ -433,7 +433,7 @@ class TestTheMigrationOffersIt(unittest.TestCase): upgrade.lst_command_executed = [] upgrade.write_config = lambda: None lst_cmd = [] - upgrade.run_on_terminal = lambda cmd: lst_cmd.append(cmd) or 0 + upgrade.run_captured = lambda cmd: lst_cmd.append(cmd) or 0 # Le doublon HONORE le défaut, comme le vrai `ask_gate`. upgrade.ask_gate = lambda prompt, default="": answer or default upgrade.prompt_smoke_public_url("db_upgrade_13") @@ -502,7 +502,7 @@ class TestTheMigrationOffersIt(unittest.TestCase): upgrade.dct_progression = {} upgrade.lst_command_executed = [] upgrade.write_config = lambda: None - upgrade.run_on_terminal = lambda cmd: 0 + upgrade.run_captured = lambda cmd: 0 upgrade.ask_gate = lambda prompt, default="": "n" out = io.StringIO() with contextlib.redirect_stdout(out): diff --git a/test/test_theme_on_error.py b/test/test_theme_on_error.py index 1b61111..ee97f67 100644 --- a/test/test_theme_on_error.py +++ b/test/test_theme_on_error.py @@ -169,10 +169,17 @@ class TestTheMenu(unittest.TestCase): def test_it_goes_through_the_real_terminal(self): # `uninstall_addons_theme.sh` finit par poser une question ; un # tube la rendrait invisible et l'on répondrait à l'aveugle. + # `run_captured` fabrique un pseudo-terminal : l'invite s'affiche + # comme sur un vrai, et le journal de désinstallation est gardé. bloc = self.source() debut = bloc.index('if wait_status == "6"') fin = bloc.index('if wait_status == "5"') - self.assertIn("run_on_terminal", bloc[debut:fin]) + fenetre = bloc[debut:fin] + self.assertTrue( + "run_captured" in fenetre or "run_on_terminal" in fenetre, + fenetre[:120], + ) + self.assertNotIn("todo_upgrade_execute", fenetre) def test_the_label_is_translated(self): self.assertIn( diff --git a/test/test_todo_upgrade_capture.py b/test/test_todo_upgrade_capture.py new file mode 100644 index 0000000..65cc94d --- /dev/null +++ b/test/test_todo_upgrade_capture.py @@ -0,0 +1,348 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""Capturer la sortie d'un test SANS que le test s'en aperçoive. + +Le journal d'étape gardait la commande et son code, jamais ce que la +commande avait écrit. Mesuré sur une vraie migration : « $ … +smoke_public_url.py … » suivi IMMÉDIATEMENT de « -> 1 », zéro ligne entre +les deux, trois fois de suite. L'écran d'analyse ne pouvait donc rien +montrer de ce qui avait échoué. + +Un tube aurait suffi à capturer, et aurait changé le programme : +`smoke_public_url` appelle `can_ask()`, qui exige stdin ET stdout sur un +terminal (smoke_public_url.py:63-72). Derrière un tube il cesse d'offrir +la réparation des vues COW — en silence, ce qui est le pire. + +D'où le pseudo-terminal. Ce fichier vérifie les deux moitiés de la +promesse : que la sortie arrive dans le journal, et que l'enfant continue +de voir un vrai terminal. +""" + +import io +import os +import shutil +import sys +import tempfile +import unittest +from contextlib import redirect_stdout + +REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, REPO) + +from script.todo.todo_upgrade import TodoUpgrade # noqa: E402 + + +class Pilote: + """Le strict nécessaire : `run_captured` ne touche à rien d'autre. + + Instancier le vrai pilote demanderait une configuration, une base et + un dépôt Odoo. La méthode, elle, n'a besoin que d'un journal ouvert et + de deux listes — c'est ce qui la rend testable. + """ + + RE_ANSI = TodoUpgrade.RE_ANSI + run_captured = TodoUpgrade.run_captured + + def __init__(self, handle): + self.step_log = handle + self.lst_command_executed = [] + self.dct_progression = {} + self.notes = [] + + def write_config(self): + pass + + def note_step_log(self, texte): + self.notes.append(texte) + if self.step_log: + self.step_log.write(texte + "\n") + + +class Base(unittest.TestCase): + def setUp(self): + self.dossier = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.dossier) + self.chemin = os.path.join(self.dossier, "etape.log") + + def outil(self, source): + chemin = os.path.join(self.dossier, "outil.py") + with io.open(chemin, "w", encoding="utf-8") as handle: + handle.write(source) + return f"{sys.executable} {chemin}" + + def lancer(self, cmd, avec_journal=True): + muet = os.open(os.devnull, os.O_WRONLY) + garde = os.dup(1) + os.dup2(muet, 1) + try: + return self._lancer(cmd, avec_journal) + finally: + os.dup2(garde, 1) + os.close(garde) + os.close(muet) + + def _lancer(self, cmd, avec_journal=True): + tampon = io.StringIO() + if not avec_journal: + pilote = Pilote(None) + with redirect_stdout(tampon): + code = pilote.run_captured(cmd) + return code, "", pilote + with io.open( + self.chemin, "w", encoding="utf-8", buffering=1 + ) as handle: + pilote = Pilote(handle) + with redirect_stdout(tampon): + code = pilote.run_captured(cmd) + with io.open(self.chemin, encoding="utf-8") as handle: + return code, handle.read(), pilote + + +class TestTheChildStillSeesATerminal(Base): + """La moitié qui n'a l'air de rien, et qui décide de tout.""" + + def test_stdout_is_a_terminal_for_the_child(self): + # Derrière un tube, `can_ask()` rend False et le test de fumée + # cesse d'offrir la réparation, sans le dire. + _code, journal, _p = self.lancer( + self.outil( + "import sys\nprint('T' if sys.stdout.isatty() else 'F')\n" + ) + ) + self.assertIn("\nT\n", "\n" + journal) + + def test_stdin_is_a_terminal_for_the_child(self): + _code, journal, _p = self.lancer( + self.outil( + "import sys\nprint('T' if sys.stdin.isatty() else 'F')\n" + ) + ) + self.assertIn("\nT\n", "\n" + journal) + + def test_the_exit_code_survives(self): + # C'est lui qui devient le verdict : 0 rien, 1 des trouvailles, + # 2 l'outil a échoué. Le perdre viderait `record_event` de sens. + code, _journal, _p = self.lancer( + self.outil("import sys\nsys.exit(4)\n") + ) + self.assertEqual(4, code) + + def test_a_zero_stays_a_zero(self): + code, _journal, _p = self.lancer(self.outil("pass\n")) + self.assertEqual(0, code) + + +class TestWhatTheLogKeeps(Base): + def test_the_output_is_there(self): + _code, journal, _p = self.lancer( + self.outil("print('la ligne qui manquait')\n") + ) + self.assertIn("la ligne qui manquait", journal) + + def test_the_command_and_its_code_frame_it(self): + _code, journal, _p = self.lancer(self.outil("print('milieu')\n")) + lignes = journal.splitlines() + self.assertTrue(lignes[0].startswith("$ ")) + self.assertIn("milieu", journal) + self.assertTrue(lignes[-1].strip().startswith("-> ")) + + def test_colour_is_stripped_from_the_file(self): + # L'enfant se croit — à juste titre — sur un terminal, donc il + # colorie. Une seule séquence suffit à rendre l'extrait illisible. + _code, journal, _p = self.lancer( + self.outil("print('\\x1b[31mrouge\\x1b[0m nu')\n") + ) + self.assertIn("rouge nu", journal) + self.assertNotIn("\x1b", journal) + + def test_a_sequence_split_across_two_reads_leaves_no_debris(self): + # On n'écrit qu'une fois la ligne complète : autrement les débris + # d'une séquence coupée restent dans le fichier. + source = ( + "import sys, time\n" + "sys.stdout.write('\\x1b[3')\n" + "sys.stdout.flush()\n" + "time.sleep(0.2)\n" + "sys.stdout.write('1mtardif\\x1b[0m\\n')\n" + "sys.stdout.flush()\n" + ) + _code, journal, _p = self.lancer(self.outil(source)) + self.assertIn("tardif", journal) + self.assertNotIn("\x1b", journal) + + def test_secrets_go_through_the_same_filter_as_everything_else(self): + _code, journal, _p = self.lancer( + self.outil("print('outil --db_password TRESSECRET -d base')\n") + ) + self.assertNotIn("TRESSECRET", journal) + self.assertIn("--db_password", journal) + + def test_the_last_line_without_a_newline_is_not_lost(self): + _code, journal, _p = self.lancer( + self.outil( + "import sys\nsys.stdout.write('sans saut final')\n" + "sys.stdout.flush()\n" + ) + ) + self.assertIn("sans saut final", journal) + + def test_carriage_returns_do_not_survive(self): + # Un terminal en pose à chaque ligne ; dans un fichier ils + # produisent des lignes qui s'écrasent à la lecture. + _code, journal, _p = self.lancer(self.outil("print('normale')\n")) + self.assertNotIn("\r", journal) + + +class TestWithoutAnOpenLog(Base): + """Sans journal, rien à gagner et un pty à payer.""" + + def test_the_command_still_runs_and_returns_its_code(self): + code, _journal, _p = self.lancer( + self.outil("import sys\nsys.exit(5)\n"), avec_journal=False + ) + self.assertEqual(5, code) + + def test_no_terminal_is_fabricated_when_there_is_nothing_to_keep(self): + """Sans journal, `run_captured` doit valoir `run_on_terminal`. + + Fabriquer un pseudo-terminal changerait ce que la commande voit — + couleur, invites, mise en page — pour un journal qui n'existe pas. + L'enfant écrit son verdict dans un fichier À CÔTÉ, car sa sortie + standard est précisément ce qu'on est en train de mesurer. + """ + temoin = os.path.join(self.dossier, "temoin.txt") + cmd = self.outil( + "import io, sys\n" + f"io.open({temoin!r}, 'w').write(str(sys.stdout.isatty()))\n" + ) + self.lancer(cmd, avec_journal=False) + with io.open(temoin, encoding="utf-8") as handle: + self.assertEqual("False", handle.read()) + + def test_a_terminal_IS_fabricated_when_there_is_a_log(self): + # Le pendant : c'est ce qui permet à l'outil de poser sa question. + temoin = os.path.join(self.dossier, "temoin2.txt") + cmd = self.outil( + "import io, sys\n" + f"io.open({temoin!r}, 'w').write(str(sys.stdout.isatty()))\n" + ) + self.lancer(cmd, avec_journal=True) + with io.open(temoin, encoding="utf-8") as handle: + self.assertEqual("True", handle.read()) + + def test_the_command_is_still_recorded(self): + _code, _journal, pilote = self.lancer( + self.outil("pass\n"), avec_journal=False + ) + self.assertEqual(1, len(pilote.lst_command_executed)) + self.assertEqual( + pilote.lst_command_executed, + pilote.dct_progression["command_executed"], + ) + + +class TestTheDriverUsesIt(unittest.TestCase): + """Le câblage : quels appels passent par la capture, et lesquels non.""" + + def setUp(self): + import inspect + + self.source = inspect.getsource(TodoUpgrade) + + def test_every_tool_verdict_goes_through_the_capture(self): + import inspect + + run_tool = inspect.getsource(TodoUpgrade.run_tool) + self.assertIn("run_captured", run_tool) + self.assertNotIn("run_on_terminal", run_tool) + + def test_the_hidden_models_check_is_a_recorded_verdict(self): + # Il suit la même convention de code de sortie que les autres et + # tournait pourtant sans que rien ne retienne sa conclusion. + avant = self.source.index("check_hidden_models.py") + debut = self.source.rindex("self.run_", 0, avant) + self.assertTrue( + self.source[debut:avant].startswith("self.run_tool("), + self.source[debut:avant][:60], + ) + + def test_only_full_screen_paths_stay_outside_the_capture(self): + """La frontière, épinglée par ce qu'elle laisse dehors. + + Elle a une raison mesurée : `pty.spawn` crée son terminal en 0×0 + et l'enfant lit sa taille avant le premier octet. Sans effet sur + un outil qui écrit des lignes ; fatal pour une application qui se + dispose. Tout le reste doit donc être capturé. + """ + import re + + lignes = self.source.split("\n") + dehors = [] + for rang, ligne in enumerate(lignes): + if "self.run_on_terminal(" not in ligne: + continue + suite = " ".join(x.strip() for x in lignes[rang : rang + 7]) + trouve = re.search(r"(\w+\.(?:py|sh))", suite) + if trouve: + dehors.append((trouve.group(1), "--tui" in suite)) + self.assertTrue(dehors, "plus aucun appel : la garde ne garde rien") + for outil, tui in dehors: + # Soit il demande explicitement un plein écran, soit son + # invite peut en ouvrir un — check_stale_scss le fait sur « w ». + self.assertTrue( + tui or outil == "check_stale_scss.py", + f"{outil} n'est pas un plein écran : il devrait être capturé", + ) + + def test_the_tools_captured_do_not_read_the_terminal_width(self): + # C'est ce qui rend le 0×0 sans conséquence pour eux. Si l'un s'y + # mettait, sa mise en page partirait sur une largeur nulle. + import glob + + for chemin in glob.glob( + os.path.join(REPO, "script", "odoo", "migration", "*.py") + ): + if os.path.basename(chemin) in ( + "check_stale_scss.py", + "reset_stale_cow_views.py", + ): + continue + with io.open(chemin, encoding="utf-8") as handle: + source = handle.read() + if "run_captured" in source: + continue + self.assertNotIn( + "get_terminal_size", source, os.path.basename(chemin) + ) + + def test_the_full_screen_path_still_captures_nothing(self): + # C'est sa raison d'être : un plein écran derrière un tube + # renonce et retombe sur son rapport texte. + import inspect + + import ast + + import textwrap + + arbre = ast.parse( + textwrap.dedent(inspect.getsource(TodoUpgrade.run_on_terminal)) + ) + corps = arbre.body[0].body + if isinstance(corps[0], ast.Expr) and isinstance( + corps[0].value, ast.Constant + ): + corps = corps[1:] + code = "\n".join(ast.unparse(noeud) for noeud in corps) + self.assertIn("subprocess.call", code) + self.assertNotIn("pty", code) + + def test_the_viewer_of_cow_copies_keeps_its_full_screen(self): + avant = self.source.index("cow_drift.py") + suite = self.source[avant : avant + 600] + self.assertIn("self.run_on_terminal(cmd)", suite) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_uninstall_addons_theme.py b/test/test_uninstall_addons_theme.py index af5ffa3..b95778f 100755 --- a/test/test_uninstall_addons_theme.py +++ b/test/test_uninstall_addons_theme.py @@ -156,7 +156,7 @@ class TestTheMigrationOffersIt(unittest.TestCase): )[1] # Le désinstalleur ne passe PLUS par l'exécuteur qui capture : il # pose une question, et un tube la rendrait invisible. - obj.run_on_terminal = lambda cmd: self.lst_cmd.append(cmd) or 0 + obj.run_captured = lambda cmd: self.lst_cmd.append(cmd) or 0 # Le doublon HONORE le défaut, comme le vrai `ask_gate` : sinon # les tests de défaut ne testeraient que le doublon. obj.ask_gate = lambda prompt, default="": answer or default @@ -347,7 +347,10 @@ class TestTheQuestionMustBeVisible(unittest.TestCase): from script.todo.todo_upgrade import TodoUpgrade source = inspect.getsource(TodoUpgrade.prompt_uninstall_theme) - self.assertIn("run_on_terminal", source) + # Un pseudo-terminal convient : l'invite reste visible, et la + # sortie est gardée pour l'écran d'analyse. Seul le tube de + # `todo_upgrade_execute` la rendrait muette. + self.assertIn("run_captured", source) self.assertNotIn("todo_upgrade_execute", source) def test_asking_needs_stdout_not_just_stdin(self): @@ -458,10 +461,10 @@ class TestTheMisleadingErrorCode(unittest.TestCase): source = inspect.getsource(TodoUpgrade.execute_odoo_upgrade) avant = source[: source.index("check_cow_views.py")] - self.assertGreater( - avant.rfind("run_on_terminal("), - avant.rfind("todo_upgrade_execute("), + terminal = max( + avant.rfind("run_on_terminal("), avant.rfind("run_captured(") ) + self.assertGreater(terminal, avant.rfind("todo_upgrade_execute(")) class TestExitCodes(unittest.TestCase):