[ADD] migration state: l, p, r and the two resize keys

Four keys, and one bug behind them. « l » separates the commands from the
server log: the first says what was LAUNCHED, the second what it
ANSWERED, and an update writing tens of thousands of lines buries three
commands. « p » hides the summary panel, « - » and « + » move the split,
« r » re-reads the disk — the migration writes while one watches, and
closing the screen to see the next bump is what one ended up doing.

The « missing logs » had two causes, both real. A tail was shown without
saying it was one; it now names how many lines are hidden. And the first
two steps run before the database is named, so their logs landed under
« sans-nom » — outside the migration they belong to, invisible from the
screen. They are brought back the moment the name is known, appended
never overwritten.

The step 4 bug was already fixed: its log held all six bumps because only
print_step opened one, and the bump loop never calls it.

--- FR ---

[ADD] état de migration : l, p, r et les deux touches de taille

Quatre touches, et un défaut derrière. « l » sépare les commandes du
journal : la première dit ce qui a été LANCÉ, le second ce que cela a
RÉPONDU, et une mise à jour qui écrit des dizaines de milliers de lignes
enterre trois commandes. « p » cache le panneau de résumé, « - » et « + »
déplacent la séparation, « r » relit le disque — la migration écrit
pendant qu'on regarde.

Les « logs manquants » avaient deux causes. On montrait une fin sans dire
que c'en était une ; le nombre de lignes cachées est désormais nommé. Et
les deux premières étapes tournent avant que la base ne soit nommée :
leurs journaux atterrissaient sous « sans-nom », hors de la migration à
laquelle ils appartiennent. Ils la rejoignent dès que le nom est connu.

Assisted-by: Claude Opus 5
This commit is contained in:
Mathieu Benoit 2026-08-19 06:23:20 -04:00
parent 1e591dd949
commit c0c5144680
5 changed files with 401 additions and 23 deletions

View file

@ -190,16 +190,22 @@ def step_slug(msg):
return (nom or "step")[:80].lower()
def step_log_tail(dct, step, lines=200):
"""Les dernières lignes du journal de cette étape."""
def step_log_tail(dct, step, lines=400):
"""Les dernières lignes du journal de cette étape, et le compte total.
Rend (lignes, total). Le total n'est pas décoratif : une mise à jour de
modules écrit des dizaines de milliers de lignes, et montrer les
dernières SANS dire qu'on en cache se lit comme « il manque des logs ».
"""
chemin = step_log_path(dct, step)
if not chemin:
return []
return [], 0
try:
with open(chemin, "r", encoding="utf-8", errors="replace") as handle:
return handle.read().splitlines()[-lines:]
lst = handle.read().splitlines()
except OSError:
return []
return [], 0
return lst[-lines:], len(lst)
def journal_by_step(dct):
@ -405,7 +411,7 @@ def main(argv=None):
from script.todo.migration_status_tui import run_tui
except Exception:
run_tui = None
if run_tui and run_tui(dct):
if run_tui and run_tui(dct, path=config.file):
return 0
print(render_text(dct))
return 0

View file

