Merge commit 'dev_analyse_database'
[ADD] release: look at a database before trusting it Three questions a restored or migrated database cannot answer on its own, and now can. Which website copies shadow a view, and how far they have drifted from it -- the diff is shown per copy rather than left to be guessed. Which modules the database lacks against the default package, with an offer to install those that are ready. The suggestion is not blind: a module that would not install is not proposed. Whether the filestore actually landed. A restore that dropped its attachments looked identical to one that did not. The check says whether the file is missing, whether the record behind it still exists, and offers the cleanups -- purging once at the end rather than record by record, and reporting the 30 MB it recovered. Along the way, db_restore asks the master password again instead of dying on a typo: an hour of bumps was a steep price for one letter. --- FR --- Trois questions auxquelles une base restaurée ou migrée ne savait pas répondre seule, et qu'elle sait maintenant. Quelles copies de site masquent une vue, et de combien elles s'en sont écartées — l'écart est montré copie par copie plutôt que laissé à deviner. Quels modules manquent à la base par rapport au paquet par défaut, avec une offre d'installer ceux qui sont prêts. La suggestion n'est pas aveugle : un module qui ne s'installerait pas n'est pas proposé. Si le filestore est réellement arrivé. Une restauration qui avait perdu ses pièces jointes ressemblait trait pour trait à une réussie. La vérification dit si le fichier manque, si l'enregistrement derrière existe encore, et propose les nettoyages — en purgeant une fois à la fin plutôt qu'enregistrement par enregistrement, et en annonçant les 30 Mo récupérés. Au passage, db_restore redemande le mot de passe maître au lieu de mourir sur une faute de frappe : une heure de paliers était cher payé pour une lettre. Assisted-by: Claude Opus 5
This commit is contained in:
commit
61341e803b
11 changed files with 4323 additions and 51 deletions
|
|
@ -334,9 +334,114 @@ def add_reference_arch(database, lst_finding, config_path=None, timeout=600):
|
|||
return "orm", None
|
||||
|
||||
|
||||
def attach_cow_twin_diff(lst_finding, dct_twin, dct_copy_arch):
|
||||
"""Comparer chaque copie COW à la vue de module qu'elle masque.
|
||||
|
||||
C'est LA comparaison qui compte pour une copie, et elle n'a besoin d'aucun
|
||||
registre : les deux côtés sont dans la base, appariés par leur clé —
|
||||
exactement l'appariement que fait Odoo, et la raison pour laquelle
|
||||
renommer une clé suffit à désapparier une copie.
|
||||
|
||||
Elle marche donc là où la comparaison avec la source du module est
|
||||
impossible : une base dont la version diffère du checkout, et une
|
||||
sauvegarde .zip, qui portent l'une comme l'autre les deux arch.
|
||||
|
||||
Une copie sans jumelle est laissée telle quelle : c'est une page faite
|
||||
dans l'éditeur web, il n'y a rien à quoi la comparer.
|
||||
|
||||
Fonction pure — les deux provenances lui passent leurs dictionnaires et
|
||||
concluent donc la même chose des mêmes faits. Renvoie le nombre de copies
|
||||
comparées.
|
||||
"""
|
||||
n_compared = 0
|
||||
for row in lst_finding:
|
||||
if row["category"] != "website_cow_copy":
|
||||
continue
|
||||
twin = dct_twin.get(row.get("key"))
|
||||
if not twin:
|
||||
continue
|
||||
twin_id, twin_arch = twin
|
||||
copy_arch = dct_copy_arch.get(row["id"])
|
||||
if copy_arch is None:
|
||||
continue
|
||||
differs, comparable = arch_differs(twin_arch, copy_arch)
|
||||
# Mêmes noms de champs que la comparaison avec la source du module :
|
||||
# l'écran de navigation et le rendu texte marchent alors sans savoir
|
||||
# laquelle des deux a produit la donnée.
|
||||
row["arch_ref"] = twin_arch
|
||||
row["arch_db_text"] = copy_arch
|
||||
row["twin_id"] = twin_id
|
||||
row["comparable"] = comparable
|
||||
row["differs"] = differs
|
||||
if comparable:
|
||||
row["diff_stats"] = diff_stats(side_by_side(twin_arch, copy_arch))
|
||||
n_compared += 1
|
||||
return n_compared
|
||||
|
||||
|
||||
def _cow_twin_arch(database, lst_key, **kwargs):
|
||||
"""{clé: (id, arch)} des vues de module masquées par ces copies."""
|
||||
if not lst_key:
|
||||
return {}
|
||||
values = ", ".join(
|
||||
"'" + k.replace("'", "''") + "'" for k in sorted(set(lst_key))
|
||||
)
|
||||
rows = json_query(
|
||||
database,
|
||||
f"""
|
||||
SELECT DISTINCT ON (v.key)
|
||||
v.key AS key,
|
||||
v.id AS id,
|
||||
v.arch_db::text AS arch
|
||||
FROM ir_ui_view v
|
||||
WHERE v.key IN ({values}) AND v.website_id IS NULL
|
||||
ORDER BY v.key, v.id
|
||||
""",
|
||||
**kwargs,
|
||||
)
|
||||
return {r["key"]: (r["id"], normalise_arch(r["arch"])) for r in rows}
|
||||
|
||||
|
||||
def _cow_copy_arch(database, lst_id, **kwargs):
|
||||
"""{id: arch} des seules copies retenues.
|
||||
|
||||
Rapatrier l'arch de toutes les vues ferait une ligne de sortie de plusieurs
|
||||
centaines de mégaoctets ; celle des seules copies COW en fait quelques-uns.
|
||||
"""
|
||||
if not lst_id:
|
||||
return {}
|
||||
ids = ", ".join(str(int(i)) for i in lst_id)
|
||||
rows = json_query(
|
||||
database,
|
||||
"SELECT id AS id, arch_db::text AS arch"
|
||||
f" FROM ir_ui_view WHERE id IN ({ids})",
|
||||
**kwargs,
|
||||
)
|
||||
return {r["id"]: normalise_arch(r["arch"]) for r in rows}
|
||||
|
||||
|
||||
def add_cow_twin_diff(database, lst_finding, **kwargs):
|
||||
"""Comparer les copies COW d'une BASE à leur jumelle. Deux requêtes."""
|
||||
lst_copy = [
|
||||
row
|
||||
for row in lst_finding
|
||||
if row["category"] == "website_cow_copy" and row.get("has_module_twin")
|
||||
]
|
||||
if not lst_copy:
|
||||
return 0
|
||||
dct_twin = _cow_twin_arch(
|
||||
database, [row["key"] for row in lst_copy if row.get("key")], **kwargs
|
||||
)
|
||||
dct_copy = _cow_copy_arch(
|
||||
database, [row["id"] for row in lst_copy], **kwargs
|
||||
)
|
||||
return attach_cow_twin_diff(lst_finding, dct_twin, dct_copy)
|
||||
|
||||
|
||||
def collect(
|
||||
database,
|
||||
with_diff=False,
|
||||
with_cow_diff=False,
|
||||
scope="flagged",
|
||||
config_path=None,
|
||||
timeout=120,
|
||||
|
|
@ -370,6 +475,12 @@ def collect(
|
|||
if category in ACTIONABLE:
|
||||
lst_finding.append(row)
|
||||
|
||||
n_cow_compared = 0
|
||||
if with_cow_diff:
|
||||
# Indépendant de la voie ORM : les deux arch sont en base, donc ceci
|
||||
# marche même quand la version du checkout interdit l'autre.
|
||||
n_cow_compared = add_cow_twin_diff(database, lst_finding, **kwargs)
|
||||
|
||||
arch_ref_source, arch_ref_error = "none", None
|
||||
checkout = checkout_odoo_version()
|
||||
if with_diff:
|
||||
|
|
@ -455,6 +566,7 @@ def collect(
|
|||
"scope": scope,
|
||||
"arch_ref_error": arch_ref_error,
|
||||
"n_identical_after_canonical": n_identical,
|
||||
"n_cow_compared": n_cow_compared,
|
||||
"n_views": len(lst_view),
|
||||
"counts": dct_count,
|
||||
"findings": lst_finding,
|
||||
|
|
@ -528,11 +640,30 @@ def collect_from_backup(zip_path):
|
|||
if category in ACTIONABLE:
|
||||
lst_finding.append(row)
|
||||
|
||||
# Le dump porte les arch des deux côtés : la comparaison COW est donc
|
||||
# possible depuis un zip, là où celle avec la source du module ne l'est
|
||||
# pas faute de registre.
|
||||
dct_twin = {}
|
||||
for row in dct_rows["ir_ui_view"]:
|
||||
if row.get("key") and row.get("website_id") in (None, ""):
|
||||
dct_twin.setdefault(
|
||||
row["key"],
|
||||
(row.get("id"), normalise_arch(row.get("arch_db"))),
|
||||
)
|
||||
dct_copy = {
|
||||
(int(r.get("id")) if (r.get("id") or "").isdigit() else r.get("id")): (
|
||||
normalise_arch(r.get("arch_db"))
|
||||
)
|
||||
for r in dct_rows["ir_ui_view"]
|
||||
}
|
||||
n_cow_compared = attach_cow_twin_diff(lst_finding, dct_twin, dct_copy)
|
||||
|
||||
return {
|
||||
"tool": "analyse_view_custom",
|
||||
"version": 1,
|
||||
"database": os.path.basename(zip_path),
|
||||
"source": "backup",
|
||||
"n_cow_compared": n_cow_compared,
|
||||
"backup_path": zip_path,
|
||||
"odoo_version": backup_version(dct_rows, manifest),
|
||||
"checkout_version": checkout_odoo_version(),
|
||||
|
|
@ -792,6 +923,12 @@ def main(argv=None):
|
|||
default="flagged",
|
||||
help=t("which views to compare (default: flagged)"),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--cow-diff",
|
||||
dest="cow_diff",
|
||||
action="store_true",
|
||||
help=t("compare each website copy with the module view it shadows"),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--strict",
|
||||
action="store_true",
|
||||
|
|
@ -814,7 +951,8 @@ def main(argv=None):
|
|||
else:
|
||||
data = collect(
|
||||
config.database,
|
||||
with_diff=config.diff or config.tui,
|
||||
with_diff=config.diff,
|
||||
with_cow_diff=config.cow_diff or config.tui,
|
||||
scope=config.scope,
|
||||
config_path=config.config,
|
||||
)
|
||||
|
|
|
|||
693
script/analyse/check_filestore.py
Executable file
693
script/analyse/check_filestore.py
Executable file
|
|
@ -0,0 +1,693 @@
|
|||
#!/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/<base>/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, ''),"
|
||||
" a.id::text,"
|
||||
" 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],
|
||||
"id": ligne[2],
|
||||
"field": ligne[3],
|
||||
"res_id": ligne[4],
|
||||
"name": ligne[5],
|
||||
"size": int(ligne[6] or 0),
|
||||
"mimetype": ligne[7],
|
||||
"created": ligne[8][:10],
|
||||
}
|
||||
for ligne in lignes
|
||||
if len(ligne) >= 9
|
||||
]
|
||||
|
||||
|
||||
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, par_base = {}, {}, {}
|
||||
if not os.path.isdir(racine):
|
||||
return ailleurs, niches, par_base
|
||||
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":
|
||||
# Un même fichier peut être niché dans PLUSIEURS bases —
|
||||
# le clone les recopie toutes. L'index n'en retient qu'une
|
||||
# (le premier `setdefault` gagne) : il sert à retrouver un
|
||||
# fichier, pas à compter. D'où le décompte par base, sans
|
||||
# lequel une base nichée s'entendait dire que le problème
|
||||
# était chez les voisines.
|
||||
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):
|
||||
niches.setdefault(f"{deux}/{nom}", base)
|
||||
par_base[base] = par_base.get(base, 0) + 1
|
||||
continue
|
||||
if base == sauf:
|
||||
continue
|
||||
for nom in os.listdir(sous):
|
||||
ailleurs.setdefault(f"{prefixe}/{nom}", base)
|
||||
return ailleurs, niches, par_base
|
||||
|
||||
|
||||
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 resources_alive(database, pieces):
|
||||
"""{(modele, id): True/False} — l'enregistrement visé existe-t-il ?
|
||||
|
||||
Une image perdue dont la tâche a été supprimée n'est pas une perte :
|
||||
personne ne la cherchera jamais. Une image perdue sur une tâche
|
||||
VIVANTE en est une, et c'est la seule qu'il faille regretter.
|
||||
Confondre les deux, c'est pleurer au hasard.
|
||||
|
||||
Une requête par modèle, sur les seules pièces jointes déjà classées
|
||||
perdues — jamais sur les milliers d'autres.
|
||||
"""
|
||||
par_modele = {}
|
||||
for piece in pieces:
|
||||
if not piece.get("model") or not (piece.get("res_id") or "").isdigit():
|
||||
continue
|
||||
par_modele.setdefault(piece["model"], set()).add(int(piece["res_id"]))
|
||||
vivants = {}
|
||||
for modele, ids in par_modele.items():
|
||||
table = modele.replace(".", "_").replace("'", "")
|
||||
# Un modèle abstrait ou transitoire n'a pas de table : interroger
|
||||
# une table absente rendrait None, qu'on lirait « n'existe pas ».
|
||||
# Ce serait un mensonge, et le pire sens du mensonge ici.
|
||||
existe = run_psql(
|
||||
database, f"SELECT to_regclass('{table}') IS NOT NULL"
|
||||
)
|
||||
if not existe or existe[0][0] != "t":
|
||||
continue
|
||||
liste = ", ".join(str(i) for i in sorted(ids))
|
||||
lignes = run_psql(
|
||||
database, f"SELECT id FROM {table} WHERE id IN ({liste})"
|
||||
)
|
||||
if lignes is None:
|
||||
continue
|
||||
trouves = {int(ligne[0]) for ligne in lignes if ligne and ligne[0]}
|
||||
for identifiant in ids:
|
||||
vivants[(modele, str(identifiant))] = identifiant in trouves
|
||||
return vivants
|
||||
|
||||
|
||||
def files_on_disk(racine, database):
|
||||
"""(au bon niveau, nichés) pour une base. Deux ensembles de noms."""
|
||||
base = os.path.join(racine, database)
|
||||
bons, niches = set(), set()
|
||||
if not os.path.isdir(base):
|
||||
return bons, niches
|
||||
for prefixe in os.listdir(base):
|
||||
sous = os.path.join(base, prefixe)
|
||||
if not os.path.isdir(sous):
|
||||
continue
|
||||
if prefixe == "filestore":
|
||||
for deux in os.listdir(sous):
|
||||
profond = os.path.join(sous, deux)
|
||||
if not os.path.isdir(profond):
|
||||
continue
|
||||
for nom in os.listdir(profond):
|
||||
if os.path.isfile(os.path.join(profond, nom)):
|
||||
niches.add(f"{deux}/{nom}")
|
||||
continue
|
||||
for nom in os.listdir(sous):
|
||||
if os.path.isfile(os.path.join(sous, nom)):
|
||||
bons.add(f"{prefixe}/{nom}")
|
||||
return bons, niches
|
||||
|
||||
|
||||
def verify_restore(database, zip_path, config_path=None):
|
||||
"""La sauvegarde a-t-elle bien atterri ? À vérifier UNE fois.
|
||||
|
||||
Après un clone, il n'y a rien à contrôler : `copytree` recopie la
|
||||
source telle quelle, défauts compris — le contrôle appartient à la
|
||||
restauration d'origine, pas au miroir.
|
||||
|
||||
Ce qu'on cherche est précis. `shutil.move` d'Odoo renomme quand la
|
||||
destination n'existe pas et IMBRIQUE quand elle existe. Un dossier
|
||||
`filestore/<base>/` laissé par une restauration précédente suffit
|
||||
donc à envoyer toute la sauvegarde dans
|
||||
`filestore/<base>/filestore/`, où Odoo ne regardera jamais. Mesuré :
|
||||
1168 fichiers, 133 Mo, recopiés ensuite dans les six bases de la
|
||||
chaîne par le clone, sans que rien ne le signale.
|
||||
"""
|
||||
attendus = set(scan_zip(zip_path))
|
||||
racine = filestore_root(config_path)
|
||||
bons, niches = files_on_disk(racine, database)
|
||||
return {
|
||||
"database": database,
|
||||
"zip": os.path.basename(zip_path),
|
||||
"expected": len(attendus),
|
||||
"placed": len(attendus & bons),
|
||||
"nested": len(attendus & niches),
|
||||
"missing": len(attendus - bons - niches),
|
||||
"root": os.path.join(racine, database),
|
||||
}
|
||||
|
||||
|
||||
def scan_zip(chemin):
|
||||
"""Les noms de fichiers du `filestore/` d'une sauvegarde."""
|
||||
try:
|
||||
with zipfile.ZipFile(chemin) as archive:
|
||||
return [
|
||||
membre[len("filestore/") :]
|
||||
for membre in archive.namelist()
|
||||
if membre.startswith("filestore/") and not membre.endswith("/")
|
||||
]
|
||||
except (OSError, zipfile.BadZipFile):
|
||||
return []
|
||||
|
||||
|
||||
def render_verify(rapport):
|
||||
"""Se taire quand tout va bien : un contrôle bavard finit ignoré."""
|
||||
if not rapport["expected"]:
|
||||
return []
|
||||
if not rapport["nested"] and not rapport["missing"]:
|
||||
return [
|
||||
f"✅ {t('Filestore restored:')} {rapport['placed']}"
|
||||
f"/{rapport['expected']} {t('file(s) in place')}"
|
||||
]
|
||||
lignes = [
|
||||
f"⚠ {t('Filestore restore looks wrong for')} {rapport['database']}"
|
||||
f" ({rapport['zip']}) :",
|
||||
f" {rapport['placed']}/{rapport['expected']}"
|
||||
f" {t('file(s) in place')}",
|
||||
]
|
||||
if rapport["nested"]:
|
||||
lignes.append(
|
||||
f" {rapport['nested']}"
|
||||
f" {t('landed in a nested filestore Odoo never reads')}"
|
||||
)
|
||||
lignes.append(
|
||||
f" {t('To fix:')} rsync -a --remove-source-files"
|
||||
f" {rapport['root']}/filestore/ {rapport['root']}/"
|
||||
)
|
||||
if rapport["missing"]:
|
||||
lignes.append(f" {rapport['missing']} {t('never landed at all')}")
|
||||
return lignes
|
||||
|
||||
|
||||
def is_dead_field(piece, champs_vivants):
|
||||
"""La pièce jointe porte-t-elle un champ qui n'existe plus ?
|
||||
|
||||
Sans `res_field` la question ne se pose pas : c'est un document
|
||||
téléversé, pas la valeur d'un champ. Le juger sur un champ absent le
|
||||
ferait disparaître du rapport.
|
||||
"""
|
||||
if not piece.get("field"):
|
||||
return False
|
||||
return f"{piece['model']}.{piece['field']}" not in champs_vivants
|
||||
|
||||
|
||||
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 is_dead_field(piece, champs_vivants):
|
||||
return ("dead_field", f"{piece['model']}.{piece['field']}")
|
||||
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)
|
||||
# `root` désigne le dossier de CETTE base. Y mettre la racine de tous
|
||||
# les filestores faisait chercher le dossier imbriqué à
|
||||
# `<data_dir>/filestore/filestore` — inexistant — et l'outil
|
||||
# répondait « rien à ranger » devant 1168 fichiers échoués.
|
||||
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, par_base = 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()
|
||||
morts = []
|
||||
gardees = []
|
||||
for piece in pieces:
|
||||
verdict = classify(
|
||||
piece, present, ailleurs, niches, sauvegardes, champs_vivants
|
||||
)
|
||||
if not verdict:
|
||||
# Fichier présent — mais son champ vit-il encore ? Un outil
|
||||
# nommé « fichiers absents » ne regardait pas là, et laissait
|
||||
# dormir 1860 lignes et 30 Mo que plus rien ne lit.
|
||||
if is_dead_field(piece, champs_vivants):
|
||||
gardees.append(piece)
|
||||
if str(piece.get("id", "")).isdigit():
|
||||
morts.append(int(piece["id"]))
|
||||
continue
|
||||
# La déduplication qui suit sert à compter des FICHIERS. Pour
|
||||
# effacer des LIGNES il les faut toutes : vingt-deux lignes
|
||||
# partageaient deux fichiers, et la purge n'en offrait qu'une à
|
||||
# la fois — il fallait la relancer vingt-deux fois.
|
||||
if verdict[0] == "dead_field" and str(piece.get("id", "")).isdigit():
|
||||
morts.append(int(piece["id"]))
|
||||
# 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]))
|
||||
vivants = resources_alive(database, groupes["lost"])
|
||||
for piece in groupes["lost"]:
|
||||
piece["alive"] = vivants.get((piece["model"], piece["res_id"]))
|
||||
return {
|
||||
"database": database,
|
||||
"attachments": len(pieces),
|
||||
"files_present": len(present),
|
||||
"missing": len(vus),
|
||||
"groups": groupes,
|
||||
"nested_total": par_base.get(database, 0),
|
||||
"nested_elsewhere": sum(
|
||||
combien for base, combien in par_base.items() if base != database
|
||||
),
|
||||
"root": mien,
|
||||
"dead_ids": morts,
|
||||
"dead_kept": gardees,
|
||||
"dead_kept_size": sum(piece["size"] for piece in gardees),
|
||||
}
|
||||
|
||||
|
||||
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_dead_kept(rapport) + 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 or None]:
|
||||
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']} {alive_mark(piece)}"
|
||||
)
|
||||
if limit and len(groupe) > limit:
|
||||
lignes.append(f" … {len(groupe) - limit} {t('more')}")
|
||||
return lignes + render_dead_kept(rapport) + render_nested(rapport)
|
||||
|
||||
|
||||
def render_dead_kept(rapport, limit=4):
|
||||
"""Les lignes mortes dont le FICHIER est toujours là.
|
||||
|
||||
Elles ne manquent à personne — c'est justement le problème : rien ne
|
||||
les lit, et leur fichier occupe le disque tant que la ligne existe,
|
||||
puisque le ramasse-miettes d'Odoo ne retire que ce qui n'est plus
|
||||
référencé. Un outil nommé « fichiers absents » ne regardait pas là,
|
||||
et laissait dormir 1860 lignes et 30 Mo.
|
||||
"""
|
||||
gardees = rapport.get("dead_kept") or []
|
||||
if not gardees:
|
||||
return []
|
||||
lignes = [
|
||||
"",
|
||||
f" 🕳 {len(gardees)}"
|
||||
f" {t('row(s) whose field is gone still hold their file')}"
|
||||
f" ({rapport.get('dead_kept_size', 0) // 1024} ko)",
|
||||
]
|
||||
apercu = summarise(gardees)
|
||||
for texte in apercu[: limit or None]:
|
||||
lignes.append(f" {texte}")
|
||||
if limit and len(apercu) > limit:
|
||||
lignes.append(f" … {len(apercu) - limit} {t('more')}")
|
||||
return lignes
|
||||
|
||||
|
||||
def render_nested(rapport):
|
||||
ailleurs = rapport.get("nested_elsewhere") or 0
|
||||
if not rapport.get("nested_total"):
|
||||
# Rien ici, mais peut-être chez les voisines : le dire sans
|
||||
# laisser croire que CETTE base est concernée.
|
||||
if ailleurs:
|
||||
return [
|
||||
"",
|
||||
f" ↳ {ailleurs}"
|
||||
f" {t('such file(s) sit in OTHER databases filestores.')}",
|
||||
]
|
||||
return []
|
||||
lignes = [
|
||||
"",
|
||||
f" ↳ {rapport['nested_total']}"
|
||||
f" {t('file(s) sit in a nested filestore Odoo never reads.')}",
|
||||
]
|
||||
return lignes
|
||||
|
||||
|
||||
def alive_mark(piece):
|
||||
"""Dire si l'enregistrement visé vit encore, ou qu'on ne sait pas.
|
||||
|
||||
Trois états, pas deux : « on n'a pas pu vérifier » ne doit pas se
|
||||
lire comme « il a disparu », sans quoi une vraie perte serait classée
|
||||
en fausse alerte.
|
||||
"""
|
||||
etat = piece.get("alive")
|
||||
if etat is True:
|
||||
return f"✓ {t('record still exists')}"
|
||||
if etat is False:
|
||||
return f"✗ {t('record is gone — nothing will miss it')}"
|
||||
return ""
|
||||
|
||||
|
||||
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 purge_dead_sql(rapport):
|
||||
"""Le SQL qui efface les lignes dont le champ n'existe plus, ou "".
|
||||
|
||||
On efface par IDENTIFIANT, pas par un domaine reconstruit : la liste
|
||||
a été établie en confrontant `ir_model_fields` à ce que la base
|
||||
porte, et rejouer ce raisonnement en SQL laisserait la porte ouverte
|
||||
à effacer autre chose que ce qui a été montré.
|
||||
"""
|
||||
ids = sorted(set(rapport.get("dead_ids") or []))
|
||||
if not ids:
|
||||
return ""
|
||||
liste = ", ".join(str(i) for i in ids)
|
||||
return f"DELETE FROM ir_attachment WHERE id IN ({liste})"
|
||||
|
||||
|
||||
def rows_deleted(sortie):
|
||||
"""Le nombre annoncé par « DELETE n », ou None si rien ne le dit.
|
||||
|
||||
None n'est pas zéro : « la commande n'a rien annoncé » et « elle n'a
|
||||
rien supprimé » appellent des mots différents. Les confondre ferait
|
||||
taire une panne ou inventer un succès.
|
||||
"""
|
||||
lignes = sortie if isinstance(sortie, list) else str(sortie).splitlines()
|
||||
for ligne in reversed([str(x).strip() for x in lignes]):
|
||||
if ligne.startswith("DELETE "):
|
||||
reste = ligne[len("DELETE ") :].strip()
|
||||
if reste.isdigit():
|
||||
return int(reste)
|
||||
return None
|
||||
|
||||
|
||||
def nested_dir(rapport):
|
||||
"""Le dossier imbriqué de cette base, s'il en existe un."""
|
||||
chemin = os.path.join(rapport["root"], "filestore")
|
||||
return chemin if os.path.isdir(chemin) else ""
|
||||
|
||||
|
||||
def tidy_nested_plan(rapport):
|
||||
"""(à remonter, doublons) parmi les fichiers imbriqués.
|
||||
|
||||
Deux tas très différents : ce qui MANQUE au bon niveau doit y
|
||||
remonter, ce qui y est déjà est un doublon pur. Les traiter d'un
|
||||
bloc écraserait des fichiers présents par des copies — inutile, et
|
||||
inquiétant sur une base de production.
|
||||
"""
|
||||
dossier = nested_dir(rapport)
|
||||
if not dossier:
|
||||
return [], []
|
||||
bons, _n = files_on_disk(
|
||||
os.path.dirname(rapport["root"]), os.path.basename(rapport["root"])
|
||||
)
|
||||
remonter, doublons = [], []
|
||||
for deux in sorted(os.listdir(dossier)):
|
||||
profond = os.path.join(dossier, deux)
|
||||
if not os.path.isdir(profond):
|
||||
continue
|
||||
for nom in sorted(os.listdir(profond)):
|
||||
complet = os.path.join(profond, nom)
|
||||
if not os.path.isfile(complet):
|
||||
continue
|
||||
(doublons if f"{deux}/{nom}" in bons else remonter).append(
|
||||
(complet, os.path.join(rapport["root"], deux, nom))
|
||||
)
|
||||
return remonter, doublons
|
||||
|
||||
|
||||
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())
|
||||
547
script/analyse/check_module_package.py
Executable file
547
script/analyse/check_module_package.py
Executable file
|
|
@ -0,0 +1,547 @@
|
|||
#!/usr/bin/env python3
|
||||
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
||||
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
||||
|
||||
"""Ce qu'une base n'a pas, alors que l'installation par défaut l'aurait.
|
||||
|
||||
`conf/module_list_image_db_odoo.json` dit ce qu'ERPLibre embarque : le
|
||||
package `odoo18.0_base` suggère quinze modules. Rien ne vérifiait qu'une
|
||||
base DONNÉE les a. C'est pourtant la question qu'on se pose devant une
|
||||
instance migrée depuis la 12 : elle a traversé six paliers, des modules
|
||||
ont été désinstallés en chemin pour débloquer une mise à jour, et
|
||||
personne ne sait plus ce qui manque par rapport à une installation neuve.
|
||||
|
||||
Pourquoi pas `image_db.py --check_addons_exist`
|
||||
-----------------------------------------------
|
||||
Il répond à une autre question : le module est-il sur le DISQUE. Un
|
||||
module peut être présent dans le chemin des addons et absent de la base,
|
||||
ou l'inverse — connu de la base parce qu'il y fut installé, mais son code
|
||||
n'est plus là. Les deux outils sont complémentaires, aucun ne remplace
|
||||
l'autre.
|
||||
|
||||
Cinq verdicts, pas un seul « manquant »
|
||||
---------------------------------------
|
||||
Les confondre rendrait le rapport inutile, car l'action diffère à chaque
|
||||
fois : un module `available` s'installe d'un clic ; un module `unknown`
|
||||
demande d'abord de réparer le chemin des addons ; un `uninstallable` a
|
||||
une dépendance cassée qu'aucune installation ne contournera. Un rapport
|
||||
qui dit « 7 modules manquants » sans distinguer ces cas oblige à tout
|
||||
reprendre à la main.
|
||||
|
||||
Lecture seule, garantie par le serveur — on inspecte parfois des bases de
|
||||
migration dont c'est la seule copie.
|
||||
|
||||
Codes de sortie : 0 rien à signaler, 1 des trouvailles, 2 l'outil a échoué.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
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"
|
||||
|
||||
# Résolu depuis CE fichier, pas depuis le répertoire courant : l'outil est
|
||||
# lancé aussi bien depuis la racine du dépôt que depuis le menu TODO ou un
|
||||
# /tmp, et un chemin relatif le faisait échouer en disant « fichier de
|
||||
# packages illisible » — un diagnostic qui envoyait chercher au mauvais
|
||||
# endroit.
|
||||
REPO_ROOT = os.path.normpath(
|
||||
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..")
|
||||
)
|
||||
PACKAGE_FILE = os.path.join(
|
||||
REPO_ROOT, "conf", "module_list_image_db_odoo.json"
|
||||
)
|
||||
|
||||
# L'ordre EST la gravité : c'est lui qui décide de la lecture du rapport.
|
||||
VERDICTS = ("unknown", "uninstallable", "pending", "available", "installed")
|
||||
|
||||
# Les états d'`ir_module_module` qu'Odoo peut porter, rangés par verdict.
|
||||
# Une base en cours de mise à jour reste dans « to install » ou « to
|
||||
# remove » : ce n'est ni installé ni disponible, c'est inachevé, et le
|
||||
# dire évite de proposer d'installer ce qui est déjà en route.
|
||||
ETAT_VERS_VERDICT = {
|
||||
"installed": "installed",
|
||||
"to upgrade": "installed",
|
||||
"to install": "pending",
|
||||
"to remove": "pending",
|
||||
"uninstalled": "available",
|
||||
"uninstallable": "uninstallable",
|
||||
}
|
||||
|
||||
ICONE = {
|
||||
"installed": "✅",
|
||||
"available": "○",
|
||||
"pending": "⏳",
|
||||
"uninstallable": "⛔",
|
||||
"unknown": "❌",
|
||||
}
|
||||
|
||||
|
||||
def run_psql(database, sql):
|
||||
"""Interroger la base en lecture seule, garantie par le SERVEUR.
|
||||
|
||||
`default_transaction_read_only` n'est pas une promesse de l'outil :
|
||||
PostgreSQL refusera l'écriture même si le SQL en contenait une.
|
||||
"""
|
||||
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 read_packages(path=PACKAGE_FILE):
|
||||
"""Le fichier des packages, ou {} s'il est illisible."""
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as handle:
|
||||
dct = json.load(handle)
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
return dct if isinstance(dct, dict) else {}
|
||||
|
||||
|
||||
def package_chain(name, packages):
|
||||
"""De la racine à `name`, en suivant `base`.
|
||||
|
||||
Un package hérite du sien : `odoo12.0_website` est construit SUR
|
||||
l'image `odoo12.0_base`, donc ses modules s'ajoutent à ceux-là. Ne
|
||||
lire que le maillon nommé sous-estimerait ce qu'une installation
|
||||
embarque, et l'outil déclarerait « rien ne manque » à tort.
|
||||
|
||||
Une boucle dans `base` ferait tourner l'outil indéfiniment : on
|
||||
s'arrête au premier nom déjà vu plutôt que de faire confiance au
|
||||
fichier.
|
||||
"""
|
||||
chaine = []
|
||||
vus = set()
|
||||
courant = name
|
||||
# Borné par le NOMBRE de packages : une chaîne ne peut pas être plus
|
||||
# longue que ce qui existe. La terminaison ne dépend donc pas de la
|
||||
# garde `vus` — retirer l'une des deux donne un résultat FAUX, que le
|
||||
# test attrape, au lieu d'un outil qui pend, que rien n'attrape.
|
||||
for _ in range(len(packages) + 1):
|
||||
if not courant or courant not in packages or courant in vus:
|
||||
break
|
||||
vus.add(courant)
|
||||
chaine.append(courant)
|
||||
courant = (packages[courant] or {}).get("base") or ""
|
||||
chaine.reverse()
|
||||
return chaine
|
||||
|
||||
|
||||
def package_modules(name, packages):
|
||||
"""{module: package qui le suggère}, héritage compris.
|
||||
|
||||
Le package d'ORIGINE est retenu, pas le dernier vu : savoir qu'un
|
||||
module vient de `odoo16.0_base` et non de `odoo16.0_website` dit s'il
|
||||
est fondamental ou accessoire.
|
||||
"""
|
||||
trouve = {}
|
||||
for maillon in package_chain(name, packages):
|
||||
for groupe in (packages[maillon] or {}).get("image_list") or []:
|
||||
for module in (groupe or {}).get("module") or []:
|
||||
trouve.setdefault(module, maillon)
|
||||
return trouve
|
||||
|
||||
|
||||
def db_version(database):
|
||||
"""« 18.0 » d'après le module `base`, ou None.
|
||||
|
||||
C'est la version qu'Odoo lui-même inscrit, pas celle du checkout : une
|
||||
base de palier 15 lue depuis un checkout 18 doit se comparer au
|
||||
package 15, sans quoi le rapport nommerait des modules qui n'existaient
|
||||
pas encore.
|
||||
"""
|
||||
lignes = run_psql(
|
||||
database,
|
||||
"SELECT latest_version FROM ir_module_module WHERE name = 'base'",
|
||||
)
|
||||
if not lignes or not lignes[0][0]:
|
||||
return None
|
||||
morceaux = lignes[0][0].split(".")
|
||||
if len(morceaux) < 2:
|
||||
return None
|
||||
return f"{morceaux[0]}.{morceaux[1]}"
|
||||
|
||||
|
||||
def default_package(version):
|
||||
"""« 18.0 » -> « odoo18.0_base ». None si l'on ne sait pas."""
|
||||
return f"odoo{version}_base" if version else None
|
||||
|
||||
|
||||
def column_types(database, table):
|
||||
"""{colonne: type} pour une table. {} si la base ne répond pas."""
|
||||
lignes = run_psql(
|
||||
database,
|
||||
"SELECT column_name, data_type FROM information_schema.columns"
|
||||
f" WHERE table_name = '{table}'",
|
||||
)
|
||||
return {ligne[0]: ligne[1] for ligne in lignes or [] if len(ligne) >= 2}
|
||||
|
||||
|
||||
def as_text(colonne, types):
|
||||
"""Lire une colonne en texte, qu'elle soit varchar ou jsonb.
|
||||
|
||||
Les champs traduisibles d'Odoo sont passés en jsonb à la 16 :
|
||||
`shortdesc` est un varchar en 12-15 et un `{"en_US": "…"}` ensuite.
|
||||
Cet outil vise les bases de PALIER d'une migration, donc les deux
|
||||
formes se présentent dans la même session — supposer l'une fait
|
||||
échouer la requête entière sur l'autre, et l'outil déclare alors la
|
||||
base illisible alors qu'elle se porte bien.
|
||||
"""
|
||||
if types.get(colonne) == "jsonb":
|
||||
return (
|
||||
f"coalesce({colonne} ->> 'en_US', {colonne} ->> 'fr_FR',"
|
||||
f" {colonne} ->> 'fr_CA', '')"
|
||||
)
|
||||
return f"coalesce({colonne}, '')"
|
||||
|
||||
|
||||
def census(database):
|
||||
"""{module: (état, résumé, application, auteur)} pour TOUTE la base.
|
||||
|
||||
None si la base ne répond pas — à distinguer d'une base vide, qui
|
||||
rendrait un dictionnaire vide et ne veut pas dire la même chose.
|
||||
"""
|
||||
types = column_types(database, "ir_module_module")
|
||||
if "name" not in types:
|
||||
return None
|
||||
lignes = run_psql(
|
||||
database,
|
||||
f"SELECT name, state, {as_text('shortdesc', types)},"
|
||||
" case when application then '1' else '0' end,"
|
||||
f" {as_text('author', types)} FROM ir_module_module",
|
||||
)
|
||||
if lignes is None:
|
||||
return None
|
||||
return {
|
||||
ligne[0]: (ligne[1], ligne[2], ligne[3] == "1", ligne[4])
|
||||
for ligne in lignes
|
||||
if len(ligne) >= 5
|
||||
}
|
||||
|
||||
|
||||
def dependencies(database):
|
||||
"""{module: [ce dont il dépend]}. {} si la table ne répond pas."""
|
||||
lignes = run_psql(
|
||||
database,
|
||||
"SELECT m.name, d.name FROM ir_module_module_dependency d"
|
||||
" JOIN ir_module_module m ON m.id = d.module_id",
|
||||
)
|
||||
if not lignes:
|
||||
return {}
|
||||
dct = {}
|
||||
for ligne in lignes:
|
||||
if len(ligne) >= 2:
|
||||
dct.setdefault(ligne[0], []).append(ligne[1])
|
||||
return dct
|
||||
|
||||
|
||||
def verdict_of(module, connus):
|
||||
"""Le verdict d'un module suggéré, d'après ce que la base en sait."""
|
||||
if module not in connus:
|
||||
return "unknown"
|
||||
return ETAT_VERS_VERDICT.get(connus[module][0], "uninstallable")
|
||||
|
||||
|
||||
def blocking_dependencies(module, connus, depend):
|
||||
"""Ce qu'installer `module` réclamerait et que la base n'a pas.
|
||||
|
||||
Un module « disponible » ne l'est pas toujours vraiment : si trois de
|
||||
ses dépendances sont absentes du chemin des addons, l'installer
|
||||
échouera. Le dire ici évite de le découvrir en cliquant.
|
||||
"""
|
||||
manquantes = []
|
||||
for nom in sorted(set(depend.get(module) or [])):
|
||||
if verdict_of(nom, connus) in ("unknown", "uninstallable"):
|
||||
manquantes.append(nom)
|
||||
return manquantes
|
||||
|
||||
|
||||
def audit(database, package=None, packages=None, path=PACKAGE_FILE):
|
||||
"""Tout ce que le rapport a besoin de savoir, en une passe."""
|
||||
packages = read_packages(path) if packages is None else packages
|
||||
connus = census(database)
|
||||
if connus is None:
|
||||
return {"unavailable": True, "database": database}
|
||||
version = db_version(database)
|
||||
nom = package or default_package(version)
|
||||
suggere = package_modules(nom, packages) if nom else {}
|
||||
depend = dependencies(database)
|
||||
|
||||
lignes = []
|
||||
for module in sorted(suggere):
|
||||
verdict = verdict_of(module, connus)
|
||||
etat = connus.get(module, ("", "", False, ""))
|
||||
lignes.append(
|
||||
{
|
||||
"module": module,
|
||||
"from": suggere[module],
|
||||
"verdict": verdict,
|
||||
"state": etat[0],
|
||||
"shortdesc": etat[1],
|
||||
"needs": (
|
||||
blocking_dependencies(module, connus, depend)
|
||||
if verdict == "available"
|
||||
else []
|
||||
),
|
||||
}
|
||||
)
|
||||
par_etat = {}
|
||||
for etat, _desc, _app, _auteur in connus.values():
|
||||
par_etat[etat] = par_etat.get(etat, 0) + 1
|
||||
installes = {
|
||||
nom_mod
|
||||
for nom_mod, valeur in connus.items()
|
||||
if valeur[0] in ("installed", "to upgrade")
|
||||
}
|
||||
return {
|
||||
"database": database,
|
||||
"version": version,
|
||||
"package": nom,
|
||||
"package_known": bool(nom and nom in packages),
|
||||
"chain": package_chain(nom, packages) if nom else [],
|
||||
"lines": lignes,
|
||||
"by_state": par_etat,
|
||||
"total": len(connus),
|
||||
"installed": sorted(installes),
|
||||
"extra": sorted(installes - set(suggere)),
|
||||
"authors": authors_of(connus, installes),
|
||||
}
|
||||
|
||||
|
||||
def authors_of(connus, installes):
|
||||
"""[(auteur, combien)] parmi les modules installés, les gros d'abord.
|
||||
|
||||
« L'ensemble des modules » d'une base ne se lit pas en listant trois
|
||||
cents noms : l'auteur dit d'où ils viennent — Odoo, OCA, ou la maison.
|
||||
"""
|
||||
compte = {}
|
||||
for nom in installes:
|
||||
auteur = (connus[nom][3] or t("unknown author")).strip()
|
||||
compte[auteur] = compte.get(auteur, 0) + 1
|
||||
return sorted(compte.items(), key=lambda item: (-item[1], item[0]))
|
||||
|
||||
|
||||
def missing(rapport):
|
||||
"""Les lignes qui réclament une action, les plus graves en tête."""
|
||||
ordre = {verdict: rang for rang, verdict in enumerate(VERDICTS)}
|
||||
return sorted(
|
||||
(
|
||||
ligne
|
||||
for ligne in rapport.get("lines") or []
|
||||
if ligne["verdict"] != "installed"
|
||||
),
|
||||
key=lambda ligne: (ordre[ligne["verdict"]], ligne["module"]),
|
||||
)
|
||||
|
||||
|
||||
def installable(rapport):
|
||||
"""Les modules qu'on peut RÉELLEMENT installer, dans l'ordre affiché.
|
||||
|
||||
Seuls les « available » : la base les connaît et ils attendent. Un
|
||||
« unknown » n'est pas dans le chemin des addons — l'installer échoue
|
||||
avant de commencer ; un « uninstallable » a une dépendance cassée
|
||||
qu'aucune installation ne contournera ; un « pending » est déjà en
|
||||
route. Les proposer ferait une liste plus longue et trois échecs.
|
||||
"""
|
||||
return [
|
||||
ligne["module"]
|
||||
for ligne in missing(rapport)
|
||||
if ligne["verdict"] == "available"
|
||||
]
|
||||
|
||||
|
||||
def parse_selection(answer, candidates):
|
||||
"""(choisis, jetons refusés) d'après « 1 3 5 », « a », ou rien.
|
||||
|
||||
Les jetons refusés sont RENDUS, jamais avalés : demander cinq modules
|
||||
et en recevoir quatre sans que rien ne le dise est la pire issue —
|
||||
on croit l'installation complète. L'appelant doit pouvoir le montrer.
|
||||
|
||||
La virgule vaut l'espace : les listes affichées ailleurs par l'outil
|
||||
sont séparées par des virgules, et refuser « 1,3 » ne protégerait de
|
||||
rien tout en obligeant à retaper.
|
||||
"""
|
||||
reponse = (answer or "").strip()
|
||||
if not reponse:
|
||||
return [], []
|
||||
if reponse.lower() in ("a", "all"):
|
||||
return list(candidates), []
|
||||
choisis, refuses, vus = [], [], set()
|
||||
for jeton in reponse.replace(",", " ").split():
|
||||
if not jeton.isdigit():
|
||||
refuses.append(jeton)
|
||||
continue
|
||||
rang = int(jeton)
|
||||
if not 1 <= rang <= len(candidates):
|
||||
refuses.append(jeton)
|
||||
continue
|
||||
nom = candidates[rang - 1]
|
||||
if nom not in vus:
|
||||
vus.add(nom)
|
||||
choisis.append(nom)
|
||||
return choisis, refuses
|
||||
|
||||
|
||||
def render(rapport, limit=0):
|
||||
"""Le rapport, en clair. `limit` borne les listes longues (0 = tout)."""
|
||||
if rapport.get("unavailable"):
|
||||
return [
|
||||
f"❌ {t('Cannot read the database: ')}{rapport['database']}",
|
||||
]
|
||||
lignes = [
|
||||
f"📦 {t('Modules of')} {rapport['database']}"
|
||||
f" ({t('Odoo')} {rapport['version'] or '?'})",
|
||||
]
|
||||
if not rapport["package_known"]:
|
||||
# Ne pas se taire : un package inconnu rend TOUT module « manquant »,
|
||||
# et l'on croirait à une base vide plutôt qu'à un nom mal choisi.
|
||||
lignes.append(
|
||||
f" ⚠ {t('No default package known for this version')}"
|
||||
f" ({rapport['package'] or '?'})"
|
||||
)
|
||||
lignes.append(f" {t('Compared against nothing — census only.')}")
|
||||
else:
|
||||
lignes.append(
|
||||
f" {t('Default package:')} {' → '.join(rapport['chain'])}"
|
||||
f" ({len(rapport['lines'])} {t('suggested module(s)')})"
|
||||
)
|
||||
lignes.append("")
|
||||
|
||||
lignes.append(f" {t('Census')} : {rapport['total']} {t('known')}")
|
||||
for etat, combien in sorted(
|
||||
rapport["by_state"].items(), key=lambda item: -item[1]
|
||||
):
|
||||
lignes.append(f" {etat:<16} {combien:>5}")
|
||||
if rapport["authors"]:
|
||||
lignes.append("")
|
||||
lignes.append(f" {t('Installed modules by author')} :")
|
||||
for auteur, combien in rapport["authors"][: limit or None]:
|
||||
lignes.append(f" {auteur[:52]:<54} {combien:>4}")
|
||||
lignes.append("")
|
||||
|
||||
absents = missing(rapport)
|
||||
if not absents:
|
||||
if rapport["package_known"]:
|
||||
lignes.append(f" ✅ {t('Every suggested module is installed.')}")
|
||||
return lignes
|
||||
|
||||
lignes.append(
|
||||
f" {len(absents)} {t('suggested module(s) not installed')} :"
|
||||
)
|
||||
for verdict in VERDICTS:
|
||||
groupe = [ligne for ligne in absents if ligne["verdict"] == verdict]
|
||||
if not groupe:
|
||||
continue
|
||||
lignes.append(
|
||||
f" {ICONE[verdict]} {len(groupe)} {t(EXPLICATION[verdict])}"
|
||||
)
|
||||
for ligne in groupe[: limit or None]:
|
||||
detail = (
|
||||
f" — {ligne['shortdesc'][:44]}" if ligne["shortdesc"] else ""
|
||||
)
|
||||
lignes.append(f" {ligne['module']:<34}{detail}")
|
||||
if ligne["needs"]:
|
||||
lignes.append(
|
||||
f" ↳ {t('also needs')} :"
|
||||
f" {', '.join(ligne['needs'][:6])}"
|
||||
)
|
||||
if limit and len(groupe) > limit:
|
||||
lignes.append(f" … {len(groupe) - limit} {t('more')}")
|
||||
return lignes
|
||||
|
||||
|
||||
# Ce qu'il faut FAIRE, pas seulement ce que c'est : un rapport qui nomme
|
||||
# l'état sans nommer le geste laisse le travail entier à faire.
|
||||
EXPLICATION = {
|
||||
"unknown": "absent from the addons path — sync the repo first",
|
||||
"uninstallable": "present but broken — fix the dependency",
|
||||
"pending": "half-way — finish the pending update",
|
||||
"available": "known to the database — install it",
|
||||
"installed": "installed",
|
||||
}
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description=(
|
||||
"List every module of a database and report which ones the"
|
||||
" default ERPLibre package suggests but the database lacks."
|
||||
)
|
||||
)
|
||||
parser.add_argument("-d", "--database", help="database to inspect")
|
||||
parser.add_argument(
|
||||
"-p",
|
||||
"--package",
|
||||
help="package to compare against (default: odoo<version>_base)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--file", default=PACKAGE_FILE, help="package definition file"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--list-packages",
|
||||
action="store_true",
|
||||
help="print every known package and exit",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--limit",
|
||||
type=int,
|
||||
default=0,
|
||||
help="cap long lists (0 = no cap)",
|
||||
)
|
||||
parser.add_argument("--json", action="store_true", help="machine output")
|
||||
config = parser.parse_args(argv)
|
||||
|
||||
packages = read_packages(config.file)
|
||||
if not packages:
|
||||
print(f"❌ {t('Cannot read the package file: ')}{config.file}")
|
||||
return 2
|
||||
if config.list_packages:
|
||||
for nom in sorted(packages):
|
||||
combien = len(package_modules(nom, packages))
|
||||
marque = (
|
||||
" (disabled)" if (packages[nom] or {}).get("disable") else ""
|
||||
)
|
||||
print(f"{nom:<46} {combien:>4} module(s){marque}")
|
||||
return 0
|
||||
if not config.database:
|
||||
parser.error("--database is required (or use --list-packages)")
|
||||
|
||||
rapport = audit(config.database, package=config.package, packages=packages)
|
||||
if rapport.get("unavailable"):
|
||||
print(f"❌ {t('Cannot read the database: ')}{config.database}")
|
||||
return 2
|
||||
if config.json:
|
||||
print(
|
||||
json.dumps(rapport, indent=2, sort_keys=True, ensure_ascii=False)
|
||||
)
|
||||
else:
|
||||
print("\n".join(render(rapport, limit=config.limit)))
|
||||
return 1 if missing(rapport) else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -7,9 +7,15 @@ import configparser
|
|||
import getpass
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import subprocess
|
||||
from subprocess import check_output
|
||||
|
||||
sys.path.append(
|
||||
os.path.normpath(os.path.join(os.path.dirname(__file__), "..", ".."))
|
||||
)
|
||||
|
||||
logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
|
@ -78,6 +84,71 @@ def get_master_password():
|
|||
_logger.error("Password echoed, danger!")
|
||||
|
||||
|
||||
# Assez pour une faute de frappe répétée, pas assez pour qu'une boucle
|
||||
# oubliée tourne toute la nuit devant une invite que personne ne lit.
|
||||
MAX_ESSAIS_MOT_DE_PASSE = 10
|
||||
|
||||
|
||||
def password_refused(sortie):
|
||||
"""Odoo a-t-il refusé le mot de passe maître, ou autre chose ?
|
||||
|
||||
La distinction porte tout. Reposer la question sur n'importe quel
|
||||
échec cacherait la vraie panne derrière dix invites, et l'on
|
||||
chercherait un mot de passe alors que la base est cassée.
|
||||
|
||||
Odoo lève `AccessDenied` — la classe apparaît dans la trace, et son
|
||||
message traduit peut varier. On reconnaît donc la CLASSE.
|
||||
"""
|
||||
return "AccessDenied" in (sortie or "")
|
||||
|
||||
|
||||
def probe_master_password(arg_base):
|
||||
"""(accepté, sortie) — éprouver le mot de passe sur `--list`.
|
||||
|
||||
La commande la plus inoffensive : elle ne touche à rien et rend le
|
||||
même refus qu'une restauration. Valider ici évite d'échouer à
|
||||
mi-parcours, une fois la base déjà supprimée.
|
||||
"""
|
||||
done = subprocess.run(
|
||||
f"{arg_base} --list".split(" "),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
return done.returncode == 0, (done.stdout or "") + (done.stderr or "")
|
||||
|
||||
|
||||
def ask_master_password(arg_base, essais=MAX_ESSAIS_MOT_DE_PASSE):
|
||||
"""Le mot de passe maître, redemandé tant qu'Odoo le refuse.
|
||||
|
||||
None si l'on renonce — invite vide, essais épuisés, ou panne qui
|
||||
n'a rien à voir avec le mot de passe.
|
||||
|
||||
Une faute de frappe arrêtait la migration net, sur une trace
|
||||
`CalledProcessError` que rien n'attrapait. Après une heure de
|
||||
paliers, c'est cher payé pour une lettre.
|
||||
"""
|
||||
for tour in range(1, essais + 1):
|
||||
mot = get_master_password()
|
||||
if not mot:
|
||||
return None
|
||||
candidat = f"{arg_base} --master_password={mot}"
|
||||
accepte, sortie = probe_master_password(candidat)
|
||||
if accepte:
|
||||
return mot
|
||||
if not password_refused(sortie):
|
||||
# Autre chose est cassé : le dire, et ne pas noyer la panne
|
||||
# sous dix invites de mot de passe.
|
||||
_logger.error(sortie.strip()[-1500:])
|
||||
return None
|
||||
restants = essais - tour
|
||||
if restants:
|
||||
_logger.warning(
|
||||
f"Master password refused, {restants} attempt(s) left."
|
||||
)
|
||||
_logger.error("Master password refused too many times.")
|
||||
return None
|
||||
|
||||
|
||||
def get_list_db_cache(arg_base):
|
||||
arg = f"{arg_base} --list"
|
||||
out = check_output(arg.split(" ")).decode()
|
||||
|
|
@ -86,6 +157,109 @@ def get_list_db_cache(arg_base):
|
|||
return lst_db, lst_db_cache
|
||||
|
||||
|
||||
def verify_filestore(database, image):
|
||||
"""Contrôler qu'une restauration a bien posé ses fichiers.
|
||||
|
||||
Une seule fois, à la restauration d'origine. Après un clone il n'y a
|
||||
rien à vérifier : `copytree` recopie la source telle quelle, défauts
|
||||
compris — le contrôle appartient à ce qui a créé le défaut, pas à ce
|
||||
qui l'a dupliqué.
|
||||
|
||||
Le contrôle n'interrompt pas : la base est restaurée et utilisable,
|
||||
c'est la DISPOSITION des fichiers qui est suspecte. Refuser ici
|
||||
casserait des chaînes qui marchent, pour un défaut qui se répare
|
||||
d'une commande.
|
||||
"""
|
||||
chemin = os.path.join("image_db", f"{image}.zip")
|
||||
if not os.path.isfile(chemin):
|
||||
return
|
||||
try:
|
||||
from script.analyse import check_filestore
|
||||
except Exception: # pragma: no cover - l'outil d'analyse est optionnel
|
||||
return
|
||||
rapport = check_filestore.verify_restore(database, chemin)
|
||||
for ligne in check_filestore.render_verify(rapport):
|
||||
print(ligne)
|
||||
if rapport.get("nested"):
|
||||
offer_tidy(check_filestore, rapport)
|
||||
|
||||
|
||||
def offer_tidy(check_filestore, rapport):
|
||||
"""Proposer de ranger TOUT DE SUITE, là où le défaut naît.
|
||||
|
||||
C'est le seul endroit qui vaille. Le nichage se produit une fois, à
|
||||
la restauration, puis le clone le recopie tel quel : mesuré, les six
|
||||
bases de la chaîne portaient les mêmes 1168 fichiers. Ranger ici,
|
||||
c'est ranger une fois ; ranger plus tard, c'est six fois.
|
||||
|
||||
Rien ne se fait sans réponse humaine, et rien du tout hors d'un
|
||||
terminal : ce script tourne aussi sans personne devant, et une
|
||||
question posée à un `stdin` fermé arrêterait la migration.
|
||||
"""
|
||||
if not sys.stdin.isatty():
|
||||
return
|
||||
remonter, doublons = check_filestore.tidy_nested_plan(rapport)
|
||||
if not remonter and not doublons:
|
||||
return
|
||||
print(f" {len(remonter)} à remonter, {len(doublons)} doublons purs")
|
||||
try:
|
||||
reponse = input("💬 Ranger maintenant ? (y/N) : ").strip().lower()
|
||||
except EOFError:
|
||||
return
|
||||
if reponse not in ("y", "yes", "o"):
|
||||
return
|
||||
for source, cible in remonter:
|
||||
os.makedirs(os.path.dirname(cible), exist_ok=True)
|
||||
shutil.move(source, cible)
|
||||
for source, _cible in doublons:
|
||||
os.remove(source)
|
||||
shutil.rmtree(check_filestore.nested_dir(rapport), ignore_errors=True)
|
||||
print(f"✅ {len(remonter)} remontés, {len(doublons)} doublons supprimés.")
|
||||
|
||||
|
||||
def restore_or_clone(config, arg_base, cache_database, lst_db_cache):
|
||||
"""Restaurer depuis l'image, ou cloner le cache déjà restauré.
|
||||
|
||||
Le contrôle du filestore ne suit QUE les vraies restaurations. Le
|
||||
clone recopie sa source telle quelle : contrôler le miroir dirait
|
||||
deux fois la même chose, et la seconde au mauvais endroit.
|
||||
"""
|
||||
if cache_database not in lst_db_cache and not config.ignore_cache:
|
||||
_logger.info(
|
||||
f"## Create cache {cache_database} from image {config.image} ##"
|
||||
)
|
||||
arg = (
|
||||
f"{arg_base} --restore"
|
||||
f" --restore_image {config.image} --database {cache_database}"
|
||||
)
|
||||
print(check_output(arg.split(" ")).decode())
|
||||
verify_filestore(cache_database, config.image)
|
||||
|
||||
if config.ignore_cache:
|
||||
_logger.info(
|
||||
f"## Restoring {config.image} to database {config.database} ##"
|
||||
)
|
||||
arg = (
|
||||
f"{arg_base} --restore --restore_image"
|
||||
f" {config.image} --database {config.database}"
|
||||
)
|
||||
else:
|
||||
_logger.info(
|
||||
f"## Clone cache {cache_database} to database"
|
||||
f" {config.database} ##"
|
||||
)
|
||||
arg = (
|
||||
f"{arg_base} --clone --from_database"
|
||||
f" {cache_database} --database {config.database}"
|
||||
)
|
||||
if config.neutralize:
|
||||
arg += " --neutralize"
|
||||
print(arg)
|
||||
print(check_output(arg.split(" ")).decode())
|
||||
if config.ignore_cache:
|
||||
verify_filestore(config.database, config.image)
|
||||
|
||||
|
||||
def main():
|
||||
config = get_config()
|
||||
|
||||
|
|
@ -109,12 +283,11 @@ def main():
|
|||
|
||||
has_admin_password = config_parser.get("options", "admin_passwd")
|
||||
if has_admin_password and has_admin_password != "admin":
|
||||
master_password = get_master_password()
|
||||
master_password = ask_master_password(arg_base)
|
||||
if not master_password:
|
||||
_logger.error("Missing master password, cancel transaction.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
arg_base += f" --master_password={master_password}"
|
||||
arg_base += f" --master_password={master_password}"
|
||||
else:
|
||||
_logger.info("No master password needed... Continue")
|
||||
|
||||
|
|
@ -139,40 +312,7 @@ def main():
|
|||
print(out)
|
||||
if config.only_drop:
|
||||
return
|
||||
# Check cache exist
|
||||
if cache_database not in lst_db_cache and not config.ignore_cache:
|
||||
_logger.info(
|
||||
f"## Create cache {cache_database} from image"
|
||||
f" {config.image} ##"
|
||||
)
|
||||
arg = (
|
||||
f"{arg_base} --restore"
|
||||
f" --restore_image {config.image} --database {cache_database}"
|
||||
)
|
||||
out = check_output(arg.split(" ")).decode()
|
||||
print(out)
|
||||
# Clone database
|
||||
if config.ignore_cache:
|
||||
_logger.info(
|
||||
f"## Restoring {config.image} to database {config.database} ##"
|
||||
)
|
||||
arg = (
|
||||
f"{arg_base} --restore --restore_image"
|
||||
f" {config.image} --database {config.database}"
|
||||
)
|
||||
else:
|
||||
_logger.info(
|
||||
f"## Clone cache {cache_database} to database {config.database} ##"
|
||||
)
|
||||
arg = (
|
||||
f"{arg_base} --clone --from_database"
|
||||
f" {cache_database} --database {config.database}"
|
||||
)
|
||||
if config.neutralize:
|
||||
arg += " --neutralize"
|
||||
print(arg)
|
||||
out = check_output(arg.split(" ")).decode()
|
||||
print(out)
|
||||
restore_or_clone(config, arg_base, cache_database, lst_db_cache)
|
||||
|
||||
if not config.clean_cache and not config.database:
|
||||
print("Nothing to do.")
|
||||
|
|
|
|||
|
|
@ -10166,13 +10166,18 @@ class TODO:
|
|||
print(t("Command not found !"))
|
||||
|
||||
def prompt_execute_analyse(self):
|
||||
"""Analyses en lecture seule d'une base Odoo.
|
||||
"""Analyses d'une base Odoo, en lecture seule.
|
||||
|
||||
Aucune entrée de ce menu n'écrit : la connexion psql est ouverte avec
|
||||
`default_transaction_read_only=on`, donc c'est le serveur qui refuse
|
||||
toute écriture, pas une promesse du code.
|
||||
Toute LECTURE passe par une connexion psql ouverte avec
|
||||
`default_transaction_read_only=on` : c'est le serveur qui refuse
|
||||
l'écriture, pas une promesse du code.
|
||||
|
||||
Une seule action écrit — installer les modules suggérés, à la fin
|
||||
de l'analyse [5]. Elle ne part jamais seule : question explicite,
|
||||
défaut à « non », liste à confirmer, et refus net si le checkout
|
||||
n'est pas sur la version de la base.
|
||||
"""
|
||||
print(f"🤖 {t('Analyse a database, without ever writing to it!')}")
|
||||
print(f"🤖 {t('Analyse a database. Reading never writes.')}")
|
||||
choices = [
|
||||
{"section": t("Structure")},
|
||||
{"prompt_description": t("Tables and database size")},
|
||||
|
|
@ -10185,6 +10190,18 @@ class TODO:
|
|||
{"prompt_description": t("Studio and hand-made x_ fields")},
|
||||
{"section": t("Migration")},
|
||||
{"prompt_description": t("Quality of a migration, step by step")},
|
||||
{"section": t("Modules")},
|
||||
{
|
||||
"prompt_description": t(
|
||||
"Modules missing from the default package"
|
||||
)
|
||||
},
|
||||
{"section": t("Files")},
|
||||
{
|
||||
"prompt_description": t(
|
||||
"Attachment files missing from the filestore"
|
||||
)
|
||||
},
|
||||
]
|
||||
help_info = self.fill_help_info(choices)
|
||||
|
||||
|
|
@ -10201,9 +10218,288 @@ class TODO:
|
|||
self.execute_analyse_custom_field()
|
||||
elif status == "4":
|
||||
self.execute_analyse_migration_quality()
|
||||
elif status == "5":
|
||||
self.execute_analyse_module_package()
|
||||
elif status == "6":
|
||||
self.execute_analyse_filestore()
|
||||
else:
|
||||
print(t("Command not found !"))
|
||||
|
||||
def execute_analyse_module_package(self):
|
||||
"""Ce que la base n'a pas, alors que l'installation par défaut l'a.
|
||||
|
||||
Pas de choix « sauvegarde .zip » ici, contrairement aux autres
|
||||
analyses : l'outil interroge `ir_module_module`, qu'un zip
|
||||
n'expose pas sans restauration. Proposer l'option pour la refuser
|
||||
ensuite ferait perdre le temps de la choisir.
|
||||
"""
|
||||
from script.analyse import check_module_package as modules
|
||||
|
||||
database = self._analyse_select_database()
|
||||
if not database:
|
||||
return
|
||||
try:
|
||||
rapport = modules.audit(database)
|
||||
except Exception as exc:
|
||||
print(f"❌ {t('Analysis failed: ')}{exc}")
|
||||
return
|
||||
if rapport.get("unavailable"):
|
||||
print(f"❌ {t('Cannot read the database: ')}{database}")
|
||||
return
|
||||
print("\n".join(modules.render(rapport, limit=8)))
|
||||
self._analyse_offer_install(database, rapport)
|
||||
|
||||
def handler(rank):
|
||||
if rank == 1:
|
||||
print("\n".join(modules.render(rapport, limit=0)))
|
||||
elif rank == 2:
|
||||
for nom in sorted(modules.read_packages()):
|
||||
print(f" {nom}")
|
||||
else:
|
||||
self._analyse_export_json(
|
||||
rapport, os.path.basename(database), "module_package"
|
||||
)
|
||||
|
||||
self._analyse_follow_up(
|
||||
[
|
||||
{"prompt_description": t("Show every entry")},
|
||||
{"prompt_description": t("List the known packages")},
|
||||
{"prompt_description": t("Export as JSON")},
|
||||
],
|
||||
handler,
|
||||
)
|
||||
|
||||
def execute_analyse_filestore(self):
|
||||
"""Ce qui manque au filestore, et ce qu'on peut encore récupérer.
|
||||
|
||||
Pas d'option « sauvegarde .zip » : l'outil compare une BASE à son
|
||||
filestore, et un zip porte les deux ensemble par construction —
|
||||
il n'y a rien à y trouver.
|
||||
"""
|
||||
from script.analyse import check_filestore as filestore
|
||||
|
||||
database = self._analyse_select_database()
|
||||
if not database:
|
||||
return
|
||||
print(f"⧖ {t('Scanning filestores and backups…')}")
|
||||
try:
|
||||
rapport = filestore.audit(database)
|
||||
except Exception as exc:
|
||||
print(f"❌ {t('Analysis failed: ')}{exc}")
|
||||
return
|
||||
if rapport.get("unavailable"):
|
||||
print(f"❌ {t('Cannot read the database: ')}{database}")
|
||||
return
|
||||
etat = {"rapport": rapport}
|
||||
print("\n".join(filestore.render(rapport, limit=20)))
|
||||
|
||||
def relire():
|
||||
"""Relire APRÈS une réparation.
|
||||
|
||||
Sans cela « Tout afficher » rejouait le rapport d'avant :
|
||||
on purgeait, on relisait, et l'on voyait encore ce qui
|
||||
venait de disparaître. Pire, on repurgeait des lignes déjà
|
||||
effacées en croyant le travail inachevé.
|
||||
"""
|
||||
print(f"⧖ {t('Scanning filestores and backups…')}")
|
||||
etat["rapport"] = filestore.audit(database)
|
||||
|
||||
def handler(rank):
|
||||
if rank == 1:
|
||||
print("\n".join(filestore.render(etat["rapport"], limit=0)))
|
||||
elif rank == 2:
|
||||
if self._filestore_purge_dead(database, etat["rapport"]):
|
||||
relire()
|
||||
elif rank == 3:
|
||||
if self._filestore_tidy_nested(etat["rapport"]):
|
||||
relire()
|
||||
else:
|
||||
self._analyse_export_json(
|
||||
etat["rapport"], os.path.basename(database), "filestore"
|
||||
)
|
||||
|
||||
self._analyse_follow_up(
|
||||
[
|
||||
{"prompt_description": t("Show every entry")},
|
||||
{
|
||||
"prompt_description": t(
|
||||
"🧹 Purge attachments whose field no longer exists"
|
||||
)
|
||||
},
|
||||
{
|
||||
"prompt_description": t(
|
||||
"🧹 Tidy the nested filestore Odoo never reads"
|
||||
)
|
||||
},
|
||||
{"prompt_description": t("Export as JSON")},
|
||||
],
|
||||
handler,
|
||||
)
|
||||
|
||||
def _filestore_purge_dead(self, database, rapport):
|
||||
"""Effacer les pièces jointes dont le champ a disparu.
|
||||
|
||||
La seule ÉCRITURE en base de tout le menu Analyse. Elle porte sur
|
||||
des lignes que plus rien ne lit — `res.country.image` est devenu
|
||||
`image_url`, calculé, en 13 — mais elle reste une suppression :
|
||||
question explicite, défaut à « non », et le compte est relu avant
|
||||
de partir.
|
||||
"""
|
||||
from script.analyse import check_filestore as filestore
|
||||
from script.todo import auto_ask
|
||||
|
||||
lignes = rapport["groups"]["dead_field"]
|
||||
sql = filestore.purge_dead_sql(rapport)
|
||||
if not sql:
|
||||
print(f"ℹ️ {t('Nothing to purge.')}")
|
||||
return False
|
||||
print()
|
||||
for texte in filestore.summarise(lignes):
|
||||
print(f" {texte}")
|
||||
question = (
|
||||
f"💬 {t('Delete these')} {len(lignes)}"
|
||||
f" {t('attachment row(s) for good?')} (y/N) : "
|
||||
)
|
||||
if auto_ask.ask(question, default="n").strip().lower() not in (
|
||||
"y",
|
||||
"yes",
|
||||
"o",
|
||||
):
|
||||
print(f"ℹ️ {t('Nothing was deleted.')}")
|
||||
return False
|
||||
status, sortie = self.execute.exec_command_live(
|
||||
f'psql -d {database} -c "{sql}"',
|
||||
source_erplibre=False,
|
||||
single_source_erplibre=True,
|
||||
return_status_and_output=True,
|
||||
)
|
||||
if status:
|
||||
print(f"❌ {t('The purge failed.')}")
|
||||
return False
|
||||
# Le nombre ANNONCÉ par PostgreSQL, pas celui qu'on espérait :
|
||||
# rejouer une purge déjà faite rendait « DELETE 0 » et l'outil
|
||||
# se félicitait quand même d'avoir supprimé.
|
||||
efface = filestore.rows_deleted(sortie)
|
||||
if efface is None:
|
||||
print(f"⚠ {t('The purge ran but said nothing.')}")
|
||||
return True
|
||||
print(f"✅ {efface} {t('attachment row(s) deleted.')}")
|
||||
return True
|
||||
|
||||
def _filestore_tidy_nested(self, rapport):
|
||||
"""Remonter ce qui manque, effacer les doublons purs.
|
||||
|
||||
Deux tas, deux gestes. Écraser un fichier présent par une copie
|
||||
identique ne gagnerait rien et brouillerait la trace ; c'est
|
||||
pourquoi les doublons sont comptés à part et jamais déplacés.
|
||||
"""
|
||||
from script.analyse import check_filestore as filestore
|
||||
from script.todo import auto_ask
|
||||
|
||||
remonter, doublons = filestore.tidy_nested_plan(rapport)
|
||||
if not remonter and not doublons:
|
||||
print(f"ℹ️ {t('No nested filestore to tidy.')}")
|
||||
return False
|
||||
print()
|
||||
print(f" {len(remonter)} {t('file(s) to move up')}")
|
||||
print(f" {len(doublons)} {t('pure duplicate(s) to delete')}")
|
||||
print(f" {t('Directory:')} {filestore.nested_dir(rapport)}")
|
||||
if auto_ask.ask(
|
||||
f"💬 {t('Go ahead?')} (y/N) : ", default="n"
|
||||
).strip().lower() not in ("y", "yes", "o"):
|
||||
print(f"ℹ️ {t('Nothing was moved.')}")
|
||||
return False
|
||||
deplaces, effaces = 0, 0
|
||||
for source, cible in remonter:
|
||||
os.makedirs(os.path.dirname(cible), exist_ok=True)
|
||||
shutil.move(source, cible)
|
||||
deplaces += 1
|
||||
for source, _cible in doublons:
|
||||
os.remove(source)
|
||||
effaces += 1
|
||||
dossier = filestore.nested_dir(rapport)
|
||||
if dossier:
|
||||
shutil.rmtree(dossier, ignore_errors=True)
|
||||
print(
|
||||
f"✅ {deplaces} {t('moved up')}, {effaces}"
|
||||
f" {t('duplicate(s) removed')}."
|
||||
)
|
||||
return True
|
||||
|
||||
def _analyse_offer_install(self, database, rapport):
|
||||
"""Proposer d'installer ce qui manque, quand c'est installable.
|
||||
|
||||
Seuls les modules « available » sont offerts. Le dire est
|
||||
nécessaire : le rapport vient d'en annoncer onze, la liste n'en
|
||||
montre qu'un, et sans un mot on croirait à un bogue.
|
||||
|
||||
C'est la seule écriture de tout le menu Analyse, d'où trois
|
||||
garde-fous : la version du checkout doit être celle de la base —
|
||||
un Odoo 18 lancé sur une base 12 la réécrit avant d'échouer —, la
|
||||
question par défaut est « non », et la liste choisie est
|
||||
confirmée avant que rien ne parte.
|
||||
"""
|
||||
from script.analyse import check_module_package as modules
|
||||
from script.odoo.migration import database_cleanup
|
||||
from script.todo import auto_ask
|
||||
|
||||
candidats = modules.installable(rapport)
|
||||
if not candidats:
|
||||
return
|
||||
autres = len(modules.missing(rapport)) - len(candidats)
|
||||
|
||||
souci = database_cleanup.require_matching_version(database)
|
||||
if souci:
|
||||
print(f"\n⚠ {souci}")
|
||||
print(f" {t('Cannot install from here.')}")
|
||||
return
|
||||
|
||||
print()
|
||||
detail = f" ({autres} {t('need repair first')})" if autres else ""
|
||||
question = (
|
||||
f"💬 {t('Install some of the')} {len(candidats)}"
|
||||
f" {t('suggested module(s) waiting in this database?')}{detail}"
|
||||
f" (y/N) : "
|
||||
)
|
||||
if auto_ask.ask(question, default="n").strip().lower() not in (
|
||||
"y",
|
||||
"yes",
|
||||
"o",
|
||||
):
|
||||
return
|
||||
|
||||
print()
|
||||
for rang, nom in enumerate(candidats, start=1):
|
||||
print(f" [{rang}] {nom}")
|
||||
print(f" [a] {t('every one of them')}")
|
||||
print(f" {t('Enter = cancel')}")
|
||||
choisis, refuses = modules.parse_selection(
|
||||
auto_ask.ask(f"💬 {t('Numbers, space separated:')} ", default=""),
|
||||
candidats,
|
||||
)
|
||||
# Un jeton refusé n'est JAMAIS avalé : en demander cinq et en
|
||||
# recevoir quatre sans un mot ferait croire l'installation faite.
|
||||
if refuses:
|
||||
print(f"⚠ {t('Ignored, not in the list:')} {' '.join(refuses)}")
|
||||
if not choisis:
|
||||
print(f"ℹ️ {t('Nothing selected.')}")
|
||||
return
|
||||
|
||||
print()
|
||||
print(f" {t('About to install into')} {database} :")
|
||||
print(f" {', '.join(choisis)}")
|
||||
if auto_ask.ask(
|
||||
f"💬 {t('Go ahead?')} (y/N) : ", default="n"
|
||||
).strip().lower() not in ("y", "yes", "o"):
|
||||
print(f"ℹ️ {t('Nothing selected.')}")
|
||||
return
|
||||
self.execute.exec_command_live(
|
||||
f"./script/addons/install_addons.sh {database}"
|
||||
f" {','.join(choisis)}",
|
||||
source_erplibre=False,
|
||||
single_source_erplibre=True,
|
||||
)
|
||||
|
||||
def execute_analyse_migration_quality(self):
|
||||
"""Ce qu'une migration a gagné et perdu, palier par palier.
|
||||
|
||||
|
|
@ -10374,6 +10670,8 @@ class TODO:
|
|||
|
||||
def run(**kwargs):
|
||||
try:
|
||||
# Une sauvegarde compare déjà ses copies COW à la lecture :
|
||||
# les deux arch sont dans le dump, il n'y a rien à demander.
|
||||
state["data"] = (
|
||||
analyse.collect_from_backup(database)
|
||||
if is_backup
|
||||
|
|
@ -10430,6 +10728,19 @@ class TODO:
|
|||
print(analyse.render(data, verbose=True, hints=False))
|
||||
|
||||
lst_choice = [{"prompt_description": t("Show every view")}]
|
||||
# La comparaison des copies COW n'a besoin d'aucun registre : les deux
|
||||
# arch sont dans la base, appariées par leur clé. Elle est donc offerte
|
||||
# partout, y compris sur une sauvegarde et sur une base dont la version
|
||||
# diffère du checkout — là où l'autre comparaison est refusée.
|
||||
has_cow = bool(state["data"]["counts"].get("website_cow_copy"))
|
||||
if has_cow:
|
||||
lst_choice.append(
|
||||
{
|
||||
"prompt_description": t(
|
||||
"Compare the website copies with the view they shadow"
|
||||
)
|
||||
}
|
||||
)
|
||||
if can_compare:
|
||||
lst_choice += [
|
||||
{
|
||||
|
|
@ -10446,19 +10757,35 @@ class TODO:
|
|||
]
|
||||
lst_choice.append({"prompt_description": t("Export as JSON")})
|
||||
|
||||
def compare_cow():
|
||||
if not run(with_cow_diff=True):
|
||||
return
|
||||
state["tried"] = True
|
||||
data = state["data"]
|
||||
print(analyse.render(data, hints=False))
|
||||
n = data.get("n_cow_compared") or 0
|
||||
n_diff = len([r for r in data["findings"] if r.get("differs")])
|
||||
print(
|
||||
f" {n} {t('website copies compared with their module view,')}"
|
||||
f" {n_diff} {t('differ.')}"
|
||||
)
|
||||
|
||||
def handler(rank):
|
||||
data = state["data"]
|
||||
offset = 1 if has_cow else 0
|
||||
if rank == 1:
|
||||
print(analyse.render(data, verbose=True, hints=False))
|
||||
elif not can_compare or rank == len(lst_choice):
|
||||
elif has_cow and rank == 2:
|
||||
compare_cow()
|
||||
elif rank == len(lst_choice):
|
||||
self._analyse_export_json(
|
||||
data, os.path.basename(database), "view_custom"
|
||||
)
|
||||
elif rank == 2:
|
||||
elif can_compare and rank == 2 + offset:
|
||||
compare("flagged")
|
||||
elif rank == 3:
|
||||
elif can_compare and rank == 3 + offset:
|
||||
compare("all")
|
||||
elif rank == 4:
|
||||
elif can_compare and rank == 4 + offset:
|
||||
browse()
|
||||
|
||||
self._analyse_follow_up(lst_choice, handler)
|
||||
|
|
|
|||
|
|
@ -4091,6 +4091,50 @@ TRANSLATIONS = {
|
|||
"en": "🔬 Analyse - Odoo database analysis",
|
||||
},
|
||||
"Analyse": {"fr": "Analyse", "en": "Analysis"},
|
||||
"Analyse a database. Reading never writes.": {
|
||||
"fr": "Analyser une base ; la lecture n'écrit jamais.",
|
||||
"en": "Analyse a database. Reading never writes.",
|
||||
},
|
||||
"About to install into": {
|
||||
"fr": "Sur le point d'installer dans",
|
||||
"en": "About to install into",
|
||||
},
|
||||
"Cannot install from here.": {
|
||||
"fr": "Installation impossible d'ici.",
|
||||
"en": "Cannot install from here.",
|
||||
},
|
||||
"Enter = cancel": {
|
||||
"fr": "Entrée = annuler",
|
||||
"en": "Enter = cancel",
|
||||
},
|
||||
"Go ahead?": {
|
||||
"fr": "On y va ?",
|
||||
"en": "Go ahead?",
|
||||
},
|
||||
"Ignored, not in the list:": {
|
||||
"fr": "Ignorés, absents de la liste :",
|
||||
"en": "Ignored, not in the list:",
|
||||
},
|
||||
"Install some of the": {
|
||||
"fr": "Installer une partie des",
|
||||
"en": "Install some of the",
|
||||
},
|
||||
"Numbers, space separated:": {
|
||||
"fr": "Numéros, séparés par des espaces :",
|
||||
"en": "Numbers, space separated:",
|
||||
},
|
||||
"every one of them": {
|
||||
"fr": "tous",
|
||||
"en": "every one of them",
|
||||
},
|
||||
"need repair first": {
|
||||
"fr": "à réparer d'abord",
|
||||
"en": "need repair first",
|
||||
},
|
||||
"suggested module(s) waiting in this database?": {
|
||||
"fr": "module(s) suggéré(s) qui attendent dans cette base ?",
|
||||
"en": "suggested module(s) waiting in this database?",
|
||||
},
|
||||
"Analyse a database, without ever writing to it!": {
|
||||
"fr": "Analyser une base, sans jamais y écrire !",
|
||||
"en": "Analyse a database, without ever writing to it!",
|
||||
|
|
@ -6071,12 +6115,12 @@ TRANSLATIONS = {
|
|||
"en": "📦 Modules missing from the default package",
|
||||
},
|
||||
"Show every entry": {
|
||||
"fr": "Tout afficher",
|
||||
"en": "Show every entry",
|
||||
"fr": "📜 Tout afficher",
|
||||
"en": "📜 Show every entry",
|
||||
},
|
||||
"List the known packages": {
|
||||
"fr": "Lister les packages connus",
|
||||
"en": "List the known packages",
|
||||
"fr": "📚 Lister les packages connus",
|
||||
"en": "📚 List the known packages",
|
||||
},
|
||||
"Already repaired: the group exists.": {
|
||||
"fr": "Déjà réparé : le groupe existe.",
|
||||
|
|
|
|||
|
|
@ -2825,6 +2825,7 @@ class TodoUpgrade:
|
|||
f"✨ {t('Re-update i18n, purge the data and the tables')}"
|
||||
f" ({t('except mail_test and mail_test_full')})"
|
||||
)
|
||||
self.prompt_purge_dead_attachments(database_name_upgrade)
|
||||
# waiting_input = self.ask("💬print Press any keyboard key to continue...")
|
||||
msg = "6 - Migration finished"
|
||||
self.print_step(msg)
|
||||
|
|
|
|||
|
|
@ -268,3 +268,93 @@ class TestRender(unittest.TestCase):
|
|||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
|
||||
class TestCowTwinDiff(unittest.TestCase):
|
||||
"""Comparer une copie de site web à la vue de module qu'elle masque.
|
||||
|
||||
C'est LA comparaison qui compte pour une copie, et elle n'a besoin d'aucun
|
||||
registre : les deux arch sont dans la base, appariées par la clé. Elle
|
||||
marche donc là où la comparaison avec la source du module est refusée —
|
||||
une base dont la version diffère du checkout, et une sauvegarde .zip.
|
||||
"""
|
||||
|
||||
def finding(self, **override):
|
||||
row = view(
|
||||
id=10, key="website.homepage", website_id=1, has_module_twin=True
|
||||
)
|
||||
row["category"], row["reason"] = A.classify(row)
|
||||
row.update(override)
|
||||
return row
|
||||
|
||||
def test_a_copy_that_differs_is_measured(self):
|
||||
row = self.finding()
|
||||
n = A.attach_cow_twin_diff(
|
||||
[row],
|
||||
{"website.homepage": (5, "<t><div/></t>")},
|
||||
{10: "<t><div/><span/></t>"},
|
||||
)
|
||||
self.assertEqual(n, 1)
|
||||
self.assertTrue(row["differs"])
|
||||
self.assertTrue(row["comparable"])
|
||||
self.assertEqual(row["twin_id"], 5)
|
||||
self.assertEqual(
|
||||
row["diff_stats"]["added"] + row["diff_stats"]["changed"], 1
|
||||
)
|
||||
|
||||
def test_a_copy_identical_to_its_twin(self):
|
||||
# 35 des 62 copies d'une vraie base sont dans ce cas : elles ne
|
||||
# portent aucune personnalisation, et le dire change la décision.
|
||||
row = self.finding()
|
||||
A.attach_cow_twin_diff(
|
||||
[row], {"website.homepage": (5, "<t/>")}, {10: "<t/>"}
|
||||
)
|
||||
self.assertFalse(row["differs"])
|
||||
self.assertTrue(row["comparable"])
|
||||
|
||||
def test_indentation_alone_is_not_a_difference(self):
|
||||
row = self.finding()
|
||||
A.attach_cow_twin_diff(
|
||||
[row],
|
||||
{"website.homepage": (5, "<t><div/></t>")},
|
||||
{10: "<t>\n <div/>\n</t>"},
|
||||
)
|
||||
self.assertFalse(row["differs"])
|
||||
|
||||
def test_a_copy_without_a_twin_is_left_alone(self):
|
||||
# Une page faite dans l'éditeur web n'a rien à quoi se comparer.
|
||||
row = self.finding(has_module_twin=False)
|
||||
n = A.attach_cow_twin_diff([row], {}, {10: "<t/>"})
|
||||
self.assertEqual(n, 0)
|
||||
self.assertNotIn("arch_ref", row)
|
||||
self.assertNotIn("differs", row)
|
||||
|
||||
def test_a_view_that_is_not_a_copy_is_left_alone(self):
|
||||
row = view(id=11, key="sale.order_form", arch_fs="x.xml")
|
||||
row["category"], row["reason"] = A.classify(row)
|
||||
n = A.attach_cow_twin_diff(
|
||||
[row], {"sale.order_form": (1, "<form/>")}, {11: "<form/>"}
|
||||
)
|
||||
self.assertEqual(n, 0)
|
||||
self.assertNotIn("differs", row)
|
||||
|
||||
def test_it_uses_the_same_field_names_as_the_module_comparison(self):
|
||||
# Le nom des champs EST le contrat : l'écran de navigation et le rendu
|
||||
# texte marchent alors sans savoir laquelle des deux comparaisons a
|
||||
# produit la donnée.
|
||||
row = self.finding()
|
||||
A.attach_cow_twin_diff(
|
||||
[row], {"website.homepage": (5, "<t/>")}, {10: "<t><i/></t>"}
|
||||
)
|
||||
for field in ("arch_ref", "arch_db_text", "differs", "comparable"):
|
||||
self.assertIn(field, row, field)
|
||||
|
||||
def test_a_missing_arch_is_not_a_false_verdict(self):
|
||||
# Sans l'arch de la copie, il n'y a pas eu de comparaison : ne rien
|
||||
# conclure vaut mieux que conclure « identique ».
|
||||
row = self.finding()
|
||||
n = A.attach_cow_twin_diff(
|
||||
[row], {"website.homepage": (5, "<t/>")}, {}
|
||||
)
|
||||
self.assertEqual(n, 0)
|
||||
self.assertNotIn("differs", row)
|
||||
|
|
|
|||
1243
test/test_check_filestore.py
Normal file
1243
test/test_check_filestore.py
Normal file
File diff suppressed because it is too large
Load diff
875
test/test_check_module_package.py
Normal file
875
test/test_check_module_package.py
Normal file
|
|
@ -0,0 +1,875 @@
|
|||
#!/usr/bin/env python3
|
||||
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
||||
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
||||
|
||||
"""Ce que l'outil doit garantir avant qu'on lui fasse confiance.
|
||||
|
||||
Deux propriétés valent tous les autres tests. La première : un module
|
||||
absent n'est pas UNE catégorie — dire « manquant » là où il fallait dire
|
||||
« absent du chemin des addons » envoie installer ce qui ne peut pas
|
||||
l'être. La seconde : l'outil vise des bases 12 à 18 dans la même session,
|
||||
et `shortdesc` y est tantôt varchar tantôt jsonb ; se tromper fait
|
||||
échouer la requête entière et l'outil déclare la base illisible alors
|
||||
qu'elle se porte bien.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
|
||||
sys.path.append(
|
||||
os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
|
||||
)
|
||||
|
||||
from script.analyse import check_module_package as modules # noqa: E402
|
||||
from script.todo import todo_i18n # noqa: E402
|
||||
|
||||
PACKAGES = {
|
||||
"odoo18.0_base": {
|
||||
"base": "",
|
||||
"image_list": [{"module": ["queue_job", "web_dark_mode"]}],
|
||||
},
|
||||
"odoo18.0_website": {
|
||||
"base": "odoo18.0_base",
|
||||
"image_list": [{"module": ["website_extra", "queue_job"]}],
|
||||
},
|
||||
"boucle_a": {"base": "boucle_b", "image_list": [{"module": ["a"]}]},
|
||||
"boucle_b": {"base": "boucle_a", "image_list": [{"module": ["b"]}]},
|
||||
}
|
||||
|
||||
|
||||
class TestThePackageFile(unittest.TestCase):
|
||||
def test_the_path_is_absolute(self):
|
||||
# Un chemin relatif ne marchait que depuis la racine du dépôt :
|
||||
# lancé du menu TODO ou d'ailleurs, l'outil disait « fichier de
|
||||
# packages illisible » et envoyait chercher au mauvais endroit.
|
||||
self.assertTrue(os.path.isabs(modules.PACKAGE_FILE))
|
||||
|
||||
def test_the_real_file_parses(self):
|
||||
reel = modules.read_packages()
|
||||
self.assertIn("odoo18.0_base", reel)
|
||||
|
||||
def test_an_unreadable_file_is_empty_not_a_crash(self):
|
||||
self.assertEqual(modules.read_packages("/nowhere/at/all.json"), {})
|
||||
|
||||
def test_a_file_that_is_not_a_dict_is_refused(self):
|
||||
chemin = os.path.join(os.path.dirname(__file__), "..", "README.md")
|
||||
self.assertEqual(modules.read_packages(chemin), {})
|
||||
|
||||
|
||||
class TestTheChain(unittest.TestCase):
|
||||
def test_it_runs_root_first(self):
|
||||
self.assertEqual(
|
||||
modules.package_chain("odoo18.0_website", PACKAGES),
|
||||
["odoo18.0_base", "odoo18.0_website"],
|
||||
)
|
||||
|
||||
def test_a_root_is_alone(self):
|
||||
self.assertEqual(
|
||||
modules.package_chain("odoo18.0_base", PACKAGES), ["odoo18.0_base"]
|
||||
)
|
||||
|
||||
def test_an_unknown_name_gives_nothing(self):
|
||||
self.assertEqual(modules.package_chain("pas_moi", PACKAGES), [])
|
||||
|
||||
def test_a_cycle_stops_instead_of_spinning(self):
|
||||
# Sans garde, `base` circulaire ferait tourner l'outil sans fin.
|
||||
chaine = modules.package_chain("boucle_a", PACKAGES)
|
||||
self.assertEqual(sorted(chaine), ["boucle_a", "boucle_b"])
|
||||
|
||||
|
||||
class TestWhatAPackageSuggests(unittest.TestCase):
|
||||
def test_it_inherits_from_its_base(self):
|
||||
trouve = modules.package_modules("odoo18.0_website", PACKAGES)
|
||||
self.assertEqual(
|
||||
sorted(trouve), ["queue_job", "web_dark_mode", "website_extra"]
|
||||
)
|
||||
|
||||
def test_it_keeps_the_package_of_ORIGIN(self):
|
||||
# Savoir qu'un module vient de la base et non de l'extension dit
|
||||
# s'il est fondamental ou accessoire ; garder le dernier vu
|
||||
# attribuerait tout à la feuille.
|
||||
trouve = modules.package_modules("odoo18.0_website", PACKAGES)
|
||||
self.assertEqual(trouve["queue_job"], "odoo18.0_base")
|
||||
self.assertEqual(trouve["website_extra"], "odoo18.0_website")
|
||||
|
||||
def test_the_real_18_package_has_its_fifteen(self):
|
||||
reel = modules.read_packages()
|
||||
self.assertEqual(
|
||||
len(modules.package_modules("odoo18.0_base", reel)), 15
|
||||
)
|
||||
|
||||
|
||||
class TestTheVersion(unittest.TestCase):
|
||||
def test_it_keeps_two_components(self):
|
||||
self.assertEqual(modules.default_package("18.0"), "odoo18.0_base")
|
||||
|
||||
def test_no_version_gives_no_package(self):
|
||||
self.assertIsNone(modules.default_package(None))
|
||||
|
||||
|
||||
class TestReadingAColumn(unittest.TestCase):
|
||||
"""`shortdesc` est varchar jusqu'en 15, jsonb ensuite."""
|
||||
|
||||
def test_a_plain_column_is_read_plainly(self):
|
||||
sql = modules.as_text("shortdesc", {"shortdesc": "character varying"})
|
||||
self.assertNotIn("->>", sql)
|
||||
self.assertIn("shortdesc", sql)
|
||||
|
||||
def test_a_jsonb_column_is_extracted(self):
|
||||
sql = modules.as_text("shortdesc", {"shortdesc": "jsonb"})
|
||||
self.assertIn("->>", sql)
|
||||
self.assertIn("en_US", sql)
|
||||
|
||||
def test_an_unknown_column_is_treated_as_plain(self):
|
||||
self.assertNotIn("->>", modules.as_text("shortdesc", {}))
|
||||
|
||||
|
||||
class TestTheVerdicts(unittest.TestCase):
|
||||
CONNUS = {
|
||||
"installe": ("installed", "", False, "Odoo"),
|
||||
"a_maj": ("to upgrade", "", False, "Odoo"),
|
||||
"dispo": ("uninstalled", "", False, "OCA"),
|
||||
"casse": ("uninstallable", "", False, "OCA"),
|
||||
"en_cours": ("to install", "", False, "OCA"),
|
||||
}
|
||||
|
||||
def test_each_state_maps_to_its_verdict(self):
|
||||
for nom, attendu in (
|
||||
("installe", "installed"),
|
||||
("a_maj", "installed"),
|
||||
("dispo", "available"),
|
||||
("casse", "uninstallable"),
|
||||
("en_cours", "pending"),
|
||||
):
|
||||
self.assertEqual(
|
||||
modules.verdict_of(nom, self.CONNUS), attendu, nom
|
||||
)
|
||||
|
||||
def test_a_module_the_database_never_heard_of_is_unknown(self):
|
||||
# Le distinguer de « disponible » est tout l'intérêt : il n'y a
|
||||
# rien à installer tant que le chemin des addons est incomplet.
|
||||
self.assertEqual(
|
||||
modules.verdict_of("jamais_vu", self.CONNUS), "unknown"
|
||||
)
|
||||
|
||||
def test_an_unexpected_state_is_not_silently_installed(self):
|
||||
# Un état qu'Odoo ajouterait demain ne doit pas passer pour
|
||||
# installé : mieux vaut le signaler que de le taire.
|
||||
etrange = {"x": ("something_new", "", False, "")}
|
||||
self.assertNotEqual(modules.verdict_of("x", etrange), "installed")
|
||||
|
||||
|
||||
class TestBlockingDependencies(unittest.TestCase):
|
||||
CONNUS = {
|
||||
"moi": ("uninstalled", "", False, ""),
|
||||
"ok": ("installed", "", False, ""),
|
||||
"dispo": ("uninstalled", "", False, ""),
|
||||
"casse": ("uninstallable", "", False, ""),
|
||||
}
|
||||
|
||||
def test_only_what_truly_blocks_is_listed(self):
|
||||
depend = {"moi": ["ok", "dispo", "casse", "absent"]}
|
||||
self.assertEqual(
|
||||
modules.blocking_dependencies("moi", self.CONNUS, depend),
|
||||
["absent", "casse"],
|
||||
)
|
||||
|
||||
def test_no_dependency_is_no_problem(self):
|
||||
self.assertEqual(
|
||||
modules.blocking_dependencies("moi", self.CONNUS, {}), []
|
||||
)
|
||||
|
||||
|
||||
class FakeBase:
|
||||
"""Une base qui répond, sans PostgreSQL.
|
||||
|
||||
On aiguille sur le SQL plutôt que sur l'ordre des appels : un test qui
|
||||
compte les appels casse dès qu'on réordonne le code sans rien changer
|
||||
au comportement.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
connus,
|
||||
depend=None,
|
||||
jsonb=False,
|
||||
muette=False,
|
||||
sans_colonnes=False,
|
||||
sans_recensement=False,
|
||||
):
|
||||
self.connus = connus
|
||||
self.depend = depend or {}
|
||||
self.jsonb = jsonb
|
||||
self.muette = muette
|
||||
self.sans_colonnes = sans_colonnes
|
||||
self.sans_recensement = sans_recensement
|
||||
self.vues = []
|
||||
|
||||
def __call__(self, database, sql):
|
||||
self.vues.append(sql)
|
||||
if self.muette:
|
||||
return None
|
||||
if "information_schema.columns" in sql:
|
||||
if self.sans_colonnes:
|
||||
return []
|
||||
desc = "jsonb" if self.jsonb else "character varying"
|
||||
return [
|
||||
["name", "character varying"],
|
||||
["shortdesc", desc],
|
||||
["author", "character varying"],
|
||||
["state", "character varying"],
|
||||
]
|
||||
if "latest_version" in sql:
|
||||
return [["18.0.1.3"]]
|
||||
if "ir_module_module_dependency" in sql:
|
||||
return [
|
||||
[mod, dep] for mod, lst in self.depend.items() for dep in lst
|
||||
]
|
||||
if "FROM ir_module_module" in sql:
|
||||
if self.sans_recensement:
|
||||
return None
|
||||
return [
|
||||
[nom, etat, desc, "1" if app else "0", auteur]
|
||||
for nom, (etat, desc, app, auteur) in self.connus.items()
|
||||
]
|
||||
return []
|
||||
|
||||
|
||||
CONNUS = {
|
||||
"queue_job": ("uninstalled", "Job Queue", False, "OCA"),
|
||||
"web_dark_mode": ("installed", "Dark Mode", False, "OCA"),
|
||||
"base": ("installed", "Base", False, "Odoo S.A."),
|
||||
"casse": ("uninstallable", "Cassé", False, "OCA"),
|
||||
}
|
||||
|
||||
|
||||
class TestTheAudit(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.vrai = modules.run_psql
|
||||
|
||||
def tearDown(self):
|
||||
modules.run_psql = self.vrai
|
||||
|
||||
def audite(self, connus=None, depend=None, jsonb=False, package=None):
|
||||
modules.run_psql = FakeBase(
|
||||
CONNUS if connus is None else connus, depend, jsonb
|
||||
)
|
||||
return modules.audit("db", package=package, packages=PACKAGES)
|
||||
|
||||
def test_it_finds_the_package_from_the_database_version(self):
|
||||
rapport = self.audite()
|
||||
self.assertEqual(rapport["version"], "18.0")
|
||||
self.assertEqual(rapport["package"], "odoo18.0_base")
|
||||
self.assertTrue(rapport["package_known"])
|
||||
|
||||
def test_it_classifies_each_suggested_module(self):
|
||||
rapport = self.audite()
|
||||
verdicts = {
|
||||
ligne["module"]: ligne["verdict"] for ligne in rapport["lines"]
|
||||
}
|
||||
self.assertEqual(
|
||||
verdicts, {"queue_job": "available", "web_dark_mode": "installed"}
|
||||
)
|
||||
|
||||
def test_a_module_absent_from_the_database_is_unknown(self):
|
||||
rapport = self.audite(
|
||||
connus={"base": ("installed", "", False, "Odoo")}
|
||||
)
|
||||
verdicts = {
|
||||
ligne["module"]: ligne["verdict"] for ligne in rapport["lines"]
|
||||
}
|
||||
self.assertEqual(verdicts["queue_job"], "unknown")
|
||||
|
||||
def test_it_reports_what_installing_would_still_need(self):
|
||||
rapport = self.audite(
|
||||
depend={"queue_job": ["casse", "base", "fantome"]}
|
||||
)
|
||||
ligne = [x for x in rapport["lines"] if x["module"] == "queue_job"][0]
|
||||
self.assertEqual(ligne["needs"], ["casse", "fantome"])
|
||||
|
||||
def test_a_jsonb_database_is_read_not_refused(self):
|
||||
# Sur une base 16+, `shortdesc` est jsonb. Une requête écrite pour
|
||||
# du varchar y échoue ENTIÈREMENT et l'outil déclarerait la base
|
||||
# illisible.
|
||||
rapport = self.audite(jsonb=True)
|
||||
self.assertFalse(rapport.get("unavailable"))
|
||||
recensement = [
|
||||
s
|
||||
for s in modules.run_psql.vues
|
||||
if "FROM ir_module_module" in s
|
||||
and "dependency" not in s
|
||||
and "latest_version" not in s
|
||||
]
|
||||
self.assertTrue(any("->>" in s for s in recensement), recensement)
|
||||
|
||||
def test_a_silent_database_is_unavailable_not_empty(self):
|
||||
# Rendre un rapport vide ferait croire à une base sans modules.
|
||||
modules.run_psql = FakeBase({}, muette=True)
|
||||
self.assertTrue(modules.audit("db", packages=PACKAGES)["unavailable"])
|
||||
|
||||
def test_a_database_without_the_module_table_is_unavailable(self):
|
||||
# Pas de table `ir_module_module` : ce n'est pas une base Odoo.
|
||||
modules.run_psql = FakeBase(CONNUS, sans_colonnes=True)
|
||||
self.assertTrue(modules.audit("db", packages=PACKAGES)["unavailable"])
|
||||
|
||||
def test_a_census_query_that_fails_is_unavailable(self):
|
||||
# Les colonnes répondent, le recensement non — l'autre garde. Les
|
||||
# tester ensemble laissait chacune masquer la panne de l'autre.
|
||||
modules.run_psql = FakeBase(CONNUS, sans_recensement=True)
|
||||
self.assertTrue(modules.audit("db", packages=PACKAGES)["unavailable"])
|
||||
|
||||
def test_the_census_counts_every_module_not_only_the_suggested(self):
|
||||
rapport = self.audite()
|
||||
self.assertEqual(rapport["total"], len(CONNUS))
|
||||
self.assertEqual(rapport["by_state"]["installed"], 2)
|
||||
|
||||
def test_extras_are_what_is_installed_beyond_the_package(self):
|
||||
rapport = self.audite()
|
||||
self.assertIn("base", rapport["extra"])
|
||||
self.assertNotIn("web_dark_mode", rapport["extra"])
|
||||
|
||||
def test_an_unknown_package_is_flagged_not_silently_empty(self):
|
||||
rapport = self.audite(package="jamais_defini")
|
||||
self.assertFalse(rapport["package_known"])
|
||||
self.assertEqual(rapport["lines"], [])
|
||||
|
||||
|
||||
class TestTheOrderOfTheReport(unittest.TestCase):
|
||||
def ligne(self, module, verdict):
|
||||
return {
|
||||
"module": module,
|
||||
"verdict": verdict,
|
||||
"from": "p",
|
||||
"state": "",
|
||||
"shortdesc": "",
|
||||
"needs": [],
|
||||
}
|
||||
|
||||
def test_the_worst_comes_first(self):
|
||||
# « unknown » demande de réparer le chemin des addons avant tout
|
||||
# le reste : le lire en dernier ferait installer dans le vide.
|
||||
rapport = {
|
||||
"lines": [
|
||||
self.ligne("aaa", "available"),
|
||||
self.ligne("zzz", "unknown"),
|
||||
self.ligne("mmm", "uninstallable"),
|
||||
self.ligne("bbb", "installed"),
|
||||
]
|
||||
}
|
||||
self.assertEqual(
|
||||
[x["module"] for x in modules.missing(rapport)],
|
||||
["zzz", "mmm", "aaa"],
|
||||
)
|
||||
|
||||
def test_installed_modules_never_appear(self):
|
||||
rapport = {"lines": [self.ligne("z", "installed")]}
|
||||
self.assertEqual(modules.missing(rapport), [])
|
||||
|
||||
def test_every_verdict_has_a_rank_and_an_icon_and_a_wording(self):
|
||||
# Un verdict sans rang ferait planter le tri ; sans libellé, le
|
||||
# rapport nommerait l'état sans nommer le geste.
|
||||
for verdict in modules.VERDICTS:
|
||||
self.assertIn(verdict, modules.ICONE)
|
||||
self.assertIn(verdict, modules.EXPLICATION)
|
||||
for verdict in modules.ETAT_VERS_VERDICT.values():
|
||||
self.assertIn(verdict, modules.VERDICTS)
|
||||
|
||||
|
||||
class TestTheRendering(unittest.TestCase):
|
||||
def rapport(self, **extra):
|
||||
base = {
|
||||
"database": "db",
|
||||
"version": "18.0",
|
||||
"package": "odoo18.0_base",
|
||||
"package_known": True,
|
||||
"chain": ["odoo18.0_base"],
|
||||
"lines": [],
|
||||
"by_state": {"installed": 2},
|
||||
"total": 2,
|
||||
"installed": ["base"],
|
||||
"extra": [],
|
||||
"authors": [("Odoo S.A.", 2)],
|
||||
}
|
||||
base.update(extra)
|
||||
return base
|
||||
|
||||
def test_a_clean_database_says_so(self):
|
||||
texte = "\n".join(modules.render(self.rapport()))
|
||||
self.assertIn(
|
||||
todo_i18n.t("Every suggested module is installed."), texte
|
||||
)
|
||||
|
||||
def test_an_unknown_package_never_claims_success(self):
|
||||
# Comparé à rien, TOUT semble installé : le dire serait un
|
||||
# mensonge tranquille, le pire des rapports.
|
||||
texte = "\n".join(
|
||||
modules.render(self.rapport(package_known=False, package="?"))
|
||||
)
|
||||
self.assertNotIn(
|
||||
todo_i18n.t("Every suggested module is installed."), texte
|
||||
)
|
||||
self.assertIn(
|
||||
todo_i18n.t("No default package known for this version"), texte
|
||||
)
|
||||
|
||||
def test_a_missing_module_names_the_action(self):
|
||||
ligne = {
|
||||
"module": "queue_job",
|
||||
"verdict": "unknown",
|
||||
"from": "p",
|
||||
"state": "",
|
||||
"shortdesc": "Job Queue",
|
||||
"needs": [],
|
||||
}
|
||||
texte = "\n".join(modules.render(self.rapport(lines=[ligne])))
|
||||
self.assertIn("queue_job", texte)
|
||||
self.assertIn(
|
||||
todo_i18n.t("absent from the addons path — sync the repo first"),
|
||||
texte,
|
||||
)
|
||||
|
||||
def test_an_unreadable_database_renders_without_crashing(self):
|
||||
texte = "\n".join(
|
||||
modules.render({"unavailable": True, "database": "x"})
|
||||
)
|
||||
self.assertIn("x", texte)
|
||||
|
||||
def test_the_limit_caps_the_list_and_says_how_many_were_hidden(self):
|
||||
lignes = [
|
||||
{
|
||||
"module": f"m{i}",
|
||||
"verdict": "available",
|
||||
"from": "p",
|
||||
"state": "",
|
||||
"shortdesc": "",
|
||||
"needs": [],
|
||||
}
|
||||
for i in range(10)
|
||||
]
|
||||
texte = "\n".join(modules.render(self.rapport(lines=lignes), limit=3))
|
||||
self.assertIn("m0", texte)
|
||||
self.assertNotIn("m9", texte)
|
||||
self.assertIn(f"7 {todo_i18n.t('more')}", texte)
|
||||
|
||||
|
||||
class TestTheCommandLine(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.vrai = modules.run_psql
|
||||
|
||||
def tearDown(self):
|
||||
modules.run_psql = self.vrai
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
import json
|
||||
import tempfile
|
||||
|
||||
cls.dossier = tempfile.TemporaryDirectory()
|
||||
cls.fichier = os.path.join(cls.dossier.name, "packages.json")
|
||||
with io.open(cls.fichier, "w", encoding="utf-8") as handle:
|
||||
json.dump(PACKAGES, handle)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.dossier.cleanup()
|
||||
|
||||
def lance(self, argv, connus=CONNUS, muette=False):
|
||||
modules.run_psql = FakeBase(connus, muette=muette)
|
||||
argv = list(argv) + ["--file", self.fichier]
|
||||
tampon = io.StringIO()
|
||||
with redirect_stdout(tampon):
|
||||
code = modules.main(argv)
|
||||
return code, tampon.getvalue()
|
||||
|
||||
def test_nothing_missing_exits_zero(self):
|
||||
code, _ = self.lance(
|
||||
["-d", "db", "-p", "odoo18.0_base"],
|
||||
connus={
|
||||
"web_dark_mode": ("installed", "", False, ""),
|
||||
"queue_job": ("installed", "", False, ""),
|
||||
},
|
||||
)
|
||||
self.assertEqual(code, 0)
|
||||
|
||||
def test_a_finding_exits_one(self):
|
||||
code, sortie = self.lance(["-d", "db"])
|
||||
self.assertEqual(code, 1)
|
||||
self.assertIn("queue_job", sortie)
|
||||
|
||||
def test_an_unreadable_database_exits_two(self):
|
||||
# 2 dit « l'outil a échoué », pas « rien trouvé » : un script qui
|
||||
# confond les deux conclurait que tout va bien.
|
||||
code, _ = self.lance(["-d", "db"], muette=True)
|
||||
self.assertEqual(code, 2)
|
||||
|
||||
def test_listing_packages_needs_no_database(self):
|
||||
code, sortie = self.lance(["--list-packages"])
|
||||
self.assertEqual(code, 0)
|
||||
self.assertIn("odoo18.0_base", sortie)
|
||||
|
||||
def test_json_output_is_parseable(self):
|
||||
import json
|
||||
|
||||
code, sortie = self.lance(["-d", "db", "--json"])
|
||||
self.assertEqual(code, 1)
|
||||
self.assertEqual(json.loads(sortie)["package"], "odoo18.0_base")
|
||||
|
||||
|
||||
class TestTheWiring(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 f:
|
||||
return f.read()
|
||||
|
||||
def test_every_translation_key_exists(self):
|
||||
# Une clé absente s'affiche en anglais au milieu du français, et
|
||||
# rien ne le signale à l'exécution.
|
||||
src = self.source("script/analyse/check_module_package.py")
|
||||
arbre = ast.parse(src)
|
||||
cles = set()
|
||||
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)
|
||||
cles.update(modules.EXPLICATION.values())
|
||||
manquantes = [c for c in cles if c not in todo_i18n.TRANSLATIONS]
|
||||
self.assertEqual(manquantes, [])
|
||||
|
||||
def test_the_menu_offers_the_entry_and_dispatches_it(self):
|
||||
src = self.source("script/todo/todo.py")
|
||||
self.assertIn("Modules missing from the default package", src)
|
||||
self.assertIn("self.execute_analyse_module_package()", src)
|
||||
self.assertIn("def execute_analyse_module_package", src)
|
||||
|
||||
def test_the_menu_has_as_many_entries_as_branches(self):
|
||||
# Ajouter une entrée sans son aiguillage donne « Command not
|
||||
# found » sur un choix que le menu vient d'afficher.
|
||||
src = self.source("script/todo/todo.py")
|
||||
debut = src.index("def prompt_execute_analyse")
|
||||
fin = src.index("def execute_analyse_module_package")
|
||||
self.assertLess(debut, fin)
|
||||
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 TestWhatCanBeInstalled(unittest.TestCase):
|
||||
def rapport(self, *couples):
|
||||
return {
|
||||
"lines": [
|
||||
{
|
||||
"module": nom,
|
||||
"verdict": v,
|
||||
"from": "p",
|
||||
"state": "",
|
||||
"shortdesc": "",
|
||||
"needs": [],
|
||||
}
|
||||
for nom, v in couples
|
||||
]
|
||||
}
|
||||
|
||||
def test_only_available_modules_are_offered(self):
|
||||
# Proposer un « unknown » ferait une liste plus longue et un
|
||||
# échec : il n'est pas dans le chemin des addons.
|
||||
r = self.rapport(
|
||||
("a", "available"),
|
||||
("b", "unknown"),
|
||||
("c", "uninstallable"),
|
||||
("d", "pending"),
|
||||
("e", "installed"),
|
||||
)
|
||||
self.assertEqual(modules.installable(r), ["a"])
|
||||
|
||||
def test_nothing_available_offers_nothing(self):
|
||||
self.assertEqual(
|
||||
modules.installable(self.rapport(("b", "unknown"))), []
|
||||
)
|
||||
|
||||
|
||||
class TestTheSelection(unittest.TestCase):
|
||||
CANDIDATS = ["alpha", "beta", "gamma"]
|
||||
|
||||
def choisit(self, reponse):
|
||||
return modules.parse_selection(reponse, self.CANDIDATS)
|
||||
|
||||
def test_nothing_typed_selects_nothing(self):
|
||||
self.assertEqual(self.choisit(""), ([], []))
|
||||
self.assertEqual(self.choisit(" "), ([], []))
|
||||
|
||||
def test_a_takes_every_one(self):
|
||||
for mot in ("a", "A", "all", " a "):
|
||||
self.assertEqual(self.choisit(mot), (self.CANDIDATS, []), mot)
|
||||
|
||||
def test_numbers_are_one_based_and_space_separated(self):
|
||||
self.assertEqual(self.choisit("1 3"), (["alpha", "gamma"], []))
|
||||
|
||||
def test_the_typed_order_is_kept(self):
|
||||
self.assertEqual(self.choisit("3 1"), (["gamma", "alpha"], []))
|
||||
|
||||
def test_a_repeat_is_installed_once(self):
|
||||
self.assertEqual(self.choisit("2 2"), (["beta"], []))
|
||||
|
||||
def test_commas_work_too(self):
|
||||
# Les listes que l'outil affiche ailleurs sont en virgules ;
|
||||
# refuser « 1,3 » n'aurait protégé de rien.
|
||||
self.assertEqual(self.choisit("1,3"), (["alpha", "gamma"], []))
|
||||
|
||||
def test_an_out_of_range_number_is_REPORTED_not_dropped(self):
|
||||
# En demander deux et en recevoir un sans un mot ferait croire
|
||||
# l'installation complète. C'est la pire issue possible.
|
||||
self.assertEqual(self.choisit("1 9"), (["alpha"], ["9"]))
|
||||
|
||||
def test_zero_is_out_of_range(self):
|
||||
self.assertEqual(self.choisit("0"), ([], ["0"]))
|
||||
|
||||
def test_a_word_is_reported_not_ignored(self):
|
||||
self.assertEqual(self.choisit("1 pouet"), (["alpha"], ["pouet"]))
|
||||
|
||||
def test_only_rubbish_selects_nothing_and_says_so(self):
|
||||
self.assertEqual(self.choisit("x y"), ([], ["x", "y"]))
|
||||
|
||||
|
||||
class TestThePermissions(unittest.TestCase):
|
||||
"""Shebang et exécutable vont ensemble, dans les deux sens.
|
||||
|
||||
Un outil livré sans le bit d'exécution ne se lance qu'en le préfixant
|
||||
de `python3`, et rien ne le signale avant l'essai. Un module de
|
||||
bibliothèque marqué exécutable invite à le lancer alors qu'il n'a pas
|
||||
de `main` — l'inverse est tout aussi trompeur.
|
||||
"""
|
||||
|
||||
DOSSIER = os.path.normpath(
|
||||
os.path.join(os.path.dirname(__file__), "..", "script", "analyse")
|
||||
)
|
||||
|
||||
def test_shebang_and_executable_bit_agree(self):
|
||||
import glob
|
||||
import stat
|
||||
|
||||
for chemin in sorted(glob.glob(os.path.join(self.DOSSIER, "*.py"))):
|
||||
with io.open(chemin, encoding="utf-8") as handle:
|
||||
shebang = handle.readline().startswith("#!")
|
||||
executable = bool(
|
||||
stat.S_IMODE(os.stat(chemin).st_mode) & stat.S_IXUSR
|
||||
)
|
||||
self.assertEqual(shebang, executable, os.path.basename(chemin))
|
||||
|
||||
|
||||
class TestTheInstallOffer(unittest.TestCase):
|
||||
"""L'invite qui ÉCRIT. On vérifie la commande, pas les appels.
|
||||
|
||||
Un test qui se contente de constater qu'une fonction a été appelée
|
||||
laisse passer une commande mal formée ou lancée quand il ne fallait
|
||||
pas. Ici on retient la ligne de commande exacte, et surtout on exige
|
||||
qu'il n'en parte AUCUNE sur les chemins de refus.
|
||||
"""
|
||||
|
||||
RAPPORT = {
|
||||
"lines": [
|
||||
{
|
||||
"module": "queue_job",
|
||||
"verdict": "available",
|
||||
"from": "p",
|
||||
"state": "",
|
||||
"shortdesc": "",
|
||||
"needs": [],
|
||||
},
|
||||
{
|
||||
"module": "web_dark_mode",
|
||||
"verdict": "available",
|
||||
"from": "p",
|
||||
"state": "",
|
||||
"shortdesc": "",
|
||||
"needs": [],
|
||||
},
|
||||
{
|
||||
"module": "absent",
|
||||
"verdict": "unknown",
|
||||
"from": "p",
|
||||
"state": "",
|
||||
"shortdesc": "",
|
||||
"needs": [],
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
def setUp(self):
|
||||
from script.odoo.migration import database_cleanup
|
||||
from script.todo import auto_ask
|
||||
from script.todo import todo as todo_module
|
||||
|
||||
self.cleanup = database_cleanup
|
||||
self.auto_ask = auto_ask
|
||||
self.vraie_garde = database_cleanup.require_matching_version
|
||||
self.vrai_ask = auto_ask.ask
|
||||
database_cleanup.require_matching_version = lambda base: None
|
||||
|
||||
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
|
||||
)
|
||||
},
|
||||
)()
|
||||
|
||||
def tearDown(self):
|
||||
self.cleanup.require_matching_version = self.vraie_garde
|
||||
self.auto_ask.ask = self.vrai_ask
|
||||
|
||||
def joue(self, *reponses):
|
||||
"""Rendre TOUT ce que l'utilisateur voit : sortie ET invites.
|
||||
|
||||
Le texte d'une question ne passe pas par stdout — il est l'argument
|
||||
de `ask`, qu'un vrai `input()` affiche. Ne regarder que stdout
|
||||
laisserait une invite muette passer pour correcte.
|
||||
"""
|
||||
file = list(reponses)
|
||||
self.demandes = []
|
||||
|
||||
def faux_ask(prompt, default="", seconds=None):
|
||||
# Le VRAI `ask` rend le défaut quand la réponse est vide. Un
|
||||
# faux qui rend la chaîne vide telle quelle ne teste pas le
|
||||
# défaut du tout : basculer celui de « n » à « y » passait
|
||||
# alors inaperçu, et Entrée aurait installé.
|
||||
self.demandes.append(prompt)
|
||||
reponse = file.pop(0) if file else ""
|
||||
return reponse or default
|
||||
|
||||
self.auto_ask.ask = faux_ask
|
||||
tampon = io.StringIO()
|
||||
with redirect_stdout(tampon):
|
||||
self.obj._analyse_offer_install("ma_base", self.RAPPORT)
|
||||
return tampon.getvalue() + "\n".join(self.demandes)
|
||||
|
||||
def test_saying_no_runs_nothing(self):
|
||||
self.joue("n")
|
||||
self.assertEqual(self.lancees, [])
|
||||
|
||||
def test_pressing_enter_runs_nothing(self):
|
||||
# Le défaut d'une action qui écrit doit être de ne rien faire.
|
||||
# On fournit de quoi aller AU BOUT si le garde-fou cédait : sans
|
||||
# cela, la suite s'arrêtait faute de réponses et le test passait
|
||||
# même avec un défaut à « y ». Basculer le défaut installerait.
|
||||
self.joue("", "a", "y")
|
||||
self.assertEqual(self.lancees, [])
|
||||
|
||||
def test_pressing_enter_at_the_final_confirmation_runs_nothing(self):
|
||||
# Même piège sur le second garde-fou : il faut que « tout est
|
||||
# prêt » soit vrai au moment où l'on appuie sur Entrée.
|
||||
self.joue("y", "a", "")
|
||||
self.assertEqual(self.lancees, [])
|
||||
|
||||
def test_choosing_one_installs_exactly_that_one(self):
|
||||
self.joue("y", "1", "y")
|
||||
self.assertEqual(
|
||||
self.lancees,
|
||||
["./script/addons/install_addons.sh ma_base queue_job"],
|
||||
)
|
||||
|
||||
def test_choosing_a_installs_every_available_one(self):
|
||||
self.joue("y", "a", "y")
|
||||
self.assertEqual(
|
||||
self.lancees,
|
||||
[
|
||||
"./script/addons/install_addons.sh ma_base"
|
||||
" queue_job,web_dark_mode"
|
||||
],
|
||||
)
|
||||
|
||||
def test_the_unknown_module_is_never_offered(self):
|
||||
sortie = self.joue("y", "a", "y")
|
||||
self.assertNotIn("absent", self.lancees[0])
|
||||
self.assertIn("[1]", sortie)
|
||||
self.assertIn("[2]", sortie)
|
||||
self.assertNotIn("[3]", sortie)
|
||||
|
||||
def test_refusing_the_final_confirmation_runs_nothing(self):
|
||||
# Deuxième filet : on a choisi, on relit, on renonce.
|
||||
self.joue("y", "a", "n")
|
||||
self.assertEqual(self.lancees, [])
|
||||
|
||||
def test_selecting_nothing_runs_nothing(self):
|
||||
# On confirme APRÈS n'avoir rien choisi : sans le garde, la
|
||||
# commande partirait avec une liste de modules vide.
|
||||
self.joue("y", "", "y")
|
||||
self.assertEqual(self.lancees, [])
|
||||
|
||||
def test_selecting_only_rubbish_runs_nothing(self):
|
||||
# Même chemin, mais l'utilisateur a bien tapé quelque chose : ce
|
||||
# qu'il a tapé ne désigne aucun module.
|
||||
sortie = self.joue("y", "pouet 99", "y")
|
||||
self.assertEqual(self.lancees, [])
|
||||
self.assertIn("pouet", sortie)
|
||||
|
||||
def test_a_bad_token_is_shown_and_the_rest_still_installs(self):
|
||||
sortie = self.joue("y", "1 99", "y")
|
||||
self.assertIn("99", sortie)
|
||||
self.assertEqual(
|
||||
self.lancees,
|
||||
["./script/addons/install_addons.sh ma_base queue_job"],
|
||||
)
|
||||
|
||||
def test_a_version_mismatch_refuses_before_asking_anything(self):
|
||||
# Un Odoo 18 lancé sur une base 12 la RÉÉCRIT avant d'échouer :
|
||||
# ce refus doit précéder la moindre question.
|
||||
self.cleanup.require_matching_version = lambda base: "18.0 vs 12.0"
|
||||
demandes = []
|
||||
self.auto_ask.ask = lambda p, default="", seconds=None: (
|
||||
demandes.append(p) or "y"
|
||||
)
|
||||
tampon = io.StringIO()
|
||||
with redirect_stdout(tampon):
|
||||
self.obj._analyse_offer_install("ma_base", self.RAPPORT)
|
||||
self.assertEqual(self.lancees, [])
|
||||
self.assertEqual(demandes, [])
|
||||
self.assertIn("18.0 vs 12.0", tampon.getvalue())
|
||||
|
||||
def test_nothing_installable_asks_nothing_at_all(self):
|
||||
demandes = []
|
||||
self.auto_ask.ask = lambda p, default="", seconds=None: (
|
||||
demandes.append(p) or "y"
|
||||
)
|
||||
rien = {
|
||||
"lines": [
|
||||
{
|
||||
"module": "x",
|
||||
"verdict": "unknown",
|
||||
"from": "p",
|
||||
"state": "",
|
||||
"shortdesc": "",
|
||||
"needs": [],
|
||||
}
|
||||
]
|
||||
}
|
||||
with redirect_stdout(io.StringIO()):
|
||||
self.obj._analyse_offer_install("ma_base", rien)
|
||||
self.assertEqual(demandes, [])
|
||||
self.assertEqual(self.lancees, [])
|
||||
|
||||
def test_it_says_how_many_it_could_not_offer(self):
|
||||
# Le rapport vient d'en annoncer trois, la liste en montre deux :
|
||||
# sans un mot, on croirait à un bogue.
|
||||
sortie = self.joue("n")
|
||||
self.assertIn(todo_i18n.t("need repair first"), sortie)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
174
test/test_master_password_retry.py
Normal file
174
test/test_master_password_retry.py
Normal file
|
|
@ -0,0 +1,174 @@
|
|||
#!/usr/bin/env python3
|
||||
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
||||
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
||||
|
||||
"""Une faute de frappe ne doit pas coûter une migration.
|
||||
|
||||
Le mot de passe maître était demandé UNE fois. Faux ? Odoo lève
|
||||
`AccessDenied`, `check_output` lève `CalledProcessError`, rien ne
|
||||
l'attrape, et la migration meurt sur une trace. Après une heure de
|
||||
paliers, c'est cher payé pour une lettre.
|
||||
|
||||
La propriété qui porte tout : on ne redemande QUE sur un refus de mot de
|
||||
passe. Reposer la question sur n'importe quel échec cacherait la vraie
|
||||
panne derrière dix invites, et l'on chercherait un mot de passe alors
|
||||
que la base est cassée.
|
||||
"""
|
||||
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from contextlib import redirect_stderr, redirect_stdout
|
||||
|
||||
sys.path.append(
|
||||
os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
|
||||
)
|
||||
|
||||
from script.database import db_restore # noqa: E402
|
||||
|
||||
|
||||
class TestRecognisingTheRefusal(unittest.TestCase):
|
||||
def test_an_access_denied_is_a_refusal(self):
|
||||
self.assertTrue(
|
||||
db_restore.password_refused(
|
||||
"Traceback...\nodoo.exceptions.AccessDenied: Access Denied"
|
||||
)
|
||||
)
|
||||
|
||||
def test_the_class_is_what_we_match_not_the_message(self):
|
||||
# Le message est traduit : « Accès refusé » en français. La
|
||||
# CLASSE, elle, ne bouge pas.
|
||||
self.assertTrue(
|
||||
db_restore.password_refused("odoo.exceptions.AccessDenied")
|
||||
)
|
||||
self.assertFalse(db_restore.password_refused("Accès refusé"))
|
||||
|
||||
def test_anything_else_is_NOT_a_refusal(self):
|
||||
# C'est la protection : dix invites de mot de passe devant une
|
||||
# base cassée, et l'on cherche du mauvais côté.
|
||||
for sortie in (
|
||||
"psycopg2.OperationalError: could not connect",
|
||||
"FileNotFoundError: ./odoo_bin.sh",
|
||||
"",
|
||||
None,
|
||||
):
|
||||
self.assertFalse(db_restore.password_refused(sortie), sortie)
|
||||
|
||||
|
||||
class TestTheRetryLoop(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.vrais = (
|
||||
db_restore.get_master_password,
|
||||
db_restore.probe_master_password,
|
||||
)
|
||||
self.demandes = 0
|
||||
self.sondes = []
|
||||
|
||||
def tearDown(self):
|
||||
(
|
||||
db_restore.get_master_password,
|
||||
db_restore.probe_master_password,
|
||||
) = self.vrais
|
||||
|
||||
def branche(self, mots, reponses):
|
||||
suite = iter(mots)
|
||||
rep = iter(reponses)
|
||||
|
||||
def demander():
|
||||
self.demandes += 1
|
||||
return next(suite, "")
|
||||
|
||||
def sonder(arg_base):
|
||||
self.sondes.append(arg_base)
|
||||
return next(rep, (False, "AccessDenied"))
|
||||
|
||||
db_restore.get_master_password = demander
|
||||
db_restore.probe_master_password = sonder
|
||||
|
||||
def lance(self, essais=10):
|
||||
tampon = io.StringIO()
|
||||
with redirect_stdout(tampon), redirect_stderr(tampon):
|
||||
return db_restore.ask_master_password("./odoo_bin.sh db", essais)
|
||||
|
||||
def test_a_good_password_is_returned_at_once(self):
|
||||
self.branche(["bon"], [(True, "db1\ndb2")])
|
||||
self.assertEqual(self.lance(), "bon")
|
||||
self.assertEqual(self.demandes, 1)
|
||||
|
||||
def test_a_typo_is_asked_again(self):
|
||||
self.branche(
|
||||
["faux", "bon"],
|
||||
[(False, "odoo.exceptions.AccessDenied"), (True, "db1")],
|
||||
)
|
||||
self.assertEqual(self.lance(), "bon")
|
||||
self.assertEqual(self.demandes, 2)
|
||||
|
||||
def test_it_stops_after_the_allowed_attempts(self):
|
||||
# Sans borne, une invite non lue tournerait toute la nuit.
|
||||
self.branche(["faux"] * 20, [(False, "AccessDenied")] * 20)
|
||||
self.assertIsNone(self.lance(essais=10))
|
||||
self.assertEqual(self.demandes, 10)
|
||||
|
||||
def test_an_empty_prompt_gives_up_immediately(self):
|
||||
# Ctrl-D ou Entrée : on ne veut pas neuf invites de plus.
|
||||
self.branche([""], [])
|
||||
self.assertIsNone(self.lance())
|
||||
self.assertEqual(self.demandes, 1)
|
||||
self.assertEqual(self.sondes, [])
|
||||
|
||||
def test_an_unrelated_failure_stops_instead_of_asking_again(self):
|
||||
# LA propriété. Une base injoignable n'est pas un mot de passe
|
||||
# faux, et redemander dix fois cacherait la vraie panne.
|
||||
self.branche(["bon"], [(False, "psycopg2.OperationalError: refused")])
|
||||
self.assertIsNone(self.lance())
|
||||
self.assertEqual(self.demandes, 1)
|
||||
|
||||
def test_the_unrelated_failure_is_shown(self):
|
||||
self.branche(["bon"], [(False, "psycopg2.OperationalError: refused")])
|
||||
with self.assertLogs(db_restore._logger, level="ERROR") as journal:
|
||||
db_restore.ask_master_password("./odoo_bin.sh db")
|
||||
self.assertIn("OperationalError", "\n".join(journal.output))
|
||||
|
||||
def test_the_probe_carries_the_password_and_touches_nothing(self):
|
||||
# `--list` ne modifie rien : valider ici évite d'échouer à
|
||||
# mi-parcours, une fois la base déjà supprimée.
|
||||
self.branche(["bon"], [(True, "db1")])
|
||||
self.lance()
|
||||
self.assertEqual(len(self.sondes), 1)
|
||||
self.assertIn("--master_password=bon", self.sondes[0])
|
||||
|
||||
def test_each_attempt_probes_with_ITS_password(self):
|
||||
self.branche(
|
||||
["un", "deux"],
|
||||
[(False, "AccessDenied"), (True, "db1")],
|
||||
)
|
||||
self.lance()
|
||||
self.assertIn("--master_password=un", self.sondes[0])
|
||||
self.assertIn("--master_password=deux", self.sondes[1])
|
||||
|
||||
|
||||
class TestTheWiring(unittest.TestCase):
|
||||
def source(self):
|
||||
with io.open(db_restore.__file__, encoding="utf-8") as handle:
|
||||
return handle.read()
|
||||
|
||||
def test_the_flow_uses_the_retrying_version(self):
|
||||
src = self.source()
|
||||
self.assertIn("master_password = ask_master_password(arg_base)", src)
|
||||
|
||||
def test_the_probe_uses_list_which_changes_nothing(self):
|
||||
src = self.source()
|
||||
debut = src.index("def probe_master_password")
|
||||
fin = src.index("def ask_master_password")
|
||||
bloc = src[debut:fin]
|
||||
self.assertIn("--list", bloc)
|
||||
for danger in ("--drop", "--restore", "--clone"):
|
||||
self.assertNotIn(danger, bloc)
|
||||
|
||||
def test_the_bound_is_ten(self):
|
||||
self.assertEqual(db_restore.MAX_ESSAIS_MOT_DE_PASSE, 10)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in a new issue