[ADD] migration : garder la sortie des tests, par pseudo-terminal

Le journal d'étape notait la commande et son code de retour, jamais ce
qu'elle avait écrit : l'écran d'analyse ne pouvait rien montrer d'un
échec. Un tube aurait capturé et changé le programme — smoke_public_url
appelle can_ask(), qui exige stdin ET stdout sur un terminal, et derrière
un tube il cesse en silence d'offrir la réparation des vues COW. Un
pseudo-terminal lève le dilemme : l'enfant voit un vrai terminal, la
réponse tapée lui parvient, le code de retour survit. Neuf exécutions y
passent, dont check_hidden_models, qui tournait sans verdict retenu ;
deux restent dehors, pty.spawn naît en 0×0 et un plein écran s'y perdrait.

--- EN ---

The step log recorded the command and its exit code, never what it
wrote: the analysis screen could show nothing of a failure. A pipe would
have captured and changed the program — smoke_public_url calls can_ask(),
which requires stdin AND stdout to be terminals, and behind a pipe it
silently stops offering the COW view repair. A pty settles it: the child
sees a real terminal, a typed answer reaches it, the exit code survives.
Nine runs go through it, including check_hidden_models, which ran with no
verdict recorded; two stay out, pty.spawn starts at 0×0 and a full-screen
app would lay out on nothing.

Assisted-by: Claude Opus 5
(cherry picked from commit 81e9227502a0d16e6409338cf958495a5f8e3827)
This commit is contained in:
Mathieu Benoit 2026-08-27 05:54:24 -04:00
parent 1f606fed00
commit f6aa2575c3
16 changed files with 660 additions and 85 deletions

View file

@ -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():

View file

@ -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

View file

@ -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')}"

View file

@ -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",

View file

@ -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

View file

@ -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

View file

@ -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):

View file

@ -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)

View file

@ -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"))

View file

@ -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")

View file

@ -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.

View file

@ -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

View file

@ -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):

View file

@ -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(

View file

@ -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()

View file

@ -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):