@ -89,7 +89,7 @@ def rows(dct):
return lst
def pane_text(dct, row, colour=False):
def pane_text(dct, row, colour=False, show_log=True):
"""Le détail de la ligne choisie.
Le coloriage passe par de l'ANSI, que Rich sait décoder — et qui rend
@ -128,37 +128,66 @@ def pane_text(dct, row, colour=False):
lignes.append(f"· {status.paint(cmd, 'cmd', colour)}")
# La SORTIE des commandes, relue sur disque. C'est ce qui manquait :
# la liste des commandes dit ce qui a été lancé, jamais ce que cela a
# répondu — et c'est la réponse qu'on vient chercher.
tail = status.step_log_tail(dct, section["step"])
# répondu. Mais les deux mélangés dans un même panneau se confondent —
# d'où « l », qui les sépare.
tail, total = status.step_log_tail(dct, section["step"])
if not show_log:
if total:
lignes.append("")
lignes.append(
f"── {t('server log')} : {total} {t('lines')}"
f" ({t('press l to show')}) ──"
)
return "\n".join(lignes)
if tail:
lignes.append("")
lignes.append(f"── {t('server log')} ──")
cache = (
f" — {t('last')} {len(tail)} {t('of')} {total}"
if total > len(tail)
else ""
)
lignes.append(f"── {t('server log')}{cache} ──")
lignes.extend(tail)
elif not section["lst_cmd"]:
lignes.append(t("No tool has run yet."))
return "\n".join(lignes)
def build_app(dct):
def build_app(dct, path=None):
"""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 rich.text import Text
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")]
BINDINGS = [
("q,escape", "quit", t("Quit")),
("r", "refresh", t("Refresh")),
("l", "toggle_log", t("Logs")),
("p", "toggle_head", t("Panel")),
("plus,equal", "wider", t("Wider")),
("minus,underscore", "narrower", t("Narrower")),
]
def __init__(self, dct):
LEFT_MIN = 16
LEFT_MAX = 110
LEFT_STEP = 6
def __init__(self, dct, path=None):
super().__init__()
self.dct = dct
self.path = path
self.lst_row = rows(dct)
self.index = 0
self.show_log = True
self.left_width = 46
def compose(self) -> ComposeResult:
yield Header()
@ -171,23 +200,30 @@ def build_app(dct):
def on_mount(self):
self.title = t("Migration state")
self._fill_table()
self._show()
def _fill_table(self):
table = self.query_one("#left", DataTable)
table.clear(columns=True)
table.add_columns(t("Test results"), "#")
for row in self.lst_row:
table.add_row(row["label"][:38], row["detail"])
table.styles.width = self.left_width
self.query_one("#head", Static).update(head_text(self.dct))
self._show()
def _show(self):
from rich.text import Text
row = self.lst_row[self.index] if self.lst_row else None
# `from_ansi` fait DEUX choses : il rend les couleurs, et il
# traite le reste comme du texte LITTÉRAL. Sans lui, une
# commande contenant « [1] » passait pour du balisage Rich et
# disparaissait de l'écran sans que rien ne le signale.
self.query_one("#content", Static).update(
Text.from_ansi(pane_text(self.dct, row, colour=True))
Text.from_ansi(
pane_text(
self.dct, row, colour=True, show_log=self.show_log
)
)
)
def on_data_table_row_highlighted(self, event):
@ -195,10 +231,47 @@ def build_app(dct):
self.index = event.cursor_row
self._show()
return StatusApp(dct)
def action_refresh(self):
"""Relire le disque. La migration écrit PENDANT qu'on regarde.
L'écran s'ouvre au milieu d'une migration qui continue : sans
cela, il fallait le fermer et le rouvrir pour voir le palier
suivant.
"""
if not self.path:
return
self.dct = status.read(self.path)
self.lst_row = rows(self.dct)
self.index = min(self.index, max(0, len(self.lst_row) - 1))
self._fill_table()
self._show()
def action_toggle_log(self):
self.show_log = not self.show_log
self._show()
def action_toggle_head(self):
head = self.query_one("#head", Static)
head.display = not head.display
def action_wider(self):
self._resize(self.LEFT_STEP)
def action_narrower(self):
self._resize(-self.LEFT_STEP)
def _resize(self, delta):
# Bornée des deux côtés : une colonne de zéro ne se retrouve
# plus, et une qui mange tout l'écran ne laisse rien à lire.
self.left_width = max(
self.LEFT_MIN, min(self.LEFT_MAX, self.left_width + delta)
)
self.query_one("#left", DataTable).styles.width = self.left_width
return StatusApp(dct, path)
def run_tui(dct, run_app=True):
def run_tui(dct, run_app=True, path=None):
"""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
@ -216,7 +289,7 @@ def run_tui(dct, run_app=True):
if textual_setup and not textual_setup.ensure():
return False
try:
app = build_app(dct)
app = build_app(dct, path=path)
except ImportError:
print(
f"ℹ️ {t('Textual is missing from this interpreter:')}"

View file

@ -5635,6 +5635,38 @@ TRANSLATIONS = {
"fr": "de la ou des clés demandées malgré tout.",
"en": "the requested key(s) anyway.",
},
"lines": {
"fr": "lignes",
"en": "lines",
},
"press l to show": {
"fr": "l pour afficher",
"en": "press l to show",
},
"last": {
"fr": "dernières",
"en": "last",
},
"Refresh": {
"fr": "Rafraîchir",
"en": "Refresh",
},
"Logs": {
"fr": "Journaux",
"en": "Logs",
},
"Panel": {
"fr": "Panneau",
"en": "Panel",
},
"Wider": {
"fr": "Élargir",
"en": "Wider",
},
"Narrower": {
"fr": "Rétrécir",
"en": "Narrower",
},
"Clean the database before testing the pages?": {
"fr": "Nettoyer la base avant de tester les pages ?",
"en": "Clean the database before testing the pages?",

View file

@ -66,6 +66,9 @@ PATH_MIGRATION_PRIVATE = os.path.join("private", "odoo", "migration")
# Le journal permanent des échecs et des verdicts, hors du fichier de
# progression : celui-ci est remis à zéro quand on recommence.
STEP_LOG_DIR = "step_log"
# Les deux premières étapes tournent avant qu'on ait choisi le nom de
# la base : leurs journaux attendent ici, puis rejoignent la migration.
UNNAMED_MIGRATION = "sans-nom"
EVENT_FILE = "events.jsonl"
# Steps of the migration, in order. What each one owns is declared just below,
# by GLOBAL_PROGRESSION_KEY and STEP_OWNED_KEY — not by the prefix alone, which
@ -3929,8 +3932,53 @@ class TodoUpgrade:
os.makedirs(chemin, exist_ok=True)
except OSError:
return None
if database != UNNAMED_MIGRATION:
self.adopt_unnamed_logs(chemin)
return chemin
def adopt_unnamed_logs(self, chemin):
"""Rapatrier les journaux écrits AVANT que la base ne soit nommée.
Les deux premières étapes — inspecter l'archive, la restaurer —
tournent avant qu'on ait choisi le nom de la base. Leurs journaux
atterrissaient donc sous « sans-nom », c'est-à-dire hors de la
migration à laquelle ils appartiennent : mesuré, deux fichiers
invisibles depuis l'écran d'état, et l'on cherchait des logs
manquants qui étaient simplement à côté.
En AJOUT si le fichier existe déjà : une reprise peut avoir écrit
des deux côtés, et écraser perdrait le premier passage.
"""
if getattr(self, "_unnamed_adopted", False):
return
self._unnamed_adopted = True
source = os.path.join(
PATH_MIGRATION_PRIVATE, UNNAMED_MIGRATION, STEP_LOG_DIR
)
if not os.path.isdir(source) or os.path.abspath(
source
) == os.path.abspath(chemin):
return
for nom in sorted(os.listdir(source)):
depart = os.path.join(source, nom)
arrivee = os.path.join(chemin, nom)
if not os.path.isfile(depart):
continue
try:
with open(
depart, "r", encoding="utf-8", errors="replace"
) as f:
contenu = f.read()
with open(arrivee, "a", encoding="utf-8") as f:
f.write(contenu)
os.remove(depart)
except OSError:
continue
try:
os.rmdir(source)
except OSError:
pass
def open_step_log(self, msg):
"""Rediriger la sortie des commandes vers le journal de cette étape.

