Mathieu Benoit
4e0596721c
Deduplicating by file was right for COUNTING and wrong for DELETING: twenty-two rows shared two files, so the purge only ever offered one at a time and had to be replayed. It now gathers every row whose own field is dead, and never a live row sharing the same file. "root" held the root of all filestores instead of this database's directory, so tidying looked for a nested folder at <data_dir>/filestore/filestore and answered "nothing to tidy" in front of 1168 stranded files. It now finds 112 to move up and 1056 duplicates. limit=0 means "no cap" everywhere else, but the slice [:0] is empty: "Show every entry" printed "… 3 more" and showed none of them. The report was captured once, so after a purge it replayed the state from before -- one then purged rows already gone, believing the work unfinished. A repair now says whether it did something, and the report is re-read when it did. "DELETE 0" was announced as a success. The count now comes from what PostgreSQL said, and saying nothing is not the same as deleting nothing. --- FR --- Dédupliquer par fichier était juste pour COMPTER et faux pour EFFACER : vingt-deux lignes partageaient deux fichiers, la purge n'en offrait qu'une à la fois. Elle prend maintenant toutes les lignes dont le champ est mort, et jamais une ligne vivante partageant le même fichier. « root » portait la racine de tous les filestores au lieu du dossier de la base : le rangement cherchait à <data_dir>/filestore/filestore et répondait « rien à ranger » devant 1168 fichiers échoués. Il en trouve 112 à remonter et 1056 doublons. limit=0 veut dire « tout » partout ailleurs, mais [:0] est vide : « Tout afficher » annonçait « … 3 de plus » sans en montrer un seul. Le rapport n'était lu qu'une fois : après une purge il rejouait l'état d'avant, et l'on repurgeait des lignes déjà effacées. Une réparation dit maintenant si elle a fait quelque chose, et le rapport est relu alors. « DELETE 0 » passait pour un succès. Le compte vient de ce que PostgreSQL a annoncé, et ne rien dire n'est pas ne rien supprimer. Assisted-by: Claude Opus 5
1015 lines
38 KiB
Python
1015 lines
38 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)
|
||
|
||
"""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, pid="7"):
|
||
return {
|
||
"store_fname": store,
|
||
"model": model,
|
||
"id": pid,
|
||
"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_every_row_sharing_a_dead_file_is_listed_for_deletion(self):
|
||
# Vingt-deux lignes partageaient deux fichiers : la purge n'en
|
||
# offrait qu'une à la fois, et il fallait la relancer vingt-deux
|
||
# fois. Compter des FICHIERS et effacer des LIGNES ne sont pas la
|
||
# même opération.
|
||
fs.attachments = lambda base: [
|
||
piece("aa/bb", "res.country", "image", res_id="1", pid="10"),
|
||
piece("aa/bb", "res.country", "image", res_id="2", pid="11"),
|
||
piece("cc/dd", "res.country", "image", res_id="3", pid="12"),
|
||
]
|
||
rapport = fs.audit("db")
|
||
self.assertEqual(rapport["missing"], 2)
|
||
self.assertEqual(len(rapport["groups"]["dead_field"]), 2)
|
||
self.assertEqual(sorted(rapport["dead_ids"]), [10, 11, 12])
|
||
self.assertIn("(10, 11, 12)", fs.purge_dead_sql(rapport))
|
||
|
||
def test_a_live_row_sharing_a_file_is_never_swept_along(self):
|
||
# Deux lignes, un seul fichier, mais un seul champ mort : effacer
|
||
# les deux emporterait une pièce jointe bien vivante.
|
||
fs.live_fields = lambda base: {"project.task.attachment"}
|
||
fs.attachments = lambda base: [
|
||
piece("aa/bb", "res.country", "image", pid="10"),
|
||
piece("aa/bb", "project.task", "attachment", pid="11"),
|
||
]
|
||
rapport = fs.audit("db")
|
||
self.assertEqual(rapport["dead_ids"], [10])
|
||
|
||
def test_the_root_points_at_this_database_directory(self):
|
||
# Y mettre la racine de tous les filestores faisait chercher le
|
||
# dossier imbriqué à `<data_dir>/filestore/filestore` : l'outil
|
||
# répondait « rien à ranger » devant 1168 fichiers échoués.
|
||
fs.attachments = lambda base: []
|
||
self.assertEqual(fs.audit("ma_base")["root"], "/nulle/part/ma_base")
|
||
|
||
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_limit_zero_shows_EVERYTHING(self):
|
||
# « Tout afficher » passe limit=0. Une tranche [:0] est vide :
|
||
# l'écran annonçait « … 3 de plus » et ne montrait rien du tout.
|
||
groupes = {v: [] for v in fs.VERDICTS}
|
||
groupes["lost"] = [
|
||
piece(f"a/{i}", "project.task", name=f"perdu{i}.png")
|
||
for i in range(3)
|
||
]
|
||
texte = "\n".join(
|
||
fs.render(self.rapport(missing=3, groups=groupes), limit=0)
|
||
)
|
||
for i in range(3):
|
||
self.assertIn(f"perdu{i}.png", texte)
|
||
self.assertNotIn(todo_i18n.t("more"), texte)
|
||
|
||
def test_a_limit_still_caps_and_says_how_many_were_hidden(self):
|
||
groupes = {v: [] for v in fs.VERDICTS}
|
||
groupes["lost"] = [
|
||
piece(f"a/{i}", "project.task", name=f"perdu{i}.png")
|
||
for i in range(5)
|
||
]
|
||
texte = "\n".join(
|
||
fs.render(self.rapport(missing=5, groups=groupes), limit=2)
|
||
)
|
||
self.assertIn("perdu0.png", texte)
|
||
self.assertNotIn("perdu4.png", texte)
|
||
self.assertIn(f"3 {todo_i18n.t('more')}", 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 TestTheLivingRecord(unittest.TestCase):
|
||
"""Une image perdue sur un enregistrement supprimé n'est pas perdue."""
|
||
|
||
def setUp(self):
|
||
self.vrai = fs.run_psql
|
||
self.demandes = []
|
||
|
||
def tearDown(self):
|
||
fs.run_psql = self.vrai
|
||
|
||
def branche(self, tables, ids):
|
||
def faux(base, sql):
|
||
self.demandes.append(sql)
|
||
if "to_regclass" in sql:
|
||
return [["t" if any(x in sql for x in tables) else "f"]]
|
||
return [[str(i)] for i in ids]
|
||
|
||
fs.run_psql = faux
|
||
|
||
def test_a_living_record_is_reported_as_such(self):
|
||
self.branche(["project_task"], [15])
|
||
vivants = fs.resources_alive(
|
||
"db", [piece("a/1", "project.task", res_id="15")]
|
||
)
|
||
self.assertIs(vivants[("project.task", "15")], True)
|
||
|
||
def test_a_deleted_record_is_reported_as_gone(self):
|
||
self.branche(["project_task"], [])
|
||
vivants = fs.resources_alive(
|
||
"db", [piece("a/1", "project.task", res_id="15")]
|
||
)
|
||
self.assertIs(vivants[("project.task", "15")], False)
|
||
|
||
def test_a_model_without_a_table_is_left_UNKNOWN(self):
|
||
# « on n'a pas pu vérifier » ne doit pas se lire « il a disparu » :
|
||
# une vraie perte serait classée en fausse alerte.
|
||
self.branche([], [])
|
||
vivants = fs.resources_alive(
|
||
"db", [piece("a/1", "un.abstrait", res_id="1")]
|
||
)
|
||
self.assertEqual(vivants, {})
|
||
|
||
def test_an_attachment_without_a_record_is_not_queried(self):
|
||
self.branche([], [])
|
||
fs.resources_alive("db", [piece("a/1")])
|
||
self.assertEqual(self.demandes, [])
|
||
|
||
def test_the_mark_says_all_three_states(self):
|
||
self.assertIn(
|
||
todo_i18n.t("record still exists"), fs.alive_mark({"alive": True})
|
||
)
|
||
self.assertIn(
|
||
todo_i18n.t("record is gone — nothing will miss it"),
|
||
fs.alive_mark({"alive": False}),
|
||
)
|
||
self.assertEqual(fs.alive_mark({"alive": None}), "")
|
||
|
||
|
||
class TestTheRepairs(unittest.TestCase):
|
||
def rapport(self, ids=(), racine="/fs/db"):
|
||
base = {v: [] for v in fs.VERDICTS}
|
||
return {"root": racine, "groups": base, "dead_ids": list(ids)}
|
||
|
||
def test_the_purge_deletes_by_id_only(self):
|
||
# Par IDENTIFIANT, jamais par un domaine reconstruit : rejouer le
|
||
# raisonnement en SQL ouvrirait la porte à effacer autre chose
|
||
# que ce qui a été montré.
|
||
sql = fs.purge_dead_sql(self.rapport([9, 3]))
|
||
self.assertIn("WHERE id IN (3, 9)", sql)
|
||
self.assertNotIn("res_model", sql)
|
||
|
||
def test_nothing_dead_gives_no_sql(self):
|
||
self.assertEqual(fs.purge_dead_sql(self.rapport()), "")
|
||
|
||
def test_a_report_without_the_key_gives_no_sql(self):
|
||
self.assertEqual(fs.purge_dead_sql({"groups": {}}), "")
|
||
|
||
|
||
class TestReadingWhatPostgresSaid(unittest.TestCase):
|
||
def test_it_reads_the_count(self):
|
||
self.assertEqual(fs.rows_deleted(["DELETE 250"]), 250)
|
||
|
||
def test_zero_is_zero_not_success(self):
|
||
# « DELETE 0 » se félicitait d'avoir supprimé : rejouer une purge
|
||
# déjà faite annonçait un travail qui n'avait pas eu lieu.
|
||
self.assertEqual(fs.rows_deleted(["DELETE 0"]), 0)
|
||
|
||
def test_it_takes_the_LAST_word(self):
|
||
self.assertEqual(fs.rows_deleted(["DELETE 5", "bruit", "DELETE 7"]), 7)
|
||
|
||
def test_silence_is_not_zero(self):
|
||
# « rien annoncé » et « rien supprimé » appellent des mots
|
||
# différents : les confondre tait une panne.
|
||
self.assertIsNone(fs.rows_deleted(["ERROR: boom"]))
|
||
self.assertIsNone(fs.rows_deleted([]))
|
||
|
||
def test_a_plain_string_works_too(self):
|
||
self.assertEqual(fs.rows_deleted("DELETE 3\n"), 3)
|
||
|
||
|
||
class TestTidyingTheNested(unittest.TestCase):
|
||
def setUp(self):
|
||
self.racine = tempfile.mkdtemp()
|
||
self.base = os.path.join(self.racine, "ma_base")
|
||
for chemin in ("aa/deja", "filestore/aa/deja", "filestore/bb/absent"):
|
||
complet = os.path.join(self.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_it_separates_what_to_move_from_pure_duplicates(self):
|
||
# Écraser un fichier présent par une copie identique ne gagne
|
||
# rien et brouille la trace : les deux tas restent distincts.
|
||
remonter, doublons = fs.tidy_nested_plan({"root": self.base})
|
||
self.assertEqual(
|
||
[os.path.basename(a) for a, _b in remonter], ["absent"]
|
||
)
|
||
self.assertEqual([os.path.basename(a) for a, _b in doublons], ["deja"])
|
||
|
||
def test_the_destination_is_the_right_level(self):
|
||
remonter, _d = fs.tidy_nested_plan({"root": self.base})
|
||
self.assertEqual(
|
||
remonter[0][1], os.path.join(self.base, "bb", "absent")
|
||
)
|
||
|
||
def test_no_nested_directory_gives_an_empty_plan(self):
|
||
shutil.rmtree(os.path.join(self.base, "filestore"))
|
||
self.assertEqual(fs.tidy_nested_plan({"root": self.base}), ([], []))
|
||
self.assertEqual(fs.nested_dir({"root": self.base}), "")
|
||
|
||
|
||
class TestTheRepairMenu(unittest.TestCase):
|
||
"""Les réparations ÉCRIVENT. On éprouve ce qui part, pas les appels.
|
||
|
||
Un test qui constate qu'une fonction a été appelée n'aurait pas vu
|
||
que `exec_command` n'existe pas — seul `exec_command_live` est
|
||
offert. C'est arrivé ici : le code aurait planté au premier usage.
|
||
"""
|
||
|
||
def setUp(self):
|
||
from script.todo import auto_ask
|
||
from script.todo import todo as todo_module
|
||
|
||
self.auto_ask = auto_ask
|
||
self.vrai_ask = auto_ask.ask
|
||
self.lancees = []
|
||
self.obj = todo_module.TODO.__new__(todo_module.TODO)
|
||
|
||
def faux_exec(_self, cmd, **kw):
|
||
self.lancees.append(cmd)
|
||
if kw.get("return_status_and_output"):
|
||
return 0, ["DELETE 1"]
|
||
return 0
|
||
|
||
self.obj.execute = type("E", (), {"exec_command_live": faux_exec})()
|
||
|
||
def tearDown(self):
|
||
self.auto_ask.ask = self.vrai_ask
|
||
|
||
def repond(self, *reponses):
|
||
file = list(reponses)
|
||
|
||
def faux(prompt, default="", seconds=None):
|
||
reponse = file.pop(0) if file else ""
|
||
return reponse or default
|
||
|
||
self.auto_ask.ask = faux
|
||
|
||
def rapport(self, morts=(), ids=()):
|
||
groupes = {v: [] for v in fs.VERDICTS}
|
||
groupes["dead_field"] = list(morts)
|
||
return {"root": "/fs/db", "groups": groupes, "dead_ids": list(ids)}
|
||
|
||
def test_saying_no_deletes_nothing(self):
|
||
self.repond("n")
|
||
with redirect_stdout(io.StringIO()):
|
||
self.obj._filestore_purge_dead(
|
||
"db", self.rapport([piece("a/1")], [3])
|
||
)
|
||
self.assertEqual(self.lancees, [])
|
||
|
||
def test_pressing_enter_deletes_nothing(self):
|
||
# Le défaut d'une SUPPRESSION doit être de ne rien faire.
|
||
self.repond("")
|
||
with redirect_stdout(io.StringIO()):
|
||
self.obj._filestore_purge_dead(
|
||
"db", self.rapport([piece("a/1")], [3])
|
||
)
|
||
self.assertEqual(self.lancees, [])
|
||
|
||
def test_accepting_runs_a_command_that_actually_exists(self):
|
||
self.repond("y")
|
||
with redirect_stdout(io.StringIO()):
|
||
self.obj._filestore_purge_dead(
|
||
"db", self.rapport([piece("a/1")], [3])
|
||
)
|
||
self.assertEqual(len(self.lancees), 1)
|
||
self.assertIn("psql -d db", self.lancees[0])
|
||
self.assertIn("WHERE id IN (3)", self.lancees[0])
|
||
|
||
def test_nothing_dead_asks_nothing(self):
|
||
demandes = []
|
||
self.auto_ask.ask = lambda p, default="", seconds=None: (
|
||
demandes.append(p) or "y"
|
||
)
|
||
with redirect_stdout(io.StringIO()):
|
||
self.obj._filestore_purge_dead("db", self.rapport())
|
||
self.assertEqual(demandes, [])
|
||
self.assertEqual(self.lancees, [])
|
||
|
||
def test_the_menu_offers_both_repairs(self):
|
||
racine = os.path.normpath(
|
||
os.path.join(os.path.dirname(__file__), "..")
|
||
)
|
||
with io.open(
|
||
os.path.join(racine, "script", "todo", "todo.py"), encoding="utf-8"
|
||
) as handle:
|
||
src = handle.read()
|
||
self.assertIn("_filestore_purge_dead", src)
|
||
self.assertIn("_filestore_tidy_nested", src)
|
||
# Le garde du défaut doit rester collé à la question : c'est lui
|
||
# qui empêche Entrée de supprimer.
|
||
self.assertIn('auto_ask.ask(question, default="n")', src)
|
||
|
||
def test_a_delete_that_changed_nothing_is_NOT_a_success(self):
|
||
# « DELETE 0 » se félicitait d'avoir supprimé. Rejouer une purge
|
||
# déjà faite annonçait donc un travail qui n'avait pas eu lieu.
|
||
def rien(_self, cmd, **kw):
|
||
self.lancees.append(cmd)
|
||
return (
|
||
(0, ["DELETE 0"]) if kw.get("return_status_and_output") else 0
|
||
)
|
||
|
||
self.obj.execute = type("E", (), {"exec_command_live": rien})()
|
||
self.repond("y")
|
||
tampon = io.StringIO()
|
||
with redirect_stdout(tampon):
|
||
self.obj._filestore_purge_dead(
|
||
"db", self.rapport([piece("a/1")], [3])
|
||
)
|
||
self.assertIn("0 ", tampon.getvalue())
|
||
self.assertNotIn("✅ 1 ", tampon.getvalue())
|
||
|
||
def test_a_silent_command_is_flagged_not_counted(self):
|
||
def muet(_self, cmd, **kw):
|
||
self.lancees.append(cmd)
|
||
return (
|
||
(0, ["rien du tout"])
|
||
if kw.get("return_status_and_output")
|
||
else 0
|
||
)
|
||
|
||
self.obj.execute = type("E", (), {"exec_command_live": muet})()
|
||
self.repond("y")
|
||
tampon = io.StringIO()
|
||
with redirect_stdout(tampon):
|
||
self.obj._filestore_purge_dead(
|
||
"db", self.rapport([piece("a/1")], [3])
|
||
)
|
||
self.assertIn(
|
||
todo_i18n.t("The purge ran but said nothing."), tampon.getvalue()
|
||
)
|
||
|
||
def test_a_repair_reports_whether_it_did_something(self):
|
||
# C'est ce booléen qui déclenche la relecture du rapport : sans
|
||
# lui, « Tout afficher » rejouerait l'état d'avant la purge.
|
||
self.repond("n")
|
||
with redirect_stdout(io.StringIO()):
|
||
refus = self.obj._filestore_purge_dead(
|
||
"db", self.rapport([piece("a/1")], [3])
|
||
)
|
||
self.repond("y")
|
||
with redirect_stdout(io.StringIO()):
|
||
fait = self.obj._filestore_purge_dead(
|
||
"db", self.rapport([piece("a/1")], [3])
|
||
)
|
||
self.assertFalse(refus)
|
||
self.assertTrue(fait)
|
||
|
||
|
||
class TestTheReportIsRereadAfterARepair(unittest.TestCase):
|
||
"""« Tout afficher » doit montrer l'APRÈS, pas l'avant.
|
||
|
||
Sans relecture on purgeait, on relisait, et l'on voyait encore ce
|
||
qui venait de disparaître — puis on repurgeait des lignes déjà
|
||
effacées en croyant le travail inachevé. C'est arrivé en vrai.
|
||
"""
|
||
|
||
def setUp(self):
|
||
from script.todo import auto_ask
|
||
from script.todo import todo as todo_module
|
||
|
||
self.auto_ask = auto_ask
|
||
self.vrai_ask = auto_ask.ask
|
||
self.vrai_audit = fs.audit
|
||
self.appels = []
|
||
self.obj = todo_module.TODO.__new__(todo_module.TODO)
|
||
self.obj._analyse_select_database = lambda: "db"
|
||
self.obj._filestore_purge_dead = lambda base, rapport: True
|
||
self.obj._filestore_tidy_nested = lambda rapport: True
|
||
|
||
def audit(base, *a, **k):
|
||
self.appels.append(base)
|
||
return {
|
||
"database": base,
|
||
"attachments": 1,
|
||
"files_present": 1,
|
||
"missing": 0,
|
||
"nested_total": 0,
|
||
"root": "/fs/db",
|
||
"groups": {v: [] for v in fs.VERDICTS},
|
||
"dead_ids": [],
|
||
}
|
||
|
||
fs.audit = audit
|
||
self.auto_ask.ask = lambda p, default="", seconds=None: "n"
|
||
|
||
def tearDown(self):
|
||
self.auto_ask.ask = self.vrai_ask
|
||
fs.audit = self.vrai_audit
|
||
|
||
def joue(self, rang):
|
||
self.obj._analyse_follow_up = lambda choix, handler: handler(rang)
|
||
with redirect_stdout(io.StringIO()):
|
||
self.obj.execute_analyse_filestore()
|
||
|
||
def test_a_purge_triggers_a_fresh_read(self):
|
||
self.joue(2)
|
||
self.assertEqual(len(self.appels), 2)
|
||
|
||
def test_a_tidy_triggers_a_fresh_read(self):
|
||
self.joue(3)
|
||
self.assertEqual(len(self.appels), 2)
|
||
|
||
def test_merely_displaying_does_not(self):
|
||
# Relire pour afficher coûterait un balayage complet des
|
||
# filestores et des zips à chaque coup d'œil.
|
||
self.joue(1)
|
||
self.assertEqual(len(self.appels), 1)
|
||
|
||
def test_a_refused_repair_does_not_reread(self):
|
||
self.obj._filestore_purge_dead = lambda base, rapport: False
|
||
self.joue(2)
|
||
self.assertEqual(len(self.appels), 1)
|
||
|
||
|
||
class TestTidyingForReal(unittest.TestCase):
|
||
def setUp(self):
|
||
from script.todo import auto_ask
|
||
from script.todo import todo as todo_module
|
||
|
||
self.auto_ask = auto_ask
|
||
self.vrai_ask = auto_ask.ask
|
||
self.racine = tempfile.mkdtemp()
|
||
self.base = os.path.join(self.racine, "ma_base")
|
||
for chemin, contenu in (
|
||
("aa/deja", "bon"),
|
||
("filestore/aa/deja", "copie"),
|
||
("filestore/bb/absent", "utile"),
|
||
):
|
||
complet = os.path.join(self.base, chemin)
|
||
os.makedirs(os.path.dirname(complet), exist_ok=True)
|
||
with open(complet, "w", encoding="utf-8") as handle:
|
||
handle.write(contenu)
|
||
self.obj = todo_module.TODO.__new__(todo_module.TODO)
|
||
|
||
def tearDown(self):
|
||
self.auto_ask.ask = self.vrai_ask
|
||
shutil.rmtree(self.racine, ignore_errors=True)
|
||
|
||
def test_refusing_leaves_every_file_where_it_was(self):
|
||
self.auto_ask.ask = lambda p, default="", seconds=None: "n"
|
||
with redirect_stdout(io.StringIO()):
|
||
self.obj._filestore_tidy_nested({"root": self.base})
|
||
self.assertTrue(
|
||
os.path.isfile(
|
||
os.path.join(self.base, "filestore", "bb", "absent")
|
||
)
|
||
)
|
||
|
||
def test_accepting_moves_up_and_removes_the_nest(self):
|
||
self.auto_ask.ask = lambda p, default="", seconds=None: "y"
|
||
with redirect_stdout(io.StringIO()):
|
||
self.obj._filestore_tidy_nested({"root": self.base})
|
||
self.assertTrue(
|
||
os.path.isfile(os.path.join(self.base, "bb", "absent"))
|
||
)
|
||
self.assertFalse(os.path.isdir(os.path.join(self.base, "filestore")))
|
||
# Le fichier déjà présent n'a pas été ÉCRASÉ par sa copie :
|
||
# vérifier sa seule existence laisserait passer l'écrasement.
|
||
with io.open(
|
||
os.path.join(self.base, "aa", "deja"), encoding="utf-8"
|
||
) as handle:
|
||
self.assertEqual(handle.read(), "bon")
|
||
|
||
|
||
class TestVerifyingARestore(unittest.TestCase):
|
||
"""Le contrôle d'après-restauration, celui qui aurait vu le nichage."""
|
||
|
||
def setUp(self):
|
||
self.racine = tempfile.mkdtemp()
|
||
self.zip = os.path.join(self.racine, "sauv.zip")
|
||
with zipfile.ZipFile(self.zip, "w") as archive:
|
||
for nom in ("aa/un", "bb/deux", "cc/trois"):
|
||
archive.writestr(f"filestore/{nom}", "x")
|
||
archive.writestr("dump.sql", "x")
|
||
self.fs = os.path.join(self.racine, "fs")
|
||
self.vrai = fs.filestore_root
|
||
fs.filestore_root = lambda config=None: self.fs
|
||
|
||
def tearDown(self):
|
||
fs.filestore_root = self.vrai
|
||
shutil.rmtree(self.racine)
|
||
|
||
def pose(self, chemins):
|
||
for chemin in chemins:
|
||
complet = os.path.join(self.fs, "ma_base", chemin)
|
||
os.makedirs(os.path.dirname(complet), exist_ok=True)
|
||
with open(complet, "w", encoding="utf-8") as handle:
|
||
handle.write("x")
|
||
|
||
def test_a_clean_restore_reports_everything_in_place(self):
|
||
self.pose(["aa/un", "bb/deux", "cc/trois"])
|
||
r = fs.verify_restore("ma_base", self.zip)
|
||
self.assertEqual((r["expected"], r["placed"]), (3, 3))
|
||
self.assertEqual((r["nested"], r["missing"]), (0, 0))
|
||
|
||
def test_a_nested_restore_is_caught(self):
|
||
# Le défaut exact qui a coûté 133 Mo par base, sept fois.
|
||
self.pose(
|
||
["filestore/aa/un", "filestore/bb/deux", "filestore/cc/trois"]
|
||
)
|
||
r = fs.verify_restore("ma_base", self.zip)
|
||
self.assertEqual(r["nested"], 3)
|
||
self.assertEqual(r["placed"], 0)
|
||
|
||
def test_a_half_nested_restore_counts_both_sides(self):
|
||
self.pose(["aa/un", "filestore/bb/deux"])
|
||
r = fs.verify_restore("ma_base", self.zip)
|
||
self.assertEqual((r["placed"], r["nested"], r["missing"]), (1, 1, 1))
|
||
|
||
def test_files_that_never_landed_are_counted(self):
|
||
self.pose(["aa/un"])
|
||
self.assertEqual(fs.verify_restore("ma_base", self.zip)["missing"], 2)
|
||
|
||
def test_a_zip_without_a_filestore_says_nothing(self):
|
||
# Se taire quand il n'y a rien à contrôler : un contrôle bavard
|
||
# à chaque restauration finit par ne plus être lu.
|
||
vide = os.path.join(self.racine, "vide.zip")
|
||
with zipfile.ZipFile(vide, "w") as archive:
|
||
archive.writestr("dump.sql", "x")
|
||
r = fs.verify_restore("ma_base", vide)
|
||
self.assertEqual(r["expected"], 0)
|
||
self.assertEqual(fs.render_verify(r), [])
|
||
|
||
def test_the_fix_command_names_the_real_directory(self):
|
||
self.pose(["filestore/aa/un"])
|
||
texte = "\n".join(
|
||
fs.render_verify(fs.verify_restore("ma_base", self.zip))
|
||
)
|
||
self.assertIn(os.path.join(self.fs, "ma_base"), texte)
|
||
self.assertIn(todo_i18n.t("To fix:"), texte)
|
||
|
||
def test_a_clean_restore_still_says_so_briefly(self):
|
||
self.pose(["aa/un", "bb/deux", "cc/trois"])
|
||
texte = "\n".join(
|
||
fs.render_verify(fs.verify_restore("ma_base", self.zip))
|
||
)
|
||
self.assertIn(todo_i18n.t("Filestore restored:"), texte)
|
||
self.assertNotIn(todo_i18n.t("To fix:"), texte)
|
||
|
||
|
||
class TestTheRestoreWiring(unittest.TestCase):
|
||
RACINE = os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
|
||
|
||
def source(self, chemin):
|
||
with io.open(
|
||
os.path.join(self.RACINE, chemin), encoding="utf-8"
|
||
) as handle:
|
||
return handle.read()
|
||
|
||
def test_db_restore_checks_after_a_real_restore(self):
|
||
src = self.source("script/database/db_restore.py")
|
||
self.assertEqual(src.count("verify_filestore("), 3)
|
||
self.assertIn(
|
||
"if config.ignore_cache:\n verify_filestore("
|
||
"config.database",
|
||
src,
|
||
"le contrôle d'après-restauration directe n'est plus gardé"
|
||
" par ignore_cache",
|
||
)
|
||
self.assertIn(
|
||
"not config.ignore_cache:",
|
||
src,
|
||
"la création du cache n'est plus conditionnée",
|
||
)
|
||
|
||
def test_the_clone_path_is_NOT_checked(self):
|
||
# Le miroir recopie sa source, défauts compris : contrôler là
|
||
# dirait deux fois la même chose, et au mauvais endroit.
|
||
src = self.source("script/database/db_restore.py")
|
||
self.assertIn("--clone --from_database", src)
|
||
debut = src.index("--clone --from_database")
|
||
marque = "verify_filestore(config.database"
|
||
self.assertIn(
|
||
marque, src, "le contrôle d'après-restauration a disparu"
|
||
)
|
||
fin = src.index(marque, debut)
|
||
self.assertNotIn("verify_filestore", src[debut:fin])
|
||
|
||
def test_the_analyse_menu_offers_the_tool(self):
|
||
src = self.source("script/todo/todo.py")
|
||
self.assertIn("Attachment files missing from the filestore", src)
|
||
self.assertIn("self.execute_analyse_filestore()", src)
|
||
self.assertIn("def execute_analyse_filestore", src)
|
||
|
||
def test_the_menu_has_as_many_entries_as_branches(self):
|
||
src = self.source("script/todo/todo.py")
|
||
debut = src.index("def prompt_execute_analyse")
|
||
fin = src.index('print(t("Command not found !"))', debut)
|
||
bloc = src[debut:fin]
|
||
entrees = bloc.count('"prompt_description"')
|
||
branches = sum(
|
||
f'status == "{n}"' in bloc for n in range(1, entrees + 2)
|
||
)
|
||
self.assertEqual(entrees, branches)
|
||
|
||
|
||
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 TestTheFollowUpIcons(unittest.TestCase):
|
||
"""Chaque entrée « Aller plus loin » porte une icône, ou aucune.
|
||
|
||
Une seule entrée nue au milieu d'entrées ornées se lit comme un
|
||
oubli — et c'en est un. Le garde vaut mieux qu'une relecture : la
|
||
prochaine entrée ajoutée sans icône tombera ici, pas dans l'œil de
|
||
quelqu'un six mois plus tard.
|
||
"""
|
||
|
||
RACINE = os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
|
||
|
||
def libelles(self):
|
||
"""Les clés passées en `prompt_description` à `_analyse_follow_up`."""
|
||
import ast
|
||
|
||
with io.open(
|
||
os.path.join(self.RACINE, "script", "todo", "todo.py"),
|
||
encoding="utf-8",
|
||
) as handle:
|
||
arbre = ast.parse(handle.read())
|
||
trouves = []
|
||
for node in ast.walk(arbre):
|
||
if not (
|
||
isinstance(node, ast.Call)
|
||
and isinstance(node.func, ast.Attribute)
|
||
and node.func.attr == "_analyse_follow_up"
|
||
and node.args
|
||
):
|
||
continue
|
||
for element in ast.walk(node.args[0]):
|
||
if (
|
||
isinstance(element, ast.Call)
|
||
and isinstance(element.func, ast.Name)
|
||
and element.func.id == "t"
|
||
and element.args
|
||
and isinstance(element.args[0], ast.Constant)
|
||
):
|
||
trouves.append(element.args[0].value)
|
||
return trouves
|
||
|
||
def test_there_are_follow_up_entries_to_check(self):
|
||
# Sans cette borne, un jour où l'extraction ne trouve plus rien,
|
||
# le test suivant passerait en ne vérifiant rien du tout.
|
||
self.assertGreater(len(self.libelles()), 5)
|
||
|
||
def test_every_follow_up_entry_carries_an_icon(self):
|
||
for cle in self.libelles():
|
||
for langue in ("fr", "en"):
|
||
texte = todo_i18n.TRANSLATIONS.get(cle, {}).get(langue, cle)
|
||
self.assertGreaterEqual(
|
||
ord(texte[0]),
|
||
0x1F300,
|
||
f"entrée sans icône [{langue}] : {texte!r}",
|
||
)
|
||
|
||
|
||
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()
|