diff --git a/script/analyse/check_filestore.py b/script/analyse/check_filestore.py new file mode 100755 index 0000000..903127d --- /dev/null +++ b/script/analyse/check_filestore.py @@ -0,0 +1,395 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""Des pièces jointes sans fichier : lesquelles peut-on encore récupérer ? + +« 266 fichiers absents du filestore » ne dit pas quoi faire. La question +utile est ailleurs : combien sont PERDUS, et combien dorment quelque part +sur la machine en attendant qu'on les remette ? + +Mesuré sur une migration réelle : sur 266 absents, 262 étaient des images +engendrées par des modules — dont 235 drapeaux de pays dont le champ +n'existe même plus en 18 — et QUATRE étaient de vrais documents. Ces +quatre-là sont la réponse. Les 262 autres sont du bruit qu'il ne faut pas +confondre avec eux. + +Où l'outil cherche +------------------ +1. Les autres filestores de la machine. Une migration laisse une base par + palier ; un fichier perdu dans la 12 est souvent intact dans la 15, + régénéré en chemin par une mise à jour de module. +2. Les `filestore/` NICHÉS. `shutil.move(src, dst)` d'Odoo renomme quand + la destination n'existe pas et IMBRIQUE quand elle existe : une base + restaurée deux fois sous le même nom se retrouve avec + `filestore//filestore/xx/sha`, qu'Odoo ne lira jamais. Mesuré : + 1168 fichiers, 133 Mo, recopiés à l'identique dans les sept bases de + la chaîne par le clone. +3. Les sauvegardes `.zip`. Leur répertoire central se lit sans tout + décompresser. + +Ce qui n'est PAS récupérable +---------------------------- +Un fichier introuvable partout. On le dit alors franchement, avec le +modèle et l'enregistrement auxquels il se rattache, pour qu'on puisse +juger de la perte — et non « 266 fichiers absents », devant quoi il n'y +a rien à décider. + +Un cas mérite sa propre catégorie : la pièce jointe dont le CHAMP +n'existe plus. `res.country.image` était un binaire en 12 ; en 18 c'est +`image_url`, calculé. Les 235 lignes survivent en pointant vers un champ +disparu : rien ne les lit, rien ne les régénérera, et il n'y a rien à +récupérer. Les compter comme des pertes serait faux. + +Lecture seule de bout en bout. + +Codes de sortie : 0 rien d'irrécupérable, 1 des trouvailles, 2 échec. +""" + +import os +import subprocess +import sys +import zipfile + +sys.path.append( + os.path.normpath(os.path.join(os.path.dirname(__file__), "..", "..")) +) + +try: + from script.todo.todo_i18n import t +except Exception: # pragma: no cover - repli si i18n indisponible + + def t(key: str) -> str: + return key + + +SEP = "\x1f" +REPO_ROOT = os.path.normpath( + os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..") +) + +# L'ordre EST la gravité : ce qu'on ne peut pas récupérer se lit en premier. +VERDICTS = ("lost", "in_backup", "in_other_filestore", "nested", "dead_field") + +ICONE = { + "lost": "❌", + "in_backup": "📦", + "in_other_filestore": "🗂", + "nested": "↳", + "dead_field": "🕳", +} + +EXPLICATION = { + "lost": "nowhere to be found — truly lost", + "in_backup": "still in a backup zip", + "in_other_filestore": "intact in another database's filestore", + "nested": "stranded in a nested filestore Odoo never reads", + "dead_field": "its field no longer exists — nothing reads it", +} + + +def run_psql(database, sql): + """Interroger la base en lecture seule, garantie par le SERVEUR.""" + env = os.environ.copy() + env["PGOPTIONS"] = "-c default_transaction_read_only=on" + env["PSQLRC"] = "" + done = subprocess.run( + ["psql", "-X", "-w", "-d", database, "-tAF", SEP, "-c", sql], + capture_output=True, + text=True, + env=env, + ) + if done.returncode: + return None + return [ligne.split(SEP) for ligne in done.stdout.splitlines() if ligne] + + +def data_dir(config_path=None): + """Le `data_dir` d'Odoo, lu dans la configuration. + + Le deviner reviendrait à chercher au mauvais endroit et à déclarer + tout perdu — le pire diagnostic possible pour cet outil. + """ + chemin = config_path or os.path.join(REPO_ROOT, "config.conf") + try: + with open(chemin, "r", encoding="utf-8") as handle: + for ligne in handle: + if ligne.strip().startswith("data_dir"): + return ligne.split("=", 1)[1].strip() + except OSError: + pass + return os.path.expanduser("~/.local/share/Odoo") + + +def filestore_root(config_path=None): + return os.path.join(data_dir(config_path), "filestore") + + +def attachments(database): + """Les pièces jointes stockées sur disque. None si la base se tait.""" + lignes = run_psql( + database, + "SELECT a.store_fname, coalesce(a.res_model, '')," + " coalesce(a.res_field, ''), coalesce(a.res_id::text, '')," + " coalesce(a.name, ''), coalesce(a.file_size::text, '0')," + " coalesce(a.mimetype, ''), coalesce(a.create_date::text, '')" + " FROM ir_attachment a WHERE a.store_fname IS NOT NULL" + " ORDER BY a.file_size DESC", + ) + if lignes is None: + return None + return [ + { + "store_fname": ligne[0], + "model": ligne[1], + "field": ligne[2], + "res_id": ligne[3], + "name": ligne[4], + "size": int(ligne[5] or 0), + "mimetype": ligne[6], + "created": ligne[7][:10], + } + for ligne in lignes + if len(ligne) >= 8 + ] + + +def live_fields(database): + """Les champs qui EXISTENT encore, en « modele.champ ». + + Sans eux, une pièce jointe orpheline d'un champ supprimé passerait + pour une perte alors qu'il n'y a rien à perdre. + """ + lignes = run_psql( + database, "SELECT model || '.' || name FROM ir_model_fields" + ) + return {ligne[0] for ligne in lignes or [] if ligne and ligne[0]} + + +def scan_filestores(racine, sauf=None): + """{store_fname: base} pour toutes les bases, la nôtre exclue. + + Les fichiers NICHÉS sont indexés sous leur nom logique — c'est ce + qu'on cherchera — mais notés à part : les remettre en place est un + déplacement, pas une copie depuis ailleurs. + """ + ailleurs, niches = {}, {} + if not os.path.isdir(racine): + return ailleurs, niches + for base in sorted(os.listdir(racine)): + chemin = os.path.join(racine, base) + if not os.path.isdir(chemin): + continue + for prefixe in sorted(os.listdir(chemin)): + sous = os.path.join(chemin, prefixe) + if not os.path.isdir(sous): + continue + if prefixe == "filestore": + cible, marque = niches, base + for deux in sorted(os.listdir(sous)): + profond = os.path.join(sous, deux) + if not os.path.isdir(profond): + continue + for nom in os.listdir(profond): + cible.setdefault(f"{deux}/{nom}", marque) + continue + if base == sauf: + continue + for nom in os.listdir(sous): + ailleurs.setdefault(f"{prefixe}/{nom}", base) + return ailleurs, niches + + +def scan_backups(dossier): + """{store_fname: zip} en lisant le répertoire central, sans extraire.""" + trouves = {} + if not os.path.isdir(dossier): + return trouves + for nom in sorted(os.listdir(dossier)): + if not nom.endswith(".zip"): + continue + chemin = os.path.join(dossier, nom) + try: + with zipfile.ZipFile(chemin) as archive: + for membre in archive.namelist(): + if membre.startswith("filestore/") and not membre.endswith( + "/" + ): + trouves.setdefault(membre[len("filestore/") :], nom) + except (OSError, zipfile.BadZipFile): + continue + return trouves + + +def classify(piece, present, ailleurs, niches, sauvegardes, champs_vivants): + """Le verdict d'une pièce jointe. None si son fichier est là. + + L'ordre des questions est l'ordre de l'action à mener : d'abord + « y a-t-il seulement quelque chose à récupérer », puis « où ». + """ + if piece["store_fname"] in present: + return None + # Un champ disparu n'a rien à récupérer : la ligne est une scorie. + # Le tester EN PREMIER évite de proposer une remise en place inutile. + if piece["field"]: + cle = f"{piece['model']}.{piece['field']}" + if cle not in champs_vivants: + return ("dead_field", cle) + if piece["store_fname"] in niches: + return ("nested", niches[piece["store_fname"]]) + if piece["store_fname"] in ailleurs: + return ("in_other_filestore", ailleurs[piece["store_fname"]]) + if piece["store_fname"] in sauvegardes: + return ("in_backup", sauvegardes[piece["store_fname"]]) + return ("lost", None) + + +def audit(database, config_path=None, backups=None): + """Tout ce qu'il faut savoir, en une passe. Lecture seule.""" + pieces = attachments(database) + if pieces is None: + return {"unavailable": True, "database": database} + racine = filestore_root(config_path) + mien = os.path.join(racine, database) + present = set() + if os.path.isdir(mien): + for prefixe in os.listdir(mien): + sous = os.path.join(mien, prefixe) + if prefixe == "filestore" or not os.path.isdir(sous): + continue + for nom in os.listdir(sous): + if os.path.isfile(os.path.join(sous, nom)): + present.add(f"{prefixe}/{nom}") + ailleurs, niches = scan_filestores(racine, sauf=database) + sauvegardes = scan_backups(backups or os.path.join(REPO_ROOT, "image_db")) + champs_vivants = live_fields(database) + + groupes = {verdict: [] for verdict in VERDICTS} + vus = set() + for piece in pieces: + verdict = classify( + piece, present, ailleurs, niches, sauvegardes, champs_vivants + ) + if not verdict: + continue + # Plusieurs pièces jointes partagent un fichier quand leur contenu + # est identique : le compter une fois par ligne gonflerait le + # rapport sans ajouter un seul fichier à récupérer. + if piece["store_fname"] in vus: + continue + vus.add(piece["store_fname"]) + groupes[verdict[0]].append(dict(piece, where=verdict[1])) + return { + "database": database, + "attachments": len(pieces), + "files_present": len(present), + "missing": len(vus), + "groups": groupes, + "nested_total": len(niches), + "root": racine, + } + + +def render(rapport, limit=20): + if rapport.get("unavailable"): + return [f"❌ {t('Cannot read the database: ')}{rapport['database']}"] + lignes = [ + f"🗄 {t('Filestore of')} {rapport['database']} :" + f" {rapport['attachments']} {t('stored attachment(s)')}," + f" {rapport['files_present']} {t('file(s) on disk')}" + ] + if not rapport["missing"]: + lignes.append(f" ✅ {t('every attachment file is present')}") + return lignes + render_nested(rapport) + lignes.append(f" {rapport['missing']} {t('file(s) missing')} :") + for verdict in VERDICTS: + groupe = rapport["groups"][verdict] + if not groupe: + continue + poids = sum(piece["size"] for piece in groupe) + lignes.append( + f" {ICONE[verdict]} {len(groupe)} {t(EXPLICATION[verdict])}" + f" ({poids // 1024} ko)" + ) + # On ne nomme QUE l'irrécupérable : c'est la seule liste devant + # laquelle il y a une décision à prendre. Nommer les autres + # ferait défiler trois cents lignes sans rien apprendre. + if verdict != "lost": + apercu = summarise(groupe) + for texte in apercu[:4]: + lignes.append(f" {texte}") + if len(apercu) > 4: + lignes.append(f" … {len(apercu) - 4} {t('more')}") + continue + for piece in groupe[:limit]: + ou = ( + f"{piece['model']} #{piece['res_id']}" + if piece["model"] + else "—" + ) + champ = f" [{piece['field']}]" if piece["field"] else "" + lignes.append( + f" {piece['name'][:44] or '(sans nom)':<46}" + f" {piece['size'] // 1024:>6} ko {ou}{champ}" + f" {piece['created']}" + ) + if len(groupe) > limit: + lignes.append(f" … {len(groupe) - limit} {t('more')}") + return lignes + render_nested(rapport) + + +def render_nested(rapport): + if not rapport.get("nested_total"): + return [] + return [ + "", + f" ↳ {rapport['nested_total']}" + f" {t('file(s) sit in nested filestores Odoo never reads.')}", + ] + + +def summarise(groupe): + """« modele / champ × N », pour dire beaucoup en peu de lignes.""" + compte = {} + for piece in groupe: + cle = f"{piece['model'] or '—'} / {piece['field'] or '—'}" + compte[cle] = compte.get(cle, 0) + 1 + return [ + f"{cle} × {combien}" if combien > 1 else cle + for cle, combien in sorted(compte.items(), key=lambda x: -x[1]) + ] + + +def main(argv=None): + import argparse + + parser = argparse.ArgumentParser( + description=( + "List attachments whose file is gone, and say which ones can" + " still be recovered and from where." + ) + ) + parser.add_argument("-d", "--database", required=True) + parser.add_argument("-c", "--config", help="odoo config file") + parser.add_argument("--backups", help="directory of backup .zip files") + parser.add_argument("--limit", type=int, default=20) + parser.add_argument("--json", action="store_true") + config = parser.parse_args(argv) + + rapport = audit(config.database, config.config, config.backups) + if rapport.get("unavailable"): + print(f"❌ {t('Cannot read the database: ')}{config.database}") + return 2 + if config.json: + import json + + print( + json.dumps(rapport, indent=2, sort_keys=True, ensure_ascii=False) + ) + else: + print("\n".join(render(rapport, config.limit))) + return 1 if rapport["groups"]["lost"] else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test/test_check_filestore.py b/test/test_check_filestore.py new file mode 100644 index 0000000..d00fdbe --- /dev/null +++ b/test/test_check_filestore.py @@ -0,0 +1,365 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""Séparer ce qui est perdu de ce qui dort quelque part. + +Deux propriétés portent l'outil. La première : un fichier retrouvé +ailleurs n'est PAS une perte, et le dire évite de pleurer sur 266 +fichiers quand trois seulement ont disparu. La seconde : une pièce jointe +dont le champ n'existe plus n'a rien à récupérer — la ranger avec les +pertes serait faux dans l'autre sens. +""" + +import io +import os +import shutil +import sys +import tempfile +import unittest +import zipfile +from contextlib import redirect_stdout + +sys.path.append( + os.path.normpath(os.path.join(os.path.dirname(__file__), "..")) +) + +from script.analyse import check_filestore as fs # noqa: E402 +from script.todo import todo_i18n # noqa: E402 + + +def piece(store, model="", field="", res_id="1", name="x", size=1024): + return { + "store_fname": store, + "model": model, + "field": field, + "res_id": res_id, + "name": name, + "size": size, + "mimetype": "image/png", + "created": "2019-11-30", + } + + +class TestClassify(unittest.TestCase): + VIVANTS = {"res.partner.image_1920"} + + def juge(self, p, present=(), ailleurs=None, niches=None, zips=None): + return fs.classify( + p, + set(present), + ailleurs or {}, + niches or {}, + zips or {}, + self.VIVANTS, + ) + + def test_a_present_file_is_not_a_finding(self): + self.assertIsNone(self.juge(piece("aa/bb"), present=["aa/bb"])) + + def test_a_file_found_nowhere_is_lost(self): + self.assertEqual(self.juge(piece("aa/bb")), ("lost", None)) + + def test_a_file_in_another_filestore_is_recoverable(self): + self.assertEqual( + self.juge(piece("aa/bb"), ailleurs={"aa/bb": "autre_base"}), + ("in_other_filestore", "autre_base"), + ) + + def test_a_file_in_a_backup_is_recoverable(self): + self.assertEqual( + self.juge(piece("aa/bb"), zips={"aa/bb": "sauvegarde.zip"}), + ("in_backup", "sauvegarde.zip"), + ) + + def test_a_nested_file_is_named_as_such(self): + # Le remettre en place est un DÉPLACEMENT, pas une copie depuis + # une autre base : le geste diffère, la catégorie aussi. + self.assertEqual( + self.juge(piece("aa/bb"), niches={"aa/bb": "ma_base"}), + ("nested", "ma_base"), + ) + + def test_a_dead_field_is_judged_BEFORE_looking_anywhere(self): + # Rien ne lit cette ligne : proposer de la récupérer ferait + # travailler pour rien. Testé d'abord, donc, même si le fichier + # traîne dans une sauvegarde. + verdict = self.juge( + piece("aa/bb", model="res.country", field="image"), + zips={"aa/bb": "sauvegarde.zip"}, + ) + self.assertEqual(verdict[0], "dead_field") + + def test_a_living_field_is_not_mistaken_for_a_dead_one(self): + verdict = self.juge( + piece("aa/bb", model="res.partner", field="image_1920") + ) + self.assertEqual(verdict, ("lost", None)) + + def test_an_attachment_without_a_field_is_never_dead(self): + # Un document téléversé n'a pas de `res_field` : le juger sur un + # champ absent le ferait disparaître du rapport. + self.assertEqual( + self.juge(piece("aa/bb", model="project.task"))[0], "lost" + ) + + def test_every_verdict_has_an_icon_and_a_wording(self): + for verdict in fs.VERDICTS: + self.assertIn(verdict, fs.ICONE) + self.assertIn(verdict, fs.EXPLICATION) + + def test_the_lost_come_first_in_the_order(self): + # L'ordre EST la gravité : ce qu'on ne peut pas récupérer se lit + # d'abord. + self.assertEqual(fs.VERDICTS[0], "lost") + + +class TestTheDataDir(unittest.TestCase): + def test_it_is_read_from_the_config(self): + with tempfile.NamedTemporaryFile( + "w", suffix=".conf", delete=False, encoding="utf-8" + ) as handle: + handle.write("[options]\ndata_dir = /chemin/a/moi\n") + chemin = handle.name + try: + self.assertEqual(fs.data_dir(chemin), "/chemin/a/moi") + finally: + os.unlink(chemin) + + def test_a_missing_config_falls_back_instead_of_crashing(self): + # Deviner mal ferait déclarer TOUT perdu : le pire diagnostic. + self.assertTrue(fs.data_dir("/nulle/part.conf")) + + +class TestScanning(unittest.TestCase): + def setUp(self): + self.racine = tempfile.mkdtemp() + for base, chemin in ( + ("ma_base", "aa/present"), + ("autre", "bb/ailleurs"), + ("ma_base", "filestore/cc/niche"), + ): + complet = os.path.join(self.racine, base, chemin) + os.makedirs(os.path.dirname(complet), exist_ok=True) + with open(complet, "w", encoding="utf-8") as handle: + handle.write("x") + + def tearDown(self): + shutil.rmtree(self.racine) + + def test_other_filestores_are_indexed_and_mine_is_skipped(self): + ailleurs, _n = fs.scan_filestores(self.racine, sauf="ma_base") + self.assertEqual(ailleurs.get("bb/ailleurs"), "autre") + self.assertNotIn("aa/present", ailleurs) + + def test_nested_files_are_indexed_under_their_logical_name(self): + # C'est sous « cc/niche » qu'on les cherchera, pas sous + # « filestore/cc/niche ». + _a, niches = fs.scan_filestores(self.racine, sauf=None) + self.assertEqual(niches.get("cc/niche"), "ma_base") + + def test_a_missing_root_is_empty_not_a_crash(self): + self.assertEqual(fs.scan_filestores("/nulle/part"), ({}, {})) + + def test_backups_are_read_from_the_central_directory(self): + dossier = tempfile.mkdtemp() + try: + chemin = os.path.join(dossier, "sauv.zip") + with zipfile.ZipFile(chemin, "w") as archive: + archive.writestr("filestore/dd/dedans", "x") + archive.writestr("dump.sql", "x") + trouves = fs.scan_backups(dossier) + self.assertEqual(trouves.get("dd/dedans"), "sauv.zip") + self.assertNotIn("dump.sql", trouves) + finally: + shutil.rmtree(dossier) + + def test_a_corrupt_zip_is_skipped_not_fatal(self): + dossier = tempfile.mkdtemp() + try: + with open(os.path.join(dossier, "casse.zip"), "w") as handle: + handle.write("pas un zip") + self.assertEqual(fs.scan_backups(dossier), {}) + finally: + shutil.rmtree(dossier) + + +class TestTheAudit(unittest.TestCase): + """`audit` sans PostgreSQL ni disque : on éprouve la LOGIQUE.""" + + def setUp(self): + self.vrais = ( + fs.attachments, + fs.live_fields, + fs.scan_filestores, + fs.scan_backups, + fs.filestore_root, + ) + fs.live_fields = lambda base: set() + fs.scan_filestores = lambda racine, sauf=None: ({}, {}) + fs.scan_backups = lambda dossier: {} + fs.filestore_root = lambda config=None: "/nulle/part" + + def tearDown(self): + ( + fs.attachments, + fs.live_fields, + fs.scan_filestores, + fs.scan_backups, + fs.filestore_root, + ) = self.vrais + + def test_two_attachments_sharing_a_file_count_once(self): + # Odoo déduplique par empreinte : deux pièces jointes au contenu + # identique partagent un `store_fname`. Les compter deux fois + # gonflerait « fichiers absents » sans qu'un seul fichier de plus + # soit à retrouver. + fs.attachments = lambda base: [ + piece("aa/bb", "project.task", res_id="1"), + piece("aa/bb", "project.task", res_id="2"), + ] + rapport = fs.audit("db") + self.assertEqual(rapport["missing"], 1) + self.assertEqual(len(rapport["groups"]["lost"]), 1) + + def test_distinct_files_are_counted_apart(self): + fs.attachments = lambda base: [piece("aa/bb"), piece("cc/dd")] + self.assertEqual(fs.audit("db")["missing"], 2) + + def test_a_silent_database_is_unavailable(self): + fs.attachments = lambda base: None + self.assertTrue(fs.audit("db")["unavailable"]) + + def test_a_dead_field_never_lands_in_lost(self): + # Le garde du champ disparu doit VRAIMENT couper : sans lui, ces + # lignes gonfleraient les pertes réelles. + fs.attachments = lambda base: [piece("aa/bb", "res.country", "image")] + rapport = fs.audit("db") + self.assertEqual(rapport["groups"]["lost"], []) + self.assertEqual(len(rapport["groups"]["dead_field"]), 1) + + +class TestTheReport(unittest.TestCase): + def rapport(self, **extra): + base = { + "database": "db", + "attachments": 10, + "files_present": 8, + "missing": 0, + "nested_total": 0, + "root": "/x", + "groups": {v: [] for v in fs.VERDICTS}, + } + base.update(extra) + return base + + def test_a_clean_filestore_says_so(self): + texte = "\n".join(fs.render(self.rapport())) + self.assertIn(todo_i18n.t("every attachment file is present"), texte) + + def test_only_the_lost_are_named_one_by_one(self): + groupes = {v: [] for v in fs.VERDICTS} + groupes["lost"] = [piece("a/1", "project.task", name="perdu.png")] + groupes["in_backup"] = [ + piece("b/1", "res.country", "image", name="recuperable.png") + ] + texte = "\n".join(fs.render(self.rapport(missing=2, groups=groupes))) + self.assertIn("perdu.png", texte) + # Le récupérable est RÉSUMÉ, pas listé : trois cents lignes de + # noms qu'on n'a pas à lire cacheraient les trois qui comptent. + self.assertNotIn("recuperable.png", texte) + self.assertIn("res.country / image", texte) + + def test_the_nested_pile_is_reported(self): + texte = "\n".join(fs.render(self.rapport(nested_total=1168))) + self.assertIn("1168", texte) + + def test_an_unreadable_database_renders_without_crashing(self): + texte = "\n".join(fs.render({"unavailable": True, "database": "x"})) + self.assertIn("x", texte) + + def test_summarise_groups_by_model_and_field(self): + resume = fs.summarise( + [ + piece("a/1", "res.country", "image"), + piece("a/2", "res.country", "image"), + piece("a/3", "project.task"), + ] + ) + self.assertIn("res.country / image × 2", resume) + + +class TestTheExitCodes(unittest.TestCase): + def setUp(self): + self.vrai = fs.audit + + def tearDown(self): + fs.audit = self.vrai + + def lance(self, rapport): + fs.audit = lambda *a, **k: rapport + tampon = io.StringIO() + with redirect_stdout(tampon): + code = fs.main(["-d", "db"]) + return code, tampon.getvalue() + + def propre(self, **extra): + base = { + "database": "db", + "attachments": 1, + "files_present": 1, + "missing": 0, + "nested_total": 0, + "root": "/x", + "groups": {v: [] for v in fs.VERDICTS}, + } + base.update(extra) + return base + + def test_nothing_lost_exits_zero(self): + code, _ = self.lance(self.propre()) + self.assertEqual(code, 0) + + def test_recoverable_only_still_exits_zero(self): + # Rien à décider : tout se récupère. Sortir 1 ferait échouer une + # chaîne make pour une situation saine. + groupes = {v: [] for v in fs.VERDICTS} + groupes["in_backup"] = [piece("a/1")] + code, _ = self.lance(self.propre(missing=1, groups=groupes)) + self.assertEqual(code, 0) + + def test_a_real_loss_exits_one(self): + groupes = {v: [] for v in fs.VERDICTS} + groupes["lost"] = [piece("a/1")] + code, _ = self.lance(self.propre(missing=1, groups=groupes)) + self.assertEqual(code, 1) + + def test_an_unreadable_database_exits_two(self): + code, _ = self.lance({"unavailable": True, "database": "db"}) + self.assertEqual(code, 2) + + +class TestTranslations(unittest.TestCase): + def test_every_key_exists(self): + import ast + + with io.open(fs.__file__, encoding="utf-8") as handle: + arbre = ast.parse(handle.read()) + cles = set(fs.EXPLICATION.values()) + for node in ast.walk(arbre): + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "t" + and node.args + and isinstance(node.args[0], ast.Constant) + ): + cles.add(node.args[0].value) + for cle in cles: + self.assertTrue( + cle in todo_i18n.TRANSLATIONS, f"clé sans traduction : {cle!r}" + ) + + +if __name__ == "__main__": + unittest.main()