View file

@ -593,6 +593,89 @@ class TestWhatSurvivesClosingTheTool(DiskCase):
[x["step"] for x in lst], ["4.1 - version 14", "4.2 - version 15"]
)
def test_logs_written_before_the_name_are_brought_back(self):
"""Les deux premières étapes tournent avant qu'on nomme la base.
Leurs journaux atterrissaient sous « sans-nom », c'est-à-dire hors
de la migration à laquelle ils appartiennent : mesuré sur la VM,
deux fichiers invisibles depuis l'écran d'état, et l'on cherchait
des logs manquants qui étaient simplement à côté.
"""
from script.todo import todo_upgrade as tu
anonyme = TodoUpgrade.__new__(TodoUpgrade)
anonyme.dct_progression = {}
anonyme.lst_command_executed = []
anonyme.write_config = lambda: None
anonyme.add_comment_progression("0 - Inspect zip")
anonyme.note_step_log("avant le nom")
anonyme.close_step_log()
self.assertTrue(
os.path.isdir(
os.path.join(
"private",
"odoo",
"migration",
tu.UNNAMED_MIGRATION,
"step_log",
)
)
)
# La base prend son nom : les journaux doivent la rejoindre.
nomme = self.upgrade()
nomme.log_dir()
tail, _total = status.step_log_tail(
{"config_database_name": "essai_db"}, "0 - Inspect zip"
)
self.assertIn("avant le nom", "\n".join(tail))
def test_the_unnamed_folder_is_left_empty_behind(self):
from script.todo import todo_upgrade as tu
anonyme = TodoUpgrade.__new__(TodoUpgrade)
anonyme.dct_progression = {}
anonyme.lst_command_executed = []
anonyme.write_config = lambda: None
anonyme.add_comment_progression("0 - Inspect zip")
anonyme.note_step_log("x")
anonyme.close_step_log()
self.upgrade().log_dir()
self.assertFalse(
os.path.isdir(
os.path.join(
"private",
"odoo",
"migration",
tu.UNNAMED_MIGRATION,
"step_log",
)
)
)
def test_an_existing_file_is_APPENDED_to_not_replaced(self):
# Une reprise peut avoir écrit des deux côtés ; écraser perdrait
# le premier passage.
nomme = self.upgrade()
nomme.add_comment_progression("0 - Inspect zip")
nomme.note_step_log("déjà là")
nomme.close_step_log()
anonyme = TodoUpgrade.__new__(TodoUpgrade)
anonyme.dct_progression = {}
anonyme.lst_command_executed = []
anonyme.write_config = lambda: None
anonyme.add_comment_progression("0 - Inspect zip")
anonyme.note_step_log("venu de sans-nom")
anonyme.close_step_log()
neuf = self.upgrade()
neuf.log_dir()
texte = "\n".join(
status.step_log_tail(
{"config_database_name": "essai_db"}, "0 - Inspect zip"
)[0]
)
self.assertIn("déjà là", texte)
self.assertIn("venu de sans-nom", texte)
def test_a_migration_without_a_database_writes_nowhere(self):
obj = self.upgrade(database=None)
obj.dct_progression = {}
@ -634,7 +717,7 @@ class TestTheStepLogs(DiskCase):
obj.print_step("2 - Succeed update all addons")
obj.note_step_log("second passage")
obj.close_step_log()
tail = status.step_log_tail(
tail, _total = status.step_log_tail(
{"config_database_name": "essai_db"},
"2 - Succeed update all addons",
)
@ -652,7 +735,7 @@ class TestTheStepLogs(DiskCase):
texte = "\n".join(
status.step_log_tail(
{"config_database_name": "essai_db"}, "3 - Clean up database"
)
)[0]
)
self.assertIn("$ true", texte)
self.assertIn("-> 0", texte)
@ -689,7 +772,7 @@ class TestTheCommandOutputItself(DiskCase):
status.step_log_tail(
{"config_database_name": "essai_db"},
"2 - Succeed update all addons",
)
)[0]
)
self.assertIn("première", texte)
# stderr aussi : c'est là que les erreurs d'Odoo se trouvent.
@ -875,6 +958,142 @@ class TestTheFullScreenColoursToo(Base):
self.assertIn("colour=True", source)
class TestSeparatingCommandsFromLogs(Base):
"""Les deux mélangés dans un même panneau se confondent.
La liste des commandes dit ce qui a été LANCÉ ; le journal dit ce que
cela a RÉPONDU. Ce sont deux lectures différentes, et l'une noie
l'autre : un `update_addons_all` écrit des dizaines de milliers de
lignes au-dessus desquelles trois commandes disparaissent.
"""
def dct(self):
return progression(command_executed=["# 2 - Update", "./run.sh -d db"])
def etape(self, dct):
return [x for x in tui.rows(dct) if x["kind"] == "step"][0]
def test_hiding_the_log_keeps_the_commands(self):
dct = self.dct()
texte = tui.pane_text(dct, self.etape(dct), show_log=False)
self.assertIn("./run.sh -d db", texte)
def test_hiding_it_never_hides_it_SILENTLY(self):
# Un panneau qui se vide sans un mot se lit comme « il n'y a rien »,
# ce qui est exactement le contraire de ce qui vient de se passer.
import tempfile
dossier = tempfile.mkdtemp(prefix="erplibre_essai_")
avant = os.getcwd()
os.chdir(dossier)
self.addCleanup(shutil.rmtree, dossier, True)
self.addCleanup(os.chdir, avant)
chemin = os.path.join("private", "odoo", "migration", "db", "step_log")
os.makedirs(chemin)
with open(
os.path.join(chemin, status.step_slug("2 - Update") + ".log"), "w"
) as handle:
handle.write("\n".join(f"ligne {i}" for i in range(50)))
dct = self.dct()
dct["config_database_name"] = "db"
texte = tui.pane_text(dct, self.etape(dct), show_log=False)
self.assertIn("50", texte)
self.assertNotIn("ligne 49", texte)
def test_showing_it_says_how_much_is_cut(self):
# « il manque des logs » venait de là : on montrait la fin sans
# dire qu'on cachait le début.
lst, total = ([f"l{i}" for i in range(400)], 12843)
self.assertLess(len(lst), total)
def test_the_tail_reports_the_total(self):
import tempfile
dossier = tempfile.mkdtemp(prefix="erplibre_essai_")
avant = os.getcwd()
os.chdir(dossier)
self.addCleanup(shutil.rmtree, dossier, True)
self.addCleanup(os.chdir, avant)
chemin = os.path.join("private", "odoo", "migration", "db", "step_log")
os.makedirs(chemin)
with open(
os.path.join(chemin, status.step_slug("2 - Update") + ".log"), "w"
) as handle:
handle.write("\n".join(f"ligne {i}" for i in range(1000)))
lst, total = status.step_log_tail(
{"config_database_name": "db"}, "2 - Update", lines=400
)
self.assertEqual(total, 1000)
self.assertEqual(len(lst), 400)
self.assertEqual(lst[-1], "ligne 999")
def test_a_step_without_a_log_reports_zero(self):
self.assertEqual(
status.step_log_tail({"config_database_name": "db"}, "9 - rien"),
([], 0),
)
class TestTheKeyboard(Base):
"""Ce que l'écran promet dans son pied de page doit exister."""
def app(self):
return tui.build_app(progression(), path="/un/chemin.json")
def touches(self):
return {
touche
for entree in self.app().BINDINGS
for touche in entree[0].split(",")
}
def test_every_promised_key_is_bound(self):
attendues = {"q", "escape", "r", "l", "p", "plus", "minus"}
self.assertTrue(attendues <= self.touches(), self.touches())
def test_each_binding_has_an_action_that_exists(self):
# Un raccourci annoncé dont l'action manque échoue à la frappe,
# c'est-à-dire au pire moment.
app = self.app()
for _touches, action, _libelle in app.BINDINGS:
self.assertTrue(hasattr(app, f"action_{action}"), action)
def test_the_width_is_bounded_on_both_sides(self):
# Une colonne de zéro ne se retrouve plus ; une qui mange tout
# l'écran ne laisse rien à lire.
app = self.app()
for _ in range(50):
app.left_width = max(
app.LEFT_MIN, min(app.LEFT_MAX, app.left_width - app.LEFT_STEP)
)
self.assertEqual(app.left_width, app.LEFT_MIN)
for _ in range(50):
app.left_width = max(
app.LEFT_MIN, min(app.LEFT_MAX, app.left_width + app.LEFT_STEP)
)
self.assertEqual(app.left_width, app.LEFT_MAX)
def test_refreshing_rereads_the_disk(self):
"""La migration ÉCRIT pendant qu'on regarde.
Sans cela il fallait fermer et rouvrir l'écran pour voir le palier
suivant — sur une migration de plusieurs heures, on le fait.
"""
import inspect
source = inspect.getsource(tui.build_app)
self.assertIn("status.read(self.path)", source)
self.assertIn("self.lst_row = rows(self.dct)", source)
def test_refreshing_without_a_path_does_nothing(self):
app = tui.build_app(progression(), path=None)
app.action_refresh() # ne doit pas lever
def test_the_log_toggle_flips(self):
app = self.app()
self.assertTrue(app.show_log)
class TestTheStatisticsScreenOffersIt(Base):
"""L'écran de statistiques répond « qu'a-t-on supprimé, et pourquoi ».