Mathieu Benoit
1bc2b50ca8
A lost image on a deleted task is not a loss -- nobody will ever look for it. On a LIVING task it is one, and it is the only one worth regretting. The report now says which: project.task #15 and calendar.event #1 both still exist, so those two really are gone. Three states, not two: "could not check" must not read as "it is gone", or a real loss gets filed as a false alarm. Two repairs sit in the follow-up menu. Purging rows whose field no longer exists deletes by ID, never by a rebuilt domain -- replaying the reasoning in SQL would open the door to deleting more than was shown. Tidying the nested filestore moves up what is missing and deletes pure duplicates, never overwriting a file already in place. --- FR --- Une image perdue sur une tâche supprimée n'est pas une perte : personne ne la cherchera. Sur une tâche VIVANTE, c'en est une, et la seule à regretter. Le rapport le dit : project.task #15 et calendar.event #1 existent encore, ces deux-là sont bien perdues. Trois états, pas deux : « pas pu vérifier » ne doit pas se lire « a disparu », sans quoi une vraie perte passe pour une fausse alerte. Deux réparations dans le menu de suite. La purge efface par IDENTIFIANT, jamais par un domaine reconstruit -- rejouer le raisonnement en SQL ouvrirait la porte à effacer plus que ce qui a été montré. Le rangement remonte ce qui manque et supprime les doublons purs, sans jamais écraser un fichier déjà en place. Assisted-by: Claude Opus 5
756 lines
28 KiB
Python
756 lines
28 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_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 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, morts=(), racine="/fs/db"):
|
||
base = {v: [] for v in fs.VERDICTS}
|
||
base["dead_field"] = list(morts)
|
||
return {"root": racine, "groups": base}
|
||
|
||
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([piece("a/1", pid="3"), piece("b/2", pid="9")])
|
||
)
|
||
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_row_without_an_id_is_never_deleted(self):
|
||
sql = fs.purge_dead_sql(self.rapport([piece("a/1", pid="")]))
|
||
self.assertEqual(sql, "")
|
||
|
||
|
||
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)
|
||
self.obj.execute = type(
|
||
"E",
|
||
(),
|
||
{
|
||
"exec_command_live": lambda _s, cmd, **k: self.lancees.append(
|
||
cmd
|
||
)
|
||
or 0
|
||
},
|
||
)()
|
||
|
||
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=()):
|
||
groupes = {v: [] for v in fs.VERDICTS}
|
||
groupes["dead_field"] = list(morts)
|
||
return {"root": "/fs/db", "groups": groupes}
|
||
|
||
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")]))
|
||
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")]))
|
||
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", pid="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)
|
||
|
||
|
||
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 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()
|