Mathieu Benoit
bcdc19b506
The screen could already replay from any step -- Enter on a row does it -- but that lived in a hint line under four named buttons. A capability you have to guess is not offered, and it was reported as missing. It now has a key, a button and a message, like the four others. The b key added earlier went to the line-by-line prompts during the run, not to this screen. Same letter on both, so going back is one key wherever you are. Two defects behind it. Replaying the chosen step actually SKIPPED it, resuming at the next one. And a replayed step did not rewind the variables it owns, so the second run started from the first one's leftovers. Checked in a simulated terminal: b focuses the table, two arrows and Enter return step 2, and the button focuses without choosing. --- FR --- L'écran savait déjà rejouer depuis n'importe quelle étape — Entrée sur une rangée le fait — mais cela vivait dans une ligne d'indication sous quatre boutons nommés. Une capacité qu'il faut deviner n'est pas offerte, et elle a été signalée manquante. Elle a maintenant une touche, un bouton et un message, comme les quatre autres. La touche b ajoutée plus tôt s'adressait aux invites ligne à ligne pendant l'exécution, pas à cet écran. Même lettre sur les deux : revenir en arrière est une seule touche, où qu'on soit. Deux défauts derrière cela. Rejouer l'étape choisie la SAUTAIT en fait, reprenant à la suivante. Et une étape rejouée ne remettait pas à zéro les variables qui lui appartiennent, si bien que la seconde exécution partait des restes de la première. Vérifié en terminal simulé : b donne le focus au tableau, deux flèches et Entrée ramènent l'étape 2, et le bouton met le focus sans choisir. Assisted-by: Claude Opus 5
439 lines
17 KiB
Python
439 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
|
|
|
"""L'écran de reprise doit dire ce qui a eu lieu, pas ce qu'il croit.
|
|
|
|
Une base venant d'une vieille version se met à jour AVANT la neutralisation :
|
|
l'outil le propose au tout début. Ce travail est celui de l'étape 2, fait plus
|
|
tôt — mais il n'était enregistré que sous `state_1_update_all`, et l'écran
|
|
cherchait `state_2_*`.
|
|
|
|
Deux conséquences, l'une visible et l'autre chère : l'étape restait « non
|
|
démarrée » alors qu'elle venait de tourner, et la reprise suivante relançait
|
|
`update_addons_all` sur une base déjà à jour.
|
|
"""
|
|
|
|
import unittest
|
|
|
|
from script.todo import todo_i18n
|
|
from script.todo.todo_upgrade import MIGRATION_STEP, TodoUpgrade
|
|
|
|
|
|
class StatusCase(unittest.TestCase):
|
|
def setUp(self):
|
|
# PAS set_lang() : il persiste la langue dans env_var.sh, suivi par
|
|
# git. On écrit la mémoïsation, et on la rend.
|
|
self.addCleanup(
|
|
setattr, todo_i18n, "_current_lang", todo_i18n._current_lang
|
|
)
|
|
todo_i18n._current_lang = "en"
|
|
|
|
|
|
class TestUpdateAllStatus(StatusCase):
|
|
def status(self, dct):
|
|
return TodoUpgrade.step_status(dct, 2)
|
|
|
|
def test_an_old_log_knows_only_the_early_flag(self):
|
|
# Le cas signalé : un journal écrit avant le correctif. L'étape a bien
|
|
# eu lieu, il ne faut pas la dire « non démarrée ».
|
|
icon, detail = self.status({"state_1_update_all": True})
|
|
self.assertEqual(icon, "✅")
|
|
self.assertIn("early", detail)
|
|
|
|
def test_a_new_log_says_when_it_happened(self):
|
|
icon, detail = self.status(
|
|
{
|
|
"state_1_update_all": True,
|
|
"state_2_update_all": True,
|
|
"state_2_done_early": True,
|
|
}
|
|
)
|
|
self.assertEqual(icon, "✅")
|
|
self.assertIn("early", detail)
|
|
|
|
def test_done_at_its_normal_place_says_only_done(self):
|
|
icon, detail = self.status({"state_2_update_all": True})
|
|
self.assertEqual(icon, "✅")
|
|
self.assertNotIn("early", detail)
|
|
|
|
def test_really_not_started(self):
|
|
icon, detail = self.status({"state_0_install_odoo": True})
|
|
self.assertEqual(icon, "⬜")
|
|
self.assertIn("not started", detail)
|
|
|
|
def test_the_other_steps_are_untouched(self):
|
|
# Le correctif ne vaut que pour l'étape 2 : les autres gardent leur
|
|
# règle, sinon un drapeau d'une étape en éclairerait une autre.
|
|
dct = {"state_1_update_all": True}
|
|
for step, _ in MIGRATION_STEP:
|
|
if step in (1, 2):
|
|
continue
|
|
icon, _ = TodoUpgrade.step_status(dct, step)
|
|
self.assertEqual(icon, "⬜", f"étape {step}")
|
|
|
|
|
|
class TestNeedsUpdateAll(unittest.TestCase):
|
|
"""Ce qui décide de relancer, ou non, une mise à jour de plusieurs heures."""
|
|
|
|
def test_nothing_done_yet(self):
|
|
self.assertTrue(TodoUpgrade.needs_update_all({}))
|
|
|
|
def test_its_own_flag_is_enough(self):
|
|
self.assertFalse(
|
|
TodoUpgrade.needs_update_all({"state_2_update_all": True})
|
|
)
|
|
|
|
def test_the_early_flag_survives_a_resume(self):
|
|
# LE défaut : la variable de session repart à False à la reprise, et
|
|
# seule la trace écrite subsiste. Sans la lire, on relançait tout.
|
|
self.assertFalse(
|
|
TodoUpgrade.needs_update_all({"state_1_update_all": True})
|
|
)
|
|
|
|
def test_the_session_variable_still_counts(self):
|
|
self.assertFalse(TodoUpgrade.needs_update_all({}, True))
|
|
|
|
def test_an_unrelated_flag_does_not_count(self):
|
|
self.assertTrue(
|
|
TodoUpgrade.needs_update_all({"state_1_restore_database": True})
|
|
)
|
|
|
|
|
|
class TestRewindReallyReplays(StatusCase):
|
|
"""Rembobiner doit REJOUER l'étape, pas seulement effacer sa trace.
|
|
|
|
Un journal venant d'une vieille base porte `state_1_update_all` : la mise
|
|
à jour précoce, offerte avant la neutralisation. Le travail est celui de
|
|
l'étape 2, seul le nom dit 1. Rembobiner à l'étape 2 gardait donc ce
|
|
drapeau — son préfixe le rangeait dans l'étape 1 — et l'étape sautait
|
|
alors qu'on venait de demander à la rejouer.
|
|
|
|
Effacer les clés ne suffit pas à le prouver : il faut interroger les
|
|
gardes qui décident du travail.
|
|
"""
|
|
|
|
def journal(self):
|
|
"""La forme d'un vrai journal ayant fait la mise à jour précoce."""
|
|
return {
|
|
"config_database_name": "db",
|
|
"state_0_install_odoo": True,
|
|
"state_0_search_missing_module": True,
|
|
"state_1_restore_database": True,
|
|
"state_1_update_all": True,
|
|
"state_1_neutralize_database": True,
|
|
"state_3_clean_database": True,
|
|
"state_4_upgrade_odoo_lst": [True, False],
|
|
}
|
|
|
|
def rewind(self, step):
|
|
import contextlib
|
|
import io
|
|
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
return TodoUpgrade.rewind_progression(self.journal(), step)
|
|
|
|
def test_going_back_to_step_2_runs_the_update_again(self):
|
|
# LE défaut signalé : « il continue en ignorant l'étape que j'ai
|
|
# choisie ». Trois heures de mise à jour silencieusement sautées.
|
|
self.assertTrue(
|
|
TodoUpgrade.needs_update_all(self.rewind(2)),
|
|
"rembobiner à l'étape 2 doit rejouer la mise à jour",
|
|
)
|
|
|
|
def test_the_earlier_steps_replay_it_too(self):
|
|
for step in (0, 1):
|
|
self.assertTrue(
|
|
TodoUpgrade.needs_update_all(self.rewind(step)),
|
|
f"étape {step}",
|
|
)
|
|
|
|
def test_going_back_to_a_later_step_leaves_it_done(self):
|
|
# Le symétrique : reculer à l'étape 3 ne doit PAS relancer l'étape 2.
|
|
for step in (3, 4):
|
|
self.assertFalse(
|
|
TodoUpgrade.needs_update_all(self.rewind(step)),
|
|
f"étape {step}",
|
|
)
|
|
|
|
def test_no_work_guard_of_the_chosen_step_or_later_survives(self):
|
|
# Les gardes sont lues dans la SOURCE : une garde ajoutée plus tard
|
|
# sera couverte sans que ce test soit retouché.
|
|
import ast
|
|
|
|
from script.todo import todo_upgrade
|
|
|
|
guards = set()
|
|
with open(todo_upgrade.__file__) as handle:
|
|
tree = ast.parse(handle.read())
|
|
for node in ast.walk(tree):
|
|
if not (
|
|
isinstance(node, ast.If) and isinstance(node.test, ast.UnaryOp)
|
|
):
|
|
continue
|
|
for sub in ast.walk(node.test):
|
|
if (
|
|
isinstance(sub, ast.Call)
|
|
and isinstance(sub.func, ast.Attribute)
|
|
and sub.func.attr == "get"
|
|
and sub.args
|
|
and isinstance(sub.args[0], ast.Constant)
|
|
and str(sub.args[0].value).startswith("state_")
|
|
):
|
|
guards.add(str(sub.args[0].value))
|
|
self.assertTrue(
|
|
guards, "aucune garde trouvée : le test ne prouve rien"
|
|
)
|
|
for step, _ in MIGRATION_STEP:
|
|
kept = self.rewind(step)
|
|
still_closed = [
|
|
key
|
|
for key in sorted(guards)
|
|
if todo_upgrade.flag_step(key) >= step and kept.get(key)
|
|
]
|
|
self.assertEqual(still_closed, [], f"étape {step}")
|
|
|
|
|
|
class TestEveryKeyBelongsToSomeone(unittest.TestCase):
|
|
"""Aucune clé du journal ne doit échapper au classement.
|
|
|
|
Rembobiner ne gardait que ce qui ne commençait pas par « state_ ». Vingt-
|
|
cinq clés passaient donc au travers, dont les réponses par palier de
|
|
l'étape 4 : l'étape choisie se rejouait, les suivantes repartaient sur les
|
|
décisions de la course précédente et ne demandaient plus rien.
|
|
|
|
Ce test lit les clés dans la SOURCE. Une clé ajoutée demain sans être
|
|
classée le fera échouer, au lieu de survivre en silence au rembobinage.
|
|
"""
|
|
|
|
def keys_in_source(self):
|
|
import ast
|
|
|
|
from script.todo import todo_upgrade
|
|
|
|
with open(todo_upgrade.__file__) as handle:
|
|
tree = ast.parse(handle.read())
|
|
keys = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Assign):
|
|
for target in node.targets:
|
|
if (
|
|
isinstance(target, ast.Subscript)
|
|
and isinstance(target.value, ast.Attribute)
|
|
and target.value.attr == "dct_progression"
|
|
and isinstance(target.slice, ast.Constant)
|
|
):
|
|
keys.add(str(target.slice.value))
|
|
if (
|
|
isinstance(node, ast.Call)
|
|
and isinstance(node.func, ast.Attribute)
|
|
and node.func.attr == "get"
|
|
and node.args
|
|
and isinstance(node.args[0], ast.Constant)
|
|
):
|
|
owner = node.func.value
|
|
name = getattr(owner, "attr", getattr(owner, "id", ""))
|
|
if "progression" in str(name):
|
|
keys.add(str(node.args[0].value))
|
|
return keys
|
|
|
|
def test_no_key_is_left_unclassified(self):
|
|
from script.todo import todo_upgrade
|
|
|
|
keys = self.keys_in_source()
|
|
self.assertGreater(len(keys), 30, "l'extraction n'a rien trouvé")
|
|
orphans = sorted(
|
|
key
|
|
for key in keys
|
|
if key not in todo_upgrade.GLOBAL_PROGRESSION_KEY
|
|
and todo_upgrade.flag_step(key) is None
|
|
)
|
|
self.assertEqual(
|
|
orphans, [], "à classer : global, ou à quelle étape ?"
|
|
)
|
|
|
|
def test_a_key_is_never_both_global_and_owned(self):
|
|
from script.todo import todo_upgrade
|
|
|
|
for key in sorted(todo_upgrade.GLOBAL_PROGRESSION_KEY):
|
|
self.assertIsNone(todo_upgrade.flag_step(key), key)
|
|
|
|
|
|
class TestRewindDropsTheAnswersToo(StatusCase):
|
|
"""« Il faut aussi revert les variables mises à jour entre les étapes. »"""
|
|
|
|
def journal(self):
|
|
return {
|
|
"config_database_name": "db",
|
|
"target_odoo_version": "18.0",
|
|
"migration_file": "a.zip",
|
|
"config_state_1_install_module": ["sale"],
|
|
"config_state_4_install_module": [False, ["stock"]],
|
|
"config_state_4_uninstall_module": [False, False],
|
|
"config_state_4_module_to_migrate_code": [[], ["m"]],
|
|
"dct_module_exist": {"sale": "1.0"},
|
|
"state_0_install_odoo": True,
|
|
"state_2_update_all": True,
|
|
}
|
|
|
|
def rewind(self, step):
|
|
import contextlib
|
|
import io
|
|
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
return TodoUpgrade.rewind_progression(self.journal(), step)
|
|
|
|
def test_the_step_4_answers_go_when_step_4_is_replayed(self):
|
|
# LE défaut signalé : l'étape rejouée marchait, les suivantes non.
|
|
kept = self.rewind(2)
|
|
for key in (
|
|
"config_state_4_install_module",
|
|
"config_state_4_uninstall_module",
|
|
"config_state_4_module_to_migrate_code",
|
|
):
|
|
self.assertNotIn(key, kept, key)
|
|
|
|
def test_the_step_1_answers_stay_when_step_1_is_kept(self):
|
|
# Le symétrique : rembobiner à 2 ne doit pas re-demander l'étape 1.
|
|
self.assertIn("config_state_1_install_module", self.rewind(2))
|
|
self.assertNotIn("config_state_1_install_module", self.rewind(1))
|
|
|
|
def test_the_step_0_module_search_output_goes_with_step_0(self):
|
|
# `dct_module_exist` ne nomme aucune étape mais est produit par la 0,
|
|
# et relu par la 4 : le garder après un rembobinage à 0 ferait juger
|
|
# la 4 sur un inventaire périmé.
|
|
self.assertNotIn("dct_module_exist", self.rewind(0))
|
|
self.assertIn("dct_module_exist", self.rewind(1))
|
|
|
|
def test_the_decisions_survive_every_rewind(self):
|
|
# Sans elles on ne saurait plus QUELLE base on migre : le rembobinage
|
|
# deviendrait une autre migration.
|
|
for step, _ in MIGRATION_STEP:
|
|
kept = self.rewind(step)
|
|
for key in ("config_database_name", "target_odoo_version"):
|
|
self.assertIn(key, kept, f"{key} à l'étape {step}")
|
|
|
|
|
|
class TestRewindOneVersionBump(unittest.TestCase):
|
|
"""Rejouer un palier de l'étape 4 doit aussi rendre ses réponses."""
|
|
|
|
def journal(self):
|
|
return {
|
|
"state_4_clone_odoo_lst": [True, True, True],
|
|
"state_4_uninstall_module": [True, True, False],
|
|
"config_state_4_install_module": [["sale"], ["stock"], False],
|
|
"config_state_4_module_to_migrate_code": [["m1"], ["m2"], []],
|
|
"state_4_lst_module_missing": ["a", "b"],
|
|
"state_4_len_lst_module_missing": 40,
|
|
}
|
|
|
|
def test_the_answers_of_the_replayed_bumps_are_cleared(self):
|
|
kept = TodoUpgrade.rewind_version_bump(self.journal(), 1)
|
|
self.assertEqual(
|
|
kept["config_state_4_install_module"][1:], [[], False]
|
|
)
|
|
self.assertEqual(kept["config_state_4_install_module"][0], ["sale"])
|
|
|
|
def test_a_list_entry_stays_a_list(self):
|
|
# Le code fait .append() sur cette entrée sans l'avoir revue : un
|
|
# False à sa place ferait planter le palier qu'on vient de rejouer.
|
|
kept = TodoUpgrade.rewind_version_bump(self.journal(), 1)
|
|
for entry in kept["config_state_4_module_to_migrate_code"]:
|
|
self.assertIsInstance(entry, list)
|
|
|
|
def test_a_flat_list_is_not_a_per_bump_list(self):
|
|
# `state_4_lst_module_missing` est une liste de modules, pas un
|
|
# tableau par palier : la tronquer perdrait l'inventaire.
|
|
kept = TodoUpgrade.rewind_version_bump(self.journal(), 1)
|
|
self.assertEqual(kept["state_4_lst_module_missing"], ["a", "b"])
|
|
self.assertEqual(kept["state_4_len_lst_module_missing"], 40)
|
|
|
|
def test_the_earlier_bumps_are_untouched(self):
|
|
kept = TodoUpgrade.rewind_version_bump(self.journal(), 2)
|
|
self.assertEqual(kept["state_4_clone_odoo_lst"], [True, True, False])
|
|
|
|
|
|
class TestBackGate(unittest.TestCase):
|
|
"""Revenir à une étape précédente depuis une invite en cours de route.
|
|
|
|
Ces invites ne demandaient qu'à continuer. S'apercevoir à ce moment-là
|
|
qu'une étape antérieure méritait un autre choix n'avait qu'une issue :
|
|
Ctrl+C, qui laisse la progression telle quelle et oblige à retrouver
|
|
l'écran de reprise. « b » fait le travail proprement.
|
|
"""
|
|
|
|
def setUp(self):
|
|
import os
|
|
import tempfile
|
|
|
|
self.addCleanup(os.chdir, os.getcwd())
|
|
os.chdir(tempfile.mkdtemp())
|
|
os.makedirs(".venv.erplibre", exist_ok=True)
|
|
self.addCleanup(
|
|
setattr, todo_i18n, "_current_lang", todo_i18n._current_lang
|
|
)
|
|
todo_i18n._current_lang = "en"
|
|
|
|
def gate(self, answers):
|
|
"""(a arrêté ?, valeur rendue, progression) après ces réponses."""
|
|
import builtins
|
|
import contextlib
|
|
import io
|
|
|
|
from script.todo.todo_upgrade import MigrationRewind
|
|
|
|
upgrade = TodoUpgrade.__new__(TodoUpgrade)
|
|
upgrade.dct_progression = {
|
|
"config_database_name": "db",
|
|
"state_0_install_odoo": True,
|
|
"state_1_restore_database": True,
|
|
"state_2_update_all": True,
|
|
"state_3_clean_database": True,
|
|
}
|
|
upgrade.lst_command_executed = []
|
|
seq = iter(answers)
|
|
original = builtins.input
|
|
builtins.input = lambda *a: next(seq)
|
|
self.addCleanup(setattr, builtins, "input", original)
|
|
stopped, returned = False, None
|
|
try:
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
returned = upgrade.ask_gate("? ")
|
|
except MigrationRewind:
|
|
stopped = True
|
|
return stopped, returned, upgrade.dct_progression
|
|
|
|
def test_a_normal_answer_passes_straight_through(self):
|
|
stopped, returned, _ = self.gate(["y"])
|
|
self.assertFalse(stopped)
|
|
self.assertEqual(returned, "y")
|
|
|
|
def test_b_then_a_step_rewinds_and_stops(self):
|
|
stopped, _, progression = self.gate(["b", "2"])
|
|
self.assertTrue(stopped)
|
|
# L'étape choisie et les suivantes sont effacées, les précédentes non.
|
|
self.assertIn("state_1_restore_database", progression)
|
|
self.assertNotIn("state_2_update_all", progression)
|
|
self.assertNotIn("state_3_clean_database", progression)
|
|
|
|
def test_cancelling_the_rewind_does_not_stop_the_migration(self):
|
|
# LE piège : renoncer au retour en arrière arrêtait quand même tout.
|
|
# On revient à la même invite, exactement là où l'on était.
|
|
stopped, returned, progression = self.gate(["b", "", "y"])
|
|
self.assertFalse(stopped)
|
|
self.assertEqual(returned, "y")
|
|
self.assertIn("state_3_clean_database", progression)
|
|
|
|
def test_an_unknown_step_is_not_a_rewind(self):
|
|
stopped, returned, progression = self.gate(["b", "zzz", ""])
|
|
self.assertFalse(stopped)
|
|
self.assertEqual(returned, "")
|
|
self.assertIn("state_3_clean_database", progression)
|
|
|
|
def test_the_answer_is_case_insensitive(self):
|
|
stopped, _, _ = self.gate(["B", "1"])
|
|
self.assertTrue(stopped)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|