diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index b4e69f7..0447c9d 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -3860,6 +3860,14 @@ TRANSLATIONS = { " module source. The classification above needs none; only the" " differences do. Restore it, or run this on the database.", }, + "Could not archive the previous log": { + "fr": "Impossible d'archiver le journal précédent", + "en": "Could not archive the previous log", + }, + "Previous migration log kept in": { + "fr": "Journal de la migration précédente conservé dans", + "en": "Previous migration log kept in", + }, # --- script/analyse : vues personnalisées --- "Customised views": { "fr": "Vues personnalisées", diff --git a/script/todo/todo_upgrade.py b/script/todo/todo_upgrade.py index f4ade39..5e4fab2 100755 --- a/script/todo/todo_upgrade.py +++ b/script/todo/todo_upgrade.py @@ -13,8 +13,8 @@ import zipfile from uuid import uuid4 import click -import todo_file_browser +from script.todo import todo_file_browser from script.todo.version_manager import get_odoo_version try: @@ -229,6 +229,70 @@ class TodoUpgrade: print(f" [r] {t('Keep the zip only, ask every question again')}") print(f" [q] {t('Quit without doing anything')}") + @staticmethod + def archive_progression(old_dct_progression, reason): + """Mettre le journal de côté avant qu'une nouvelle migration l'écrase. + + Le journal ne vit qu'à un seul endroit et la migration suivante écrit + par-dessus : recommencer effaçait donc tout ce qu'on savait de la + précédente — quels paliers étaient passés, quels modules manquaient, + combien de temps chacun avait pris. Autant d'éléments qu'on ne cherche + justement qu'APRÈS avoir dû recommencer. + + Le nom porte la base d'origine et l'horodatage de la copie, pour que + deux tentatives sur la même base ne se recouvrent pas et qu'un fichier + déplacé se décrive encore lui-même. Renvoie le chemin, ou None s'il n'y + avait rien qui vaille d'être gardé. + + L'archive vit sous `private/`, pas dans le venv : une réinstallation + efface `.venv.erplibre/`, et avec elle l'historique qu'on vient de + sauver. + """ + if not old_dct_progression: + return None + # Un journal qui ne porte aucun état n'apprend rien à personne. + if not any(k.startswith("state_") for k in old_dct_progression): + return None + + database_name = ( + old_dct_progression.get("config_database_name") + or os.path.splitext( + os.path.basename( + old_dct_progression.get("migration_file") or "" + ) + )[0] + or "unknown" + ) + stamp = datetime.datetime.now().strftime("%Y-%m-%d_%Hh%Mm%Ss") + directory = os.path.join( + PATH_MIGRATION_PRIVATE, database_name, "migration_log" + ) + path = os.path.join(directory, f"{database_name}_{stamp}.json") + # Deux copies dans la même seconde portent le même horodatage. Écraser + # la première annulerait exactement la perte qu'on cherche à éviter, + # et sans rien dire. + suffix = 2 + while os.path.exists(path): + path = os.path.join( + directory, f"{database_name}_{stamp}_{suffix}.json" + ) + suffix += 1 + payload = dict(old_dct_progression) + payload["archived_at"] = str(datetime.datetime.now()) + payload["archived_reason"] = reason + payload["archived_database"] = database_name + try: + os.makedirs(directory, exist_ok=True) + with open(path, "w") as handle: + json.dump(payload, handle, indent=4) + except OSError as exc: + # Ne jamais empêcher une migration de repartir pour une histoire + # de copie : on le dit, et on continue. + print(f"⚠️ {t('Could not archive the previous log')} : {exc}") + return None + print(f"📦 {t('Previous migration log kept in')} : {path}") + return path + def apply_resume_answer(self, old_dct_progression, answer, ctx): """Turn the answer into (progression, changed), or None to quit. @@ -242,9 +306,15 @@ class TodoUpgrade: return old_dct_progression, False if answer == "q": return None + # « n » et « r » repartent de zéro : dans les deux cas la progression + # enregistrée disparaît, donc dans les deux cas on la garde d'abord. if answer == "n": + self.archive_progression(old_dct_progression, "restart_from_zero") return {}, True if answer == "r": + self.archive_progression( + old_dct_progression, "restart_same_backup" + ) return { "migration_file": old_dct_progression.get("migration_file"), "date_create": old_dct_progression.get("date_create"), diff --git a/test/test_todo_upgrade_archive.py b/test/test_todo_upgrade_archive.py new file mode 100644 index 0000000..0bd445c --- /dev/null +++ b/test/test_todo_upgrade_archive.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""Le journal de migration doit survivre à la migration suivante. + +Il ne vit qu'à un seul endroit, et repartir de zéro écrit par-dessus. Ce qu'on +perdait — quels paliers étaient passés, quels modules manquaient — est +précisément ce qu'on cherche APRÈS avoir dû recommencer. + +Ces tests portent sur deux choses distinctes : que la copie contienne bien ce +qu'elle doit contenir, et qu'elle parte au bon moment — ni sur une reprise +partielle, qui ne perd rien, ni sur un journal qui n'a rien à dire. +""" + +import glob +import json +import os +import tempfile +import unittest + +from script.todo.todo_upgrade import TodoUpgrade + + +def progression(**override): + """Un journal ayant réellement avancé.""" + data = { + "date_create": "2026-08-11 12:05:08", + "date_update": "2026-08-12 01:13:23", + "migration_file": "./image_db/technolibre_2026-08-12_02h36m38s.zip", + "config_database_name": "technolibre_migration_01_neutralize", + "state_0_install_odoo": True, + "state_1_restore_database": True, + "state_4_switch_odoo_lst": [True, False, False], + } + data.update(override) + return data + + +class ArchiveCase(unittest.TestCase): + def setUp(self): + # Chaque test dans un arbre neuf : l'archive s'écrit sous un chemin + # relatif, donc le répertoire courant EST le contexte. + self.previous = os.getcwd() + self.addCleanup(os.chdir, self.previous) + os.chdir(tempfile.mkdtemp()) + + def archives(self): + return sorted( + glob.glob("private/odoo/migration/*/migration_log/*.json") + ) + + +class TestWhatIsKept(ArchiveCase): + def test_the_whole_log_is_copied_untouched(self): + old = progression() + path = TodoUpgrade.archive_progression(old, "restart_from_zero") + saved = json.load(open(path)) + for key, value in old.items(): + self.assertEqual(saved[key], value, key) + + def test_the_copy_says_when_and_from_which_database(self): + # La demande, mot pour mot : une date de la copie et le nom de la base + # d'origine. + path = TodoUpgrade.archive_progression( + progression(), "restart_from_zero" + ) + saved = json.load(open(path)) + self.assertEqual( + saved["archived_database"], "technolibre_migration_01_neutralize" + ) + self.assertTrue(saved["archived_at"].startswith("20")) + self.assertEqual(saved["archived_reason"], "restart_from_zero") + + def test_the_file_name_carries_both_too(self): + # Pour qu'un fichier déplacé hors de son dossier se décrive encore. + path = TodoUpgrade.archive_progression(progression(), "x") + name = os.path.basename(path) + self.assertTrue( + name.startswith("technolibre_migration_01_neutralize_") + ) + self.assertRegex(name, r"_\d{4}-\d{2}-\d{2}_\d{2}h\d{2}m\d{2}s\.json$") + + def test_it_lands_under_private_not_in_the_venv(self): + # Une réinstallation efface .venv.erplibre/, donc y garder l'historique + # reviendrait à le perdre au pire moment. + path = TodoUpgrade.archive_progression(progression(), "x") + self.assertTrue(path.startswith(os.path.join("private", "odoo"))) + self.assertNotIn(".venv", path) + + def test_the_database_name_falls_back_to_the_backup_name(self): + old = progression() + del old["config_database_name"] + path = TodoUpgrade.archive_progression(old, "x") + self.assertIn("technolibre_2026-08-12_02h36m38s", path) + + def test_two_restarts_in_the_same_second_do_not_overwrite_each_other(self): + # L'horodatage est à la seconde : deux copies rapprochées portent le + # même nom. Écraser la première annulerait la perte qu'on évite — et + # un test qui saute ce cas ne vérifie rien. + first = TodoUpgrade.archive_progression(progression(), "x") + second = TodoUpgrade.archive_progression( + progression(date_update="2026-08-12 02:00:00"), "y" + ) + self.assertNotEqual(first, second) + self.assertEqual(len(self.archives()), 2) + self.assertEqual( + json.load(open(first))["date_update"], "2026-08-12 01:13:23" + ) + self.assertEqual( + json.load(open(second))["date_update"], "2026-08-12 02:00:00" + ) + + +class TestWhenNothingIsKept(ArchiveCase): + def test_an_empty_log_is_not_archived(self): + self.assertIsNone(TodoUpgrade.archive_progression({}, "x")) + self.assertEqual(self.archives(), []) + + def test_a_log_without_any_state_is_not_archived(self): + # Un journal qui n'a rien enregistré n'apprend rien à personne ; + # l'archiver ne ferait qu'accumuler des fichiers vides. + self.assertIsNone( + TodoUpgrade.archive_progression({"migration_file": "a.zip"}, "x") + ) + self.assertEqual(self.archives(), []) + + +class TestWhenItTriggers(ArchiveCase): + """Seules les réponses qui repartent de zéro archivent.""" + + def answer(self, letter): + upgrade = TodoUpgrade.__new__(TodoUpgrade) + result = TodoUpgrade.apply_resume_answer( + upgrade, progression(), letter, {"versions": []} + ) + return result, self.archives() + + def test_restart_from_zero_archives(self): + result, archives = self.answer("n") + self.assertEqual(result, ({}, True)) + self.assertEqual(len(archives), 1) + + def test_restart_with_the_same_backup_archives(self): + # « r » garde le zip mais jette tous les états : la perte est la même. + result, archives = self.answer("r") + self.assertTrue(result[1]) + self.assertNotIn("state_0_install_odoo", result[0]) + self.assertEqual(len(archives), 1) + + def test_continuing_archives_nothing(self): + result, archives = self.answer("c") + self.assertEqual(archives, []) + self.assertIn("state_0_install_odoo", result[0]) + + def test_rewinding_to_a_step_archives_nothing(self): + # Une reprise partielle garde le journal : rien n'est perdu, donc rien + # n'est à sauver. + _, archives = self.answer("0") + self.assertEqual(archives, []) + + +if __name__ == "__main__": + unittest.main()