diff --git a/script/todo/qemu_deploy.py b/script/todo/qemu_deploy.py index b6804ec..695e0b1 100644 --- a/script/todo/qemu_deploy.py +++ b/script/todo/qemu_deploy.py @@ -782,6 +782,39 @@ class QemuDeployMixin: orphans.append((name, path)) return orphans + def _qemu_offer_orphan_removal(self, names): + """Propose d'effacer les qcow2 restés seuls. Rend False si on renonce. + + Un disque sans VM définie vient d'une création interrompue : la VM + n'a jamais démarré, et le fichier ne porte donc rien. Il est tout de + même PROPOSÉ et non effacé d'office — le même nom peut désigner le + disque d'une VM retirée à la main, dont on voulait garder les données. + + Sans cet effacement, deploy_qemu refuse d'écraser et la création + échoue, après avoir fait attendre. + """ + orphans = self._qemu_orphan_disks(names) + if not orphans: + return True + items = [] + for _name, path in orphans: + try: + items.append((os.path.getsize(path), path)) + except OSError: + items.append((0, path)) + self._cleanup_delete_files( + t("Orphan disks that would fail the deployment"), + items, + t("Delete them and continue? (y/N): "), + ) + restants = self._qemu_orphan_disks(names) + if not restants: + return True + print(f"\n⚠ {t('Kept - the deployment of these VMs will FAIL:')}") + for name, _path in restants: + print(f" {name}") + return self._is_yes(input(t("Continue anyway? (y/N): "))) + def _qemu_confirm_collisions(self, existing, pending_names): """Signale les noms qui heurtent l'existant, et demande confirmation. @@ -793,17 +826,13 @@ class QemuDeployMixin: orphans = self._qemu_orphan_disks(pending_names) if not existing and not orphans: return True - print(f"\n⚠ {t('Name collisions detected')} :") - skipped = t("VM already defined - SKIPPED, nothing overwritten") - for name in existing: - print(f" {name:<28.28} {skipped}") - for name, path in orphans: - print( - f" {name:<28.28} " - f"{t('disk present without VM - deployment will FAIL')}" - ) - print(f" {'':<28} {path}") - print(f" {'':<28} {t('Remove it by hand, or rename the VM.')}") + if existing: + print(f"\n⚠ {t('Name collisions detected')} :") + skipped = t("VM already defined - SKIPPED, nothing overwritten") + for name in existing: + print(f" {name:<28.28} {skipped}") + if orphans: + return self._qemu_offer_orphan_removal(pending_names) return self._is_yes( input(f"{t('Continue despite these collisions? (y/N): ')}") ) @@ -1333,6 +1362,15 @@ class QemuDeployMixin: if spec is None: return if spec: # None = annulé, {} = repli sur la CLI + # Le formulaire signale les disques orphelins mais ne peut pas + # les effacer : le faire demande root, et une invite de mot de + # passe dans une application plein écran n'a nulle part où + # s'afficher. La proposition vient donc ici, terminal rendu. + if not self._qemu_offer_orphan_removal( + [vm["name"] for vm in spec.get("vms") or []] + ): + print(t("Cancelled.")) + return self._qemu_run_spec(spec) return diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index 1a93945..e91a4f5 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -4981,6 +4981,18 @@ TRANSLATIONS = { "fr": "puis se déconnecter et se reconnecter.", "en": "then log out and back in.", }, + "Orphan disks that would fail the deployment": { + "fr": "Disques orphelins qui feraient échouer le déploiement", + "en": "Orphan disks that would fail the deployment", + }, + "Delete them and continue? (y/N): ": { + "fr": "Les effacer et continuer ? (o/N) : ", + "en": "Delete them and continue? (y/N, default: no): ", + }, + "Kept - the deployment of these VMs will FAIL:": { + "fr": "Conservés — le déploiement de ces VM ÉCHOUERA :", + "en": "Kept - the deployment of these VMs will FAIL:", + }, "Full output:": { "fr": "Sortie complète :", "en": "Full output:", diff --git a/test/test_qemu_orphan_disk.py b/test/test_qemu_orphan_disk.py new file mode 100644 index 0000000..1c4562c --- /dev/null +++ b/test/test_qemu_orphan_disk.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python3 +# © 2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) +"""Un disque resté seul bloque la création : le proposer à l'effacement. + +Une création interrompue laisse son qcow2 sans VM définie. deploy_qemu refuse +alors d'écraser, et la création échoue APRÈS avoir fait attendre. Le +formulaire plein écran, lui, signale le disque mais ne peut pas l'effacer : +cela demande root, et une invite de mot de passe n'a nulle part où s'afficher +dans une application Textual. + +Ce que ces tests gardent : + +- l'effacement est PROPOSÉ, jamais fait d'office — le même nom peut désigner + le disque d'une VM retirée à la main, dont on voulait garder les données ; +- un refus n'enchaîne pas en silence vers l'échec : il redemande ; +- le chemin plein écran passe par la même proposition que le chemin en ligne. +""" + +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +sys.argv = ["todo.py"] +from script.todo.todo import TODO # noqa: E402 + + +class PropositionEffacement(unittest.TestCase): + def setUp(self): + self.todo = TODO.__new__(TODO) + self.efface = [] + self.todo._cleanup_delete_files = self._faux_effacement + + def _faux_effacement(self, title, items, prompt): + self.efface.append([p for _s, p in items]) + + def _avec_disques(self, presents, reponses=()): + """Simule des orphelins et rend (accepté, effacements demandés).""" + it = iter(reponses) + restants = {"v": list(presents)} + + def faux_orphans(names): + return [ + (n, f"/var/lib/libvirt/images/{n}.qcow2") + for n in names + if n in restants["v"] + ] + + def effacer(title, items, prompt): + self.efface.append([p for _s, p in items]) + restants["v"] = [] + + self.todo._qemu_orphan_disks = faux_orphans + self.todo._cleanup_delete_files = effacer + with mock.patch("builtins.input", lambda *a: next(it, "")), mock.patch( + "builtins.print" + ): + ok = self.todo._qemu_offer_orphan_removal(["vm-a", "vm-b"]) + return ok, self.efface + + def test_without_orphans_nothing_is_asked(self): + self.todo._qemu_orphan_disks = lambda names: [] + with mock.patch("builtins.print"): + self.assertTrue(self.todo._qemu_offer_orphan_removal(["vm-a"])) + self.assertEqual([], self.efface) + + def test_an_orphan_is_offered_for_deletion(self): + ok, efface = self._avec_disques(["vm-a"]) + self.assertTrue(ok) + self.assertEqual([["/var/lib/libvirt/images/vm-a.qcow2"]], efface) + + def test_a_kept_disk_asks_again_instead_of_failing(self): + """Refuser l'effacement mène à un échec certain : le dire et + redemander, plutôt qu'enchaîner en silence.""" + self.todo._qemu_orphan_disks = lambda names: [ + (n, f"/var/lib/libvirt/images/{n}.qcow2") + for n in names + if n == "vm-a" + ] + with mock.patch("builtins.input", return_value="n"), mock.patch( + "builtins.print" + ): + self.assertFalse( + self.todo._qemu_offer_orphan_removal(["vm-a", "vm-b"]) + ) + + def test_a_kept_disk_can_still_be_forced_through(self): + self.todo._qemu_orphan_disks = lambda names: [ + (n, f"/var/lib/libvirt/images/{n}.qcow2") + for n in names + if n == "vm-a" + ] + with mock.patch("builtins.input", return_value="o"), mock.patch( + "builtins.print" + ): + self.assertTrue(self.todo._qemu_offer_orphan_removal(["vm-a"])) + + def test_the_size_is_read_from_the_real_file(self): + """La taille annoncée est celle du fichier : un chiffre inventé + empêcherait de juger ce qu'on efface.""" + with tempfile.TemporaryDirectory() as tmp: + chemin = os.path.join(tmp, "vm-a.qcow2") + with open(chemin, "wb") as fh: + fh.write(b"x" * 4096) + self.todo._qemu_orphan_disks = lambda names: [("vm-a", chemin)] + vus = [] + self.todo._cleanup_delete_files = lambda t_, items, p: vus.extend( + items + ) + with mock.patch("builtins.input", return_value="n"), mock.patch( + "builtins.print" + ): + self.todo._qemu_offer_orphan_removal(["vm-a"]) + self.assertEqual([(4096, chemin)], vus) + + def test_an_unreadable_file_does_not_break_the_offer(self): + self.todo._qemu_orphan_disks = lambda names: [("vm-a", "/nulle/part")] + vus = [] + self.todo._cleanup_delete_files = lambda t_, items, p: vus.extend( + items + ) + with mock.patch("builtins.input", return_value="n"), mock.patch( + "builtins.print" + ): + self.todo._qemu_offer_orphan_removal(["vm-a"]) + self.assertEqual([(0, "/nulle/part")], vus) + + +class LeFormulaireYPasseAussi(unittest.TestCase): + def test_the_tui_path_calls_the_offer(self): + """Le formulaire avertit mais ne peut pas effacer : sans cet appel, + un F5 de plus mène droit à l'échec.""" + source = Path("script/todo/qemu_deploy.py").read_text(encoding="utf-8") + debut = source.index('if self._qemu_ask_ui() == "tui":') + fin = source.index("got = self._qemu_collect_vms_cli(mod)", debut) + self.assertIn("_qemu_offer_orphan_removal", source[debut:fin]) + + +if __name__ == "__main__": + unittest.main()