[ADD] analyse: anonymiser une copie, sans IA et sans rien casser
Des mots pris dans une liste, des nombres tirés entre 0 et 1000, écrits en SQL. Aucun modèle, aucun réseau — un test le vérifie sur les imports. Le difficile n'est pas de remplacer, c'est de savoir ce qu'on n'a PAS le droit de toucher. Mesuré sur une base 18 réelle : 505 champs `selection` sont stockés en varchar, 2693 many2one sont des entiers, 194 textes sont des jsonb par langue, 301 contraintes d'unicité attendent une collision. « Tous les champs string » n'existe pas ; on croise ir_model_fields, pg_attribute et pg_constraint, et aucune des trois ne suffit seule. Trois pièges ont été trouvés en LANÇANT l'outil, pas en le relisant : PostgreSQL refuse d'indexer un ARRAY[...] sans parenthèses, res_partner.credit_limit est un jsonb qu'Odoo appelle float, et crm_lead.probability porte un CHECK qui interdit 1000. Chaque fois l'écriture a échoué et la base est restée intacte : une seule transaction, tout ou rien. Preuve sur copie jetable : empreinte du schéma identique, 848 tables, 6495 contraintes, arch_db et xmlid intacts, lang et many2one inchangés — seules les colonnes visées ont changé. --- EN --- Words from a list, numbers drawn between 0 and 1000, written in SQL. No model, no network — a test checks that on the imports. The hard part is not replacing, it is knowing what must NOT be touched. Measured on a real 18 database: 505 `selection` fields are stored as varchar, 2693 many2one are integers, 194 texts are per-language jsonb, 301 unique constraints await a collision. "All string fields" does not exist; we cross ir_model_fields, pg_attribute and pg_constraint, and none of the three is enough alone. Three traps were found by RUNNING it, not by rereading it: PostgreSQL refuses to subscript a bare ARRAY[...], res_partner.credit_limit is a jsonb Odoo calls float, and crm_lead.probability has a CHECK forbidding 1000. Each time the write failed and the database stayed intact: one transaction, all or nothing. Proof on a throwaway copy: identical schema fingerprint, 848 tables, 6495 constraints, arch_db and xmlids intact, lang and many2one unchanged — only the targeted columns changed. Assisted-by: Claude Opus 5
This commit is contained in:
parent
db9472ef69
commit
67a59d0522
6 changed files with 1064 additions and 7 deletions
550
script/analyse/anonymize.py
Executable file
550
script/analyse/anonymize.py
Executable file
|
|
@ -0,0 +1,550 @@
|
|||
#!/usr/bin/env python3
|
||||
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
||||
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
||||
|
||||
"""Remplacer les données d'une copie par des données sans intérêt.
|
||||
|
||||
Aucune IA, aucun appel réseau : des mots pris dans une liste et des
|
||||
nombres tirés au hasard, écrits par des UPDATE SQL. Ce qui rend la chose
|
||||
délicate n'est pas le remplacement, c'est de savoir CE QU'ON N'A PAS LE
|
||||
DROIT DE TOUCHER.
|
||||
|
||||
Quatre pièges, tous mesurés sur une base réelle
|
||||
-----------------------------------------------
|
||||
1. « Tous les champs string » n'existe pas. 505 champs `selection` sont
|
||||
stockés en varchar : `res.partner.lang`, `sale.order.invoice_status`.
|
||||
Y écrire un mot au hasard casse l'ORM, pas la confidentialité. On ne
|
||||
se fie donc jamais au type PostgreSQL seul, mais au `ttype` que
|
||||
`ir_model_fields` déclare.
|
||||
2. 2693 champs `many2one` sont des ENTIERS. Les tirer au hasard
|
||||
mélangerait toutes les relations de la base. Les nombres qu'on touche
|
||||
sont ceux qu'Odoo appelle integer, float ou monetary — jamais une
|
||||
relation.
|
||||
3. 194 champs texte sont en `jsonb` depuis Odoo 17, un objet par langue.
|
||||
Écrire une chaîne par-dessus détruit la colonne ; on reconstruit
|
||||
l'objet, clé par clé. C'est le piège qui a déjà coûté un /contact
|
||||
réparé en anglais et resté cassé en français.
|
||||
4. 301 contraintes d'unicité. Deux lignes qui reçoivent le même mot font
|
||||
échouer tout l'UPDATE. Sur une colonne unique, l'identifiant est
|
||||
collé au mot.
|
||||
|
||||
Ce qu'on ne touche jamais
|
||||
-------------------------
|
||||
Les modèles `ir.*` — vues, champs, xmlid : la base ne s'ouvrirait plus.
|
||||
Les langues, devises et pays : ce ne sont pas des données personnelles,
|
||||
et les casser casse les adresses et les montants. Cette liste est un
|
||||
PLANCHER : aucune liste blanche ne la lève.
|
||||
|
||||
`res.users.login` et le mot de passe restent en place par défaut. On
|
||||
anonymise pour POUVOIR partager une copie utilisable ; personne ne
|
||||
pourrait plus s'y connecter. `--include-logins` pour l'autre choix.
|
||||
|
||||
Rien n'est écrit sans `--apply` ET `--confirm <nom de la base>`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
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
|
||||
|
||||
|
||||
from script.analyse import lib_analyse # noqa: E402
|
||||
|
||||
TYPES_TEXTE = ("char", "text", "html")
|
||||
TYPES_NOMBRE = ("integer", "float", "monetary")
|
||||
|
||||
# Le plancher : aucune liste blanche ne le lève.
|
||||
PREFIXES_INTERDITS = ("ir.",)
|
||||
MODELES_INTERDITS = frozenset(
|
||||
{
|
||||
"res.groups",
|
||||
"res.lang",
|
||||
"res.currency",
|
||||
"res.currency.rate",
|
||||
"res.country",
|
||||
"res.country.state",
|
||||
"res.country.group",
|
||||
"res.config.settings",
|
||||
"base.language.install",
|
||||
"decimal.precision",
|
||||
"uom.uom",
|
||||
"uom.category",
|
||||
}
|
||||
)
|
||||
CHAMPS_INTERDITS = frozenset(
|
||||
{
|
||||
"id",
|
||||
"create_uid",
|
||||
"write_uid",
|
||||
"create_date",
|
||||
"write_date",
|
||||
"sequence",
|
||||
"active",
|
||||
"display_name",
|
||||
"__last_update",
|
||||
"arch_db",
|
||||
"arch_fs",
|
||||
"key",
|
||||
"model",
|
||||
"res_model",
|
||||
"res_field",
|
||||
"state",
|
||||
"color",
|
||||
"company_id",
|
||||
}
|
||||
)
|
||||
CHAMPS_CONNEXION = frozenset({"login", "password"})
|
||||
|
||||
# Le point de départ du mode hybride : ce qui porte des données
|
||||
# personnelles dans une base Odoo ordinaire.
|
||||
MODELES_PAR_DEFAUT = (
|
||||
"res.partner",
|
||||
"res.users",
|
||||
"res.company",
|
||||
"res.bank",
|
||||
"res.partner.bank",
|
||||
"crm.lead",
|
||||
"hr.employee",
|
||||
"mail.message",
|
||||
"mail.tracking.value",
|
||||
"account.move",
|
||||
"sale.order",
|
||||
"purchase.order",
|
||||
"project.task",
|
||||
"calendar.event",
|
||||
"survey.user_input",
|
||||
)
|
||||
|
||||
MOTS_PAR_DEFAUT = (
|
||||
"alouette",
|
||||
"bruyere",
|
||||
"cascade",
|
||||
"dolmen",
|
||||
"erable",
|
||||
"fougere",
|
||||
"givre",
|
||||
"hameau",
|
||||
"iris",
|
||||
"jonquille",
|
||||
"lichen",
|
||||
"marais",
|
||||
"nenuphar",
|
||||
"orme",
|
||||
"pinson",
|
||||
"roseau",
|
||||
"sureau",
|
||||
"tourbe",
|
||||
"varech",
|
||||
"zephyr",
|
||||
)
|
||||
|
||||
MODES = ("whitelist", "blacklist", "hybrid")
|
||||
|
||||
|
||||
def modele_interdit(modele):
|
||||
"""Le plancher, en une question."""
|
||||
if any(modele.startswith(p) for p in PREFIXES_INTERDITS):
|
||||
return True
|
||||
return modele in MODELES_INTERDITS
|
||||
|
||||
|
||||
def choisir_modeles(tous, mode, whitelist=(), blacklist=(), defauts=None):
|
||||
"""Quels modèles anonymiser, selon le mode.
|
||||
|
||||
whitelist : ceux-là et rien d'autre.
|
||||
blacklist : tous SAUF ceux-là.
|
||||
hybrid : la liste par défaut, plus la blanche, moins la noire —
|
||||
c'est le mode utile en pratique : on part de ce qui porte
|
||||
des données personnelles et on ajuste aux marges.
|
||||
|
||||
Le plancher s'applique aux trois : un modèle `ir.*` ne passe par
|
||||
aucun chemin, même nommé explicitement.
|
||||
"""
|
||||
tous = set(tous)
|
||||
blanche, noire = set(whitelist), set(blacklist)
|
||||
if mode == "whitelist":
|
||||
choisis = blanche & tous
|
||||
elif mode == "blacklist":
|
||||
choisis = tous - noire
|
||||
elif mode == "hybrid":
|
||||
base = set(MODELES_PAR_DEFAUT if defauts is None else defauts)
|
||||
choisis = ((base | blanche) - noire) & tous
|
||||
else:
|
||||
raise ValueError(f"{t('Unknown mode:')} {mode}")
|
||||
return sorted(m for m in choisis if not modele_interdit(m))
|
||||
|
||||
|
||||
def champ_retenu(champ, inclure_connexion=False):
|
||||
"""Ce champ-là se remplace-t-il ?
|
||||
|
||||
`champ` : dict avec model, name, ttype, pg_type, unique.
|
||||
"""
|
||||
if champ["name"] in CHAMPS_INTERDITS:
|
||||
return False
|
||||
if champ["name"] in CHAMPS_CONNEXION and not inclure_connexion:
|
||||
return False
|
||||
if champ["name"].endswith("_id") or champ["name"].endswith("_ids"):
|
||||
# Une relation qui aurait échappé au filtre de ttype.
|
||||
return False
|
||||
if champ.get("checked"):
|
||||
# Une contrainte CHECK dit ce que la colonne a le droit de valoir.
|
||||
# Mesuré : crm_lead.probability doit rester entre 0 et 100, et un
|
||||
# tirage à 1000 fait échouer l'UPDATE — donc, transaction unique
|
||||
# oblige, TOUTE l'anonymisation. Lire l'expression du CHECK pour
|
||||
# tirer dedans serait deviner ; on s'abstient et on le dit.
|
||||
return False
|
||||
if champ["ttype"] in TYPES_NOMBRE and champ.get("pg_type") == "jsonb":
|
||||
# Mesuré sur res_partner.credit_limit : un `float` d'Odoo peut
|
||||
# vivre dans un jsonb par société. Y écrire un nombre nu ferait
|
||||
# échouer l'UPDATE — et donc, transaction unique oblige, TOUTE
|
||||
# l'anonymisation. On s'abstient plutôt que de deviner sa forme.
|
||||
return False
|
||||
return champ["ttype"] in TYPES_TEXTE + TYPES_NOMBRE
|
||||
|
||||
|
||||
def mots_pour(nom_champ, mots):
|
||||
"""La liste de mots à utiliser pour ce champ.
|
||||
|
||||
`mots` peut être une simple liste — la même partout — ou un
|
||||
dictionnaire par nom de champ avec une entrée `*` en repli. Le
|
||||
dictionnaire permet de garder des courriels qui ressemblent à des
|
||||
courriels, ce qu'une liste unique ne sait pas faire.
|
||||
"""
|
||||
if isinstance(mots, dict):
|
||||
choix = mots.get(nom_champ) or mots.get("*") or MOTS_PAR_DEFAUT
|
||||
else:
|
||||
choix = mots or MOTS_PAR_DEFAUT
|
||||
return tuple(str(m) for m in choix) or MOTS_PAR_DEFAUT
|
||||
|
||||
|
||||
def litteral(texte):
|
||||
"""Un littéral SQL. Le seul endroit où du texte entre dans la requête."""
|
||||
return "'" + str(texte).replace("'", "''") + "'"
|
||||
|
||||
|
||||
def expression_texte(champ, mots):
|
||||
"""Le SQL qui remplace un champ texte, en préservant les NULL.
|
||||
|
||||
Un NULL qui deviendrait un mot créerait de la donnée là où il n'y en
|
||||
avait pas : la copie mentirait dans l'autre sens.
|
||||
"""
|
||||
nom = champ["name"]
|
||||
liste = mots_pour(nom, mots)
|
||||
# Les parenthèses ne sont pas décoratives : PostgreSQL refuse
|
||||
# d'indexer un constructeur ARRAY[...] directement.
|
||||
tableau = "(ARRAY[" + ",".join(litteral(m) for m in liste) + "])"
|
||||
tirage = f"{tableau}[(id % {len(liste)}) + 1]"
|
||||
if champ.get("unique"):
|
||||
# Deux lignes qui reçoivent le même mot feraient échouer TOUT
|
||||
# l'UPDATE sur une colonne unique.
|
||||
tirage = f"{tirage} || '-' || id::text"
|
||||
if champ.get("pg_type") == "jsonb":
|
||||
# Un objet par langue depuis Odoo 17 : on le reconstruit clé à
|
||||
# clé. Écrire une chaîne par-dessus détruirait la colonne.
|
||||
return (
|
||||
f"CASE WHEN {nom} IS NULL THEN NULL ELSE"
|
||||
f" (SELECT jsonb_object_agg(kv.key, {tirage})"
|
||||
f" FROM jsonb_each_text({nom}) AS kv) END"
|
||||
)
|
||||
return f"CASE WHEN {nom} IS NULL THEN NULL ELSE {tirage} END"
|
||||
|
||||
|
||||
def expression_nombre(champ):
|
||||
"""Le SQL qui remplace un nombre : au hasard, entre 0 et 1000."""
|
||||
nom = champ["name"]
|
||||
if champ["ttype"] == "integer":
|
||||
tirage = "floor(random() * 1001)::integer"
|
||||
else:
|
||||
tirage = "round((random() * 1000)::numeric, 2)"
|
||||
return f"CASE WHEN {nom} IS NULL THEN NULL ELSE {tirage} END"
|
||||
|
||||
|
||||
def sql_pour_table(table, champs, mots):
|
||||
"""Un seul UPDATE par table : toutes ses colonnes d'un coup."""
|
||||
morceaux = []
|
||||
for champ in champs:
|
||||
if champ["ttype"] in TYPES_TEXTE:
|
||||
morceaux.append(
|
||||
f"{champ['name']} = {expression_texte(champ, mots)}"
|
||||
)
|
||||
else:
|
||||
morceaux.append(f"{champ['name']} = {expression_nombre(champ)}")
|
||||
if not morceaux:
|
||||
return None
|
||||
return f"UPDATE {table} SET " + ", ".join(morceaux) + ";"
|
||||
|
||||
|
||||
def table_de(modele):
|
||||
"""Le nom de table qu'Odoo donne à ce modèle."""
|
||||
return modele.replace(".", "_")
|
||||
|
||||
|
||||
SEP = "\x1f"
|
||||
|
||||
# On croise TROIS sources, et c'est la raison d'être de cette requête :
|
||||
# `ir_model_fields` dit ce qu'Odoo croit (le ttype, seul capable de
|
||||
# distinguer un `selection` d'un vrai texte), `pg_attribute` dit ce que
|
||||
# PostgreSQL a vraiment (jsonb ou varchar), et `pg_constraint` dit ce qui
|
||||
# doit rester unique. Aucune des trois ne suffit seule.
|
||||
REQUETE_CHAMPS = """
|
||||
SELECT f.model || '\x1f' || f.name || '\x1f' || f.ttype || '\x1f'
|
||||
|| a.atttypid::regtype::text || '\x1f'
|
||||
|| CASE WHEN EXISTS (
|
||||
SELECT 1 FROM pg_constraint k
|
||||
WHERE k.conrelid = c.oid
|
||||
AND k.contype IN ('u', 'p')
|
||||
AND a.attnum = ANY(k.conkey)
|
||||
) THEN '1' ELSE '0' END || '\x1f'
|
||||
|| CASE WHEN EXISTS (
|
||||
SELECT 1 FROM pg_constraint k
|
||||
WHERE k.conrelid = c.oid
|
||||
AND k.contype = 'c'
|
||||
AND a.attnum = ANY(k.conkey)
|
||||
) THEN '1' ELSE '0' END
|
||||
FROM ir_model_fields f
|
||||
JOIN pg_class c ON c.relname = replace(f.model, '.', '_')
|
||||
AND c.relkind = 'r'
|
||||
JOIN pg_attribute a ON a.attrelid = c.oid
|
||||
AND a.attname = f.name
|
||||
AND a.attnum > 0
|
||||
AND NOT a.attisdropped
|
||||
WHERE f.store
|
||||
AND f.ttype IN ('char','text','html','integer','float','monetary')
|
||||
ORDER BY f.model, f.name
|
||||
"""
|
||||
|
||||
|
||||
def inspect(database, config_path=None):
|
||||
"""Tous les champs remplaçables de la base, avec leurs trois vérités."""
|
||||
brut = lib_analyse.run_psql(
|
||||
database, REQUETE_CHAMPS, config_path=config_path
|
||||
)
|
||||
champs = []
|
||||
for ligne in brut.splitlines():
|
||||
parts = ligne.split(SEP)
|
||||
if len(parts) != 6:
|
||||
continue
|
||||
champs.append(
|
||||
{
|
||||
"model": parts[0],
|
||||
"name": parts[1],
|
||||
"ttype": parts[2],
|
||||
"pg_type": parts[3],
|
||||
"unique": parts[4] == "1",
|
||||
"checked": parts[5] == "1",
|
||||
}
|
||||
)
|
||||
return champs
|
||||
|
||||
|
||||
def plan(
|
||||
champs,
|
||||
mode,
|
||||
whitelist=(),
|
||||
blacklist=(),
|
||||
inclure_connexion=False,
|
||||
mots=None,
|
||||
):
|
||||
"""Ce qui sera écrit, table par table — avant d'écrire quoi que ce soit.
|
||||
|
||||
Rendu séparément de l'exécution pour que le mode « à blanc » montre
|
||||
EXACTEMENT ce que `--apply` ferait, et non une approximation.
|
||||
"""
|
||||
modeles = choisir_modeles(
|
||||
{c["model"] for c in champs}, mode, whitelist, blacklist
|
||||
)
|
||||
retenus = set(modeles)
|
||||
par_modele = {}
|
||||
for champ in champs:
|
||||
if champ["model"] not in retenus:
|
||||
continue
|
||||
if not champ_retenu(champ, inclure_connexion):
|
||||
continue
|
||||
par_modele.setdefault(champ["model"], []).append(champ)
|
||||
etapes = []
|
||||
for modele in modeles:
|
||||
liste = par_modele.get(modele)
|
||||
if not liste:
|
||||
continue
|
||||
sql = sql_pour_table(table_de(modele), liste, mots)
|
||||
if sql:
|
||||
etapes.append({"model": modele, "fields": liste, "sql": sql})
|
||||
return etapes
|
||||
|
||||
|
||||
def render(etapes, applique=False, verbeux=False):
|
||||
"""Le rapport. Il dit ce qui est ÉCARTÉ autant que ce qui est pris."""
|
||||
if not etapes:
|
||||
return f"✅ {t('Nothing to anonymise with these lists.')}"
|
||||
total = sum(len(e["fields"]) for e in etapes)
|
||||
tete = (
|
||||
f"🎭 {len(etapes)} {t('model(s)')}, {total} {t('column(s)')}"
|
||||
f" — {t('written') if applique else t('dry run, nothing written')}"
|
||||
)
|
||||
lignes = [tete, ""]
|
||||
for etape in etapes:
|
||||
textes = [f for f in etape["fields"] if f["ttype"] in TYPES_TEXTE]
|
||||
nombres = [f for f in etape["fields"] if f["ttype"] in TYPES_NOMBRE]
|
||||
traduits = [f for f in textes if f["pg_type"] == "jsonb"]
|
||||
uniques = [f for f in etape["fields"] if f["unique"]]
|
||||
detail = f"{len(textes)} {t('text')}, {len(nombres)} {t('numeric')}"
|
||||
if traduits:
|
||||
detail += f", {len(traduits)} {t('translated (jsonb)')}"
|
||||
if uniques:
|
||||
detail += f", {len(uniques)} {t('unique')}"
|
||||
lignes.append(f" {etape['model']:<34} {detail}")
|
||||
if verbeux:
|
||||
for champ in etape["fields"]:
|
||||
lignes.append(
|
||||
f" {champ['name']:<30} {champ['ttype']}"
|
||||
f" / {champ['pg_type']}"
|
||||
)
|
||||
if not applique:
|
||||
lignes.append("")
|
||||
lignes.append(f" {t('Use --apply --confirm <database> to write.')}")
|
||||
return "\n".join(lignes)
|
||||
|
||||
|
||||
def charger_mots(chemin):
|
||||
"""Lire un fichier Python qui déclare MOTS.
|
||||
|
||||
Une liste — les mêmes mots partout — ou un dictionnaire par nom de
|
||||
champ avec un repli `*`. Aucun réseau, aucun modèle : des mots.
|
||||
"""
|
||||
if not chemin:
|
||||
return None
|
||||
espace = {}
|
||||
with open(chemin, "r", encoding="utf-8") as handle:
|
||||
exec(compile(handle.read(), chemin, "exec"), espace) # noqa: S102
|
||||
mots = espace.get("MOTS")
|
||||
if not mots:
|
||||
raise ValueError(f"{t('This file declares no MOTS:')} {chemin}")
|
||||
return mots
|
||||
|
||||
|
||||
def ecrire(database, etapes, config_path=None, timeout=900):
|
||||
"""Écrire les UPDATE — TOUS, ou AUCUN. None si tout a réussi.
|
||||
|
||||
Une seule transaction (`-1`) et `ON_ERROR_STOP=1`. Sans cela, une
|
||||
collision d'unicité au dixième modèle laisserait une base à moitié
|
||||
anonymisée — c'est-à-dire une base dont plus personne ne peut dire ce
|
||||
qui est vrai, et que rien ne rattrape sinon une restauration.
|
||||
|
||||
On reprend `pg_env` pour la connexion — hôte, port, mot de passe lus
|
||||
dans config.conf — et l'on ne lève QUE la lecture seule. La
|
||||
redéclarer ici, ce serait accepter qu'elle diverge un jour.
|
||||
"""
|
||||
import subprocess
|
||||
|
||||
env = lib_analyse.pg_env(config_path, timeout=timeout)
|
||||
env["PGOPTIONS"] = f"-c statement_timeout={timeout}s"
|
||||
sql = "\n".join(etape["sql"] for etape in etapes)
|
||||
done = subprocess.run(
|
||||
[
|
||||
"psql",
|
||||
"-X",
|
||||
"-w",
|
||||
"-1",
|
||||
"-v",
|
||||
"ON_ERROR_STOP=1",
|
||||
"-d",
|
||||
database,
|
||||
"-tA",
|
||||
"-c",
|
||||
sql,
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=timeout + 60,
|
||||
)
|
||||
if done.returncode:
|
||||
detail = (done.stderr or "").strip().splitlines()
|
||||
return detail[0][:200] if detail else "psql"
|
||||
return None
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description=t("Replace the data of a COPY with meaningless data."),
|
||||
)
|
||||
parser.add_argument("-d", "--database", required=True)
|
||||
parser.add_argument("-c", "--config", help="odoo config file")
|
||||
parser.add_argument("--mode", choices=MODES, default="hybrid")
|
||||
parser.add_argument(
|
||||
"--models", default="", help=t("comma separated, adds to the mode")
|
||||
)
|
||||
parser.add_argument(
|
||||
"--exclude", default="", help=t("comma separated, removed from it")
|
||||
)
|
||||
parser.add_argument("--words", help=t("python file declaring MOTS"))
|
||||
parser.add_argument("--include-logins", action="store_true")
|
||||
parser.add_argument("--apply", action="store_true")
|
||||
parser.add_argument(
|
||||
"--confirm",
|
||||
default="",
|
||||
help=t("repeat the database name; --apply refuses without it"),
|
||||
)
|
||||
parser.add_argument("--verbose", action="store_true")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
# Le garde-fou, AVANT toute lecture : personne ne doit découvrir en
|
||||
# route qu'il a lancé cela sur la mauvaise base.
|
||||
if args.apply and args.confirm != args.database:
|
||||
print(
|
||||
f"❌ {t('Refusing to write: --confirm must repeat')}"
|
||||
f" '{args.database}'.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 2
|
||||
|
||||
try:
|
||||
lib_analyse.require_odoo_database(
|
||||
args.database, config_path=args.config
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
print(f"❌ {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
try:
|
||||
mots = charger_mots(args.words)
|
||||
champs = inspect(args.database, args.config)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
print(f"❌ {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
etapes = plan(
|
||||
champs,
|
||||
args.mode,
|
||||
[m.strip() for m in args.models.split(",") if m.strip()],
|
||||
[m.strip() for m in args.exclude.split(",") if m.strip()],
|
||||
args.include_logins,
|
||||
mots,
|
||||
)
|
||||
if not args.apply:
|
||||
print(render(etapes, applique=False, verbeux=args.verbose))
|
||||
return 1 if etapes else 0
|
||||
|
||||
erreur = ecrire(args.database, etapes, args.config)
|
||||
if erreur:
|
||||
print(f"❌ {t('Nothing was written:')} {erreur}", file=sys.stderr)
|
||||
return 2
|
||||
print(render(etapes, applique=True, verbeux=args.verbose))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -68,7 +68,7 @@ ANALYSES = (
|
|||
" step-by-step databases.",
|
||||
"script": "script/analyse/check_migration_residue.py",
|
||||
"kinds": (KIND_DATABASE,),
|
||||
"needs_sql": "It reads pg_catalog — indexes and real tables — which"
|
||||
"why_not": "It reads pg_catalog — indexes and real tables — which"
|
||||
" no RPC session exposes.",
|
||||
},
|
||||
{
|
||||
|
|
@ -78,13 +78,28 @@ ANALYSES = (
|
|||
" administrator — read for the use you intend.",
|
||||
"script": "script/analyse/check_instance_state.py",
|
||||
"kinds": (KIND_DATABASE,),
|
||||
"needs_sql": "Several checks read tables no RPC session exposes,"
|
||||
"why_not": "Several checks read tables no RPC session exposes,"
|
||||
" and the lateness of a job is computed in SQL.",
|
||||
# Le même chiffre veut dire deux choses opposées selon qu'on
|
||||
# ausculte une copie ou une production : l'attente ne se devine
|
||||
# pas, elle se demande.
|
||||
"asks_expect": True,
|
||||
},
|
||||
{
|
||||
"key": "anonymize",
|
||||
"title": "Anonymise a copy",
|
||||
"why": "Replace names, texts and numbers with meaningless ones so"
|
||||
" the copy can be shared. No AI: words from a list, numbers drawn"
|
||||
" at random, written by SQL.",
|
||||
"script": "script/analyse/anonymize.py",
|
||||
"kinds": (KIND_DATABASE,),
|
||||
# La seule entrée qui ÉCRIT. Le refus d'une instance vivante n'a
|
||||
# rien à voir avec le SQL : on ne détruit pas les données d'une
|
||||
# production, même à la demande.
|
||||
"why_not": "This one WRITES. It is offered only for a database"
|
||||
" you restored here — never for an instance in service.",
|
||||
"writes": True,
|
||||
},
|
||||
{
|
||||
"key": "cow_views",
|
||||
"title": "Customised views, website copies included",
|
||||
|
|
@ -92,7 +107,7 @@ ANALYSES = (
|
|||
" shadows a module view.",
|
||||
"script": "script/analyse/analyse_view_custom.py",
|
||||
"kinds": (KIND_DATABASE,),
|
||||
"needs_sql": "Comparing a copy with the module view it hides is a"
|
||||
"why_not": "Comparing a copy with the module view it hides is a"
|
||||
" join on arch_db, done in SQL.",
|
||||
},
|
||||
{
|
||||
|
|
@ -102,7 +117,7 @@ ANALYSES = (
|
|||
" them have no column behind them.",
|
||||
"script": "script/analyse/analyse_custom_field.py",
|
||||
"kinds": (KIND_DATABASE,),
|
||||
"needs_sql": "Telling a declared field from a real column means"
|
||||
"why_not": "Telling a declared field from a real column means"
|
||||
" reading pg_attribute.",
|
||||
},
|
||||
)
|
||||
|
|
|
|||
|
|
@ -63,7 +63,7 @@ def rows(kind):
|
|||
analyse["key"],
|
||||
t(analyse["title"]),
|
||||
utilisable,
|
||||
"" if utilisable else t(analyse["needs_sql"]),
|
||||
"" if utilisable else t(analyse["why_not"]),
|
||||
)
|
||||
)
|
||||
return lignes
|
||||
|
|
@ -79,7 +79,7 @@ def detail(key, kind):
|
|||
morceaux.append(f"▶ {t('Enter to run it.')}")
|
||||
else:
|
||||
morceaux.append(f"✖ {t('Not available for this source.')}")
|
||||
morceaux.append(f" {t(analyse['needs_sql'])}")
|
||||
morceaux.append(f" {t(analyse['why_not'])}")
|
||||
morceaux.append("")
|
||||
morceaux.append(f" {analyse['script']}")
|
||||
return "\n".join(morceaux)
|
||||
|
|
|
|||
|
|
@ -9640,6 +9640,130 @@ TRANSLATIONS = {
|
|||
"fr": "Une instance en service",
|
||||
"en": "An instance in service",
|
||||
},
|
||||
"Anonymise a copy": {
|
||||
"fr": "🎭 Anonymiser une copie",
|
||||
"en": "🎭 Anonymise a copy",
|
||||
},
|
||||
"Replace names, texts and numbers with meaningless ones so the copy can be shared. No AI: words from a list, numbers drawn at random, written by SQL.": {
|
||||
"fr": "Remplacer noms, textes et nombres par des valeurs sans intérêt, pour pouvoir partager la copie. Sans IA : des mots pris dans une liste, des nombres tirés au hasard, écrits en SQL.",
|
||||
"en": "Replace names, texts and numbers with meaningless ones so the copy can be shared. No AI: words from a list, numbers drawn at random, written by SQL.",
|
||||
},
|
||||
"This one WRITES. It is offered only for a database you restored here — never for an instance in service.": {
|
||||
"fr": "Celle-ci ÉCRIT. Elle n'est offerte que pour une base restaurée ici — jamais pour une instance en service.",
|
||||
"en": "This one WRITES. It is offered only for a database you restored here — never for an instance in service.",
|
||||
},
|
||||
"This DESTROYS the data of": {
|
||||
"fr": "Ceci DÉTRUIT les données de",
|
||||
"en": "This DESTROYS the data of",
|
||||
},
|
||||
"there is no undo.": {
|
||||
"fr": "il n'y a pas de retour en arrière.",
|
||||
"en": "there is no undo.",
|
||||
},
|
||||
"Type the database name to confirm (empty to cancel): ": {
|
||||
"fr": "Retapez le nom de la base pour confirmer (vide pour annuler) : ",
|
||||
"en": "Type the database name to confirm (empty to cancel): ",
|
||||
},
|
||||
"Cancelled: nothing was written.": {
|
||||
"fr": "Annulé : rien n'a été écrit.",
|
||||
"en": "Cancelled: nothing was written.",
|
||||
},
|
||||
"Hybrid: the default personal-data models, adjusted": {
|
||||
"fr": "Hybride : les modèles à données personnelles par défaut, ajustés",
|
||||
"en": "Hybrid: the default personal-data models, adjusted",
|
||||
},
|
||||
"Whitelist: only the models I name": {
|
||||
"fr": "Liste blanche : seulement les modèles que je nomme",
|
||||
"en": "Whitelist: only the models I name",
|
||||
},
|
||||
"Blacklist: every model except those I name": {
|
||||
"fr": "Liste noire : tous les modèles sauf ceux que je nomme",
|
||||
"en": "Blacklist: every model except those I name",
|
||||
},
|
||||
"Models to ADD, comma separated (empty for none): ": {
|
||||
"fr": "Modèles à AJOUTER, séparés par des virgules (vide pour aucun) : ",
|
||||
"en": "Models to ADD, comma separated (empty for none): ",
|
||||
},
|
||||
"Models to EXCLUDE, comma separated: ": {
|
||||
"fr": "Modèles à EXCLURE, séparés par des virgules : ",
|
||||
"en": "Models to EXCLUDE, comma separated: ",
|
||||
},
|
||||
"A whitelist with no model would do nothing.": {
|
||||
"fr": "Une liste blanche sans modèle ne ferait rien.",
|
||||
"en": "A whitelist with no model would do nothing.",
|
||||
},
|
||||
"Python file declaring MOTS (empty for the built-in): ": {
|
||||
"fr": "Fichier Python déclarant MOTS (vide pour la liste intégrée) : ",
|
||||
"en": "Python file declaring MOTS (empty for the built-in): ",
|
||||
},
|
||||
"Nothing to anonymise with these lists.": {
|
||||
"fr": "Rien à anonymiser avec ces listes.",
|
||||
"en": "Nothing to anonymise with these lists.",
|
||||
},
|
||||
"column(s)": {
|
||||
"fr": "colonne(s)",
|
||||
"en": "column(s)",
|
||||
},
|
||||
"dry run, nothing written": {
|
||||
"fr": "marche à blanc, rien n'est écrit",
|
||||
"en": "dry run, nothing written",
|
||||
},
|
||||
"text": {
|
||||
"fr": "texte",
|
||||
"en": "text",
|
||||
},
|
||||
"numeric": {
|
||||
"fr": "nombre",
|
||||
"en": "numeric",
|
||||
},
|
||||
"translated (jsonb)": {
|
||||
"fr": "traduit (jsonb)",
|
||||
"en": "translated (jsonb)",
|
||||
},
|
||||
"unique": {
|
||||
"fr": "unique",
|
||||
"en": "unique",
|
||||
},
|
||||
"Use --apply --confirm <database> to write.": {
|
||||
"fr": "Utiliser --apply --confirm <base> pour écrire.",
|
||||
"en": "Use --apply --confirm <database> to write.",
|
||||
},
|
||||
"Replace the data of a COPY with meaningless data.": {
|
||||
"fr": "Remplacer les données d'une COPIE par des données sans intérêt.",
|
||||
"en": "Replace the data of a COPY with meaningless data.",
|
||||
},
|
||||
"comma separated, adds to the mode": {
|
||||
"fr": "séparés par des virgules, s'ajoutent au mode",
|
||||
"en": "comma separated, adds to the mode",
|
||||
},
|
||||
"comma separated, removed from it": {
|
||||
"fr": "séparés par des virgules, retirés du mode",
|
||||
"en": "comma separated, removed from it",
|
||||
},
|
||||
"python file declaring MOTS": {
|
||||
"fr": "fichier python déclarant MOTS",
|
||||
"en": "python file declaring MOTS",
|
||||
},
|
||||
"repeat the database name; --apply refuses without it": {
|
||||
"fr": "répéter le nom de la base ; --apply refuse sans lui",
|
||||
"en": "repeat the database name; --apply refuses without it",
|
||||
},
|
||||
"Refusing to write: --confirm must repeat": {
|
||||
"fr": "Écriture refusée : --confirm doit répéter",
|
||||
"en": "Refusing to write: --confirm must repeat",
|
||||
},
|
||||
"Nothing was written:": {
|
||||
"fr": "Rien n'a été écrit :",
|
||||
"en": "Nothing was written:",
|
||||
},
|
||||
"This file declares no MOTS:": {
|
||||
"fr": "Ce fichier ne déclare aucun MOTS :",
|
||||
"en": "This file declares no MOTS:",
|
||||
},
|
||||
"Unknown mode:": {
|
||||
"fr": "Mode inconnu :",
|
||||
"en": "Unknown mode:",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
368
test/test_anonymize.py
Normal file
368
test/test_anonymize.py
Normal file
|
|
@ -0,0 +1,368 @@
|
|||
#!/usr/bin/env python3
|
||||
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
||||
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
||||
|
||||
"""Anonymiser : ce qui compte, c'est ce qu'on REFUSE de toucher.
|
||||
|
||||
Remplacer des mots est facile. Ce qui casse une base, c'est de croire que
|
||||
« tous les champs string » veut dire quelque chose. Mesuré sur une base
|
||||
réelle en 18 : 505 champs `selection` sont stockés en varchar
|
||||
(`res.partner.lang`, `sale.order.invoice_status`), 2693 many2one sont des
|
||||
entiers, 194 textes sont des `jsonb` par langue, et 301 contraintes
|
||||
d'unicité attendent une collision.
|
||||
|
||||
Trois de ces pièges ont été trouvés en LANÇANT l'outil sur une copie
|
||||
jetable, pas en le relisant : PostgreSQL refuse d'indexer un
|
||||
`ARRAY[...]` sans parenthèses, `res_partner.credit_limit` est un jsonb
|
||||
alors qu'Odoo l'appelle `float`, et `crm_lead.probability` porte un CHECK
|
||||
qui interdit 1000. Chacun aurait fait échouer l'écriture — et l'écriture
|
||||
étant transactionnelle, chaque fois la base est restée intacte. Ce
|
||||
fichier fige ces trois-là pour qu'ils ne reviennent pas.
|
||||
|
||||
Le plancher est l'objet du premier bloc : aucun mode, aucune liste
|
||||
blanche, aucune insistance ne doit permettre d'écrire dans `ir.*`.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from script.analyse import anonymize as anon
|
||||
|
||||
MOTEUR = (
|
||||
Path(__file__).resolve().parent.parent
|
||||
/ "script"
|
||||
/ "analyse"
|
||||
/ "anonymize.py"
|
||||
)
|
||||
|
||||
|
||||
def champ(
|
||||
nom,
|
||||
ttype="char",
|
||||
pg="character varying",
|
||||
unique=False,
|
||||
checked=False,
|
||||
modele="res.partner",
|
||||
):
|
||||
return {
|
||||
"model": modele,
|
||||
"name": nom,
|
||||
"ttype": ttype,
|
||||
"pg_type": pg,
|
||||
"unique": unique,
|
||||
"checked": checked,
|
||||
}
|
||||
|
||||
|
||||
class TestTheFloorNobodyCanLift(unittest.TestCase):
|
||||
"""`ir.*` reste intouchable, quel que soit le chemin."""
|
||||
|
||||
def test_ir_models_are_refused_even_when_whitelisted(self):
|
||||
tous = {"ir.ui.view", "ir.model.fields", "res.partner"}
|
||||
for mode in anon.MODES:
|
||||
choisis = anon.choisir_modeles(
|
||||
tous, mode, whitelist=list(tous), blacklist=[]
|
||||
)
|
||||
self.assertNotIn("ir.ui.view", choisis, mode)
|
||||
self.assertNotIn("ir.model.fields", choisis, mode)
|
||||
|
||||
def test_languages_and_currencies_are_refused_too(self):
|
||||
"""Ce ne sont pas des données personnelles, et les casser casse
|
||||
les adresses et les montants."""
|
||||
tous = {"res.lang", "res.currency", "res.country", "res.partner"}
|
||||
choisis = anon.choisir_modeles(tous, "whitelist", whitelist=list(tous))
|
||||
self.assertEqual(choisis, ["res.partner"])
|
||||
|
||||
def test_a_blacklist_of_nothing_still_respects_the_floor(self):
|
||||
tous = {"ir.ui.view", "res.partner", "res.lang"}
|
||||
self.assertEqual(
|
||||
anon.choisir_modeles(tous, "blacklist", blacklist=[]),
|
||||
["res.partner"],
|
||||
)
|
||||
|
||||
|
||||
class TestWhatIsNeverReplaced(unittest.TestCase):
|
||||
"""Le cœur : distinguer un vrai texte d'un varchar qui n'en est pas un."""
|
||||
|
||||
def test_a_selection_stored_as_varchar_is_left_alone(self):
|
||||
"""505 dans la base mesurée. Y écrire un mot casse l'ORM."""
|
||||
self.assertFalse(anon.champ_retenu(champ("lang", "selection")))
|
||||
self.assertFalse(
|
||||
anon.champ_retenu(champ("invoice_status", "selection"))
|
||||
)
|
||||
|
||||
def test_a_many2one_is_left_alone(self):
|
||||
"""2693 entiers qui sont des relations."""
|
||||
self.assertFalse(
|
||||
anon.champ_retenu(champ("parent_id", "many2one", "integer"))
|
||||
)
|
||||
|
||||
def test_a_many2one_is_refused_by_its_type_not_only_its_name(self):
|
||||
"""Sur la base mesurée, tous les many2one finissent par `_id` — mais
|
||||
un module maison peut en nommer un `owner`, et la convention ne
|
||||
peut pas être la seule barrière. C'est le TYPE qui décide."""
|
||||
self.assertFalse(
|
||||
anon.champ_retenu(champ("owner", "many2one", "integer"))
|
||||
)
|
||||
self.assertFalse(
|
||||
anon.champ_retenu(champ("responsable", "many2one", "integer"))
|
||||
)
|
||||
|
||||
def test_anything_named_like_a_relation_is_left_alone(self):
|
||||
for nom in ("company_id", "tag_ids", "partner_id"):
|
||||
self.assertFalse(anon.champ_retenu(champ(nom, "integer")), nom)
|
||||
|
||||
def test_a_number_living_in_a_jsonb_is_left_alone(self):
|
||||
"""Mesuré : res_partner.credit_limit est un float DANS un jsonb."""
|
||||
self.assertFalse(
|
||||
anon.champ_retenu(champ("credit_limit", "float", "jsonb"))
|
||||
)
|
||||
|
||||
def test_a_column_under_a_check_constraint_is_left_alone(self):
|
||||
"""Mesuré : crm_lead.probability doit rester entre 0 et 100."""
|
||||
self.assertFalse(
|
||||
anon.champ_retenu(
|
||||
champ("probability", "float", "numeric", checked=True)
|
||||
)
|
||||
)
|
||||
|
||||
def test_technical_columns_are_left_alone(self):
|
||||
for nom in (
|
||||
"id",
|
||||
"create_uid",
|
||||
"write_date",
|
||||
"state",
|
||||
"sequence",
|
||||
"arch_db",
|
||||
"active",
|
||||
):
|
||||
self.assertFalse(anon.champ_retenu(champ(nom, "char")), nom)
|
||||
|
||||
def test_logins_stay_unless_asked(self):
|
||||
"""Sinon personne ne peut plus ouvrir la copie qu'on anonymise."""
|
||||
self.assertFalse(anon.champ_retenu(champ("login")))
|
||||
self.assertTrue(
|
||||
anon.champ_retenu(champ("login"), inclure_connexion=True)
|
||||
)
|
||||
|
||||
def test_a_real_text_is_taken(self):
|
||||
for ttype in ("char", "text", "html"):
|
||||
self.assertTrue(anon.champ_retenu(champ("name", ttype)), ttype)
|
||||
|
||||
def test_a_real_number_is_taken(self):
|
||||
for ttype in ("integer", "float", "monetary"):
|
||||
self.assertTrue(
|
||||
anon.champ_retenu(champ("amount", ttype, "numeric")), ttype
|
||||
)
|
||||
|
||||
|
||||
class TestTheSqlItWrites(unittest.TestCase):
|
||||
def test_a_null_stays_null(self):
|
||||
"""Un NULL devenu mot créerait de la donnée là où il n'y en avait
|
||||
pas : la copie mentirait dans l'autre sens."""
|
||||
sql = anon.expression_texte(champ("name"), ["a"])
|
||||
self.assertIn("IS NULL THEN NULL", sql)
|
||||
self.assertIn(
|
||||
"IS NULL THEN NULL",
|
||||
anon.expression_nombre(champ("x", "integer", "integer")),
|
||||
)
|
||||
|
||||
def test_the_array_is_parenthesised(self):
|
||||
"""PostgreSQL refuse d'indexer un ARRAY[...] nu — mesuré."""
|
||||
sql = anon.expression_texte(champ("name"), ["a", "b"])
|
||||
self.assertIn("(ARRAY[", sql)
|
||||
self.assertNotIn("] ARRAY[", sql)
|
||||
self.assertRegex(sql, r"\(ARRAY\[[^\]]*\]\)\[")
|
||||
|
||||
def test_a_unique_column_gets_the_id_appended(self):
|
||||
"""Deux lignes au même mot feraient échouer TOUT l'UPDATE."""
|
||||
sql = anon.expression_texte(champ("ref", unique=True), ["a"])
|
||||
self.assertIn("id::text", sql)
|
||||
self.assertNotIn(
|
||||
"id::text", anon.expression_texte(champ("ref"), ["a"])
|
||||
)
|
||||
|
||||
def test_a_translated_column_is_rebuilt_key_by_key(self):
|
||||
"""Écrire une chaîne dans un jsonb détruirait la colonne."""
|
||||
sql = anon.expression_texte(champ("comment", "html", "jsonb"), ["a"])
|
||||
self.assertIn("jsonb_object_agg", sql)
|
||||
self.assertIn("jsonb_each_text", sql)
|
||||
|
||||
def test_a_plain_text_column_is_not_treated_as_json(self):
|
||||
sql = anon.expression_texte(champ("comment", "text", "text"), ["a"])
|
||||
self.assertNotIn("jsonb", sql)
|
||||
|
||||
def test_numbers_land_between_zero_and_a_thousand(self):
|
||||
entier = anon.expression_nombre(champ("n", "integer", "integer"))
|
||||
self.assertIn("1001", entier)
|
||||
decimal = anon.expression_nombre(champ("x", "float", "numeric"))
|
||||
self.assertIn("1000", decimal)
|
||||
|
||||
def test_a_word_with_a_quote_cannot_break_out(self):
|
||||
"""Une liste de mots vient d'un fichier : elle n'est pas de confiance."""
|
||||
sql = anon.expression_texte(champ("name"), ["l'ete"])
|
||||
self.assertIn("'l''ete'", sql)
|
||||
|
||||
def test_one_update_per_table_not_per_column(self):
|
||||
sql = anon.sql_pour_table(
|
||||
"res_partner", [champ("name"), champ("ref")], None
|
||||
)
|
||||
self.assertEqual(sql.count("UPDATE"), 1)
|
||||
self.assertTrue(sql.endswith(";"))
|
||||
|
||||
def test_no_column_means_no_statement(self):
|
||||
self.assertIsNone(anon.sql_pour_table("res_partner", [], None))
|
||||
|
||||
|
||||
class TestTheModes(unittest.TestCase):
|
||||
TOUS = {"res.partner", "crm.lead", "sale.order", "ir.ui.view"}
|
||||
|
||||
def test_whitelist_takes_only_what_it_names(self):
|
||||
self.assertEqual(
|
||||
anon.choisir_modeles(self.TOUS, "whitelist", ["crm.lead"]),
|
||||
["crm.lead"],
|
||||
)
|
||||
|
||||
def test_blacklist_takes_everything_else(self):
|
||||
choisis = anon.choisir_modeles(
|
||||
self.TOUS, "blacklist", blacklist=["crm.lead"]
|
||||
)
|
||||
self.assertNotIn("crm.lead", choisis)
|
||||
self.assertIn("res.partner", choisis)
|
||||
|
||||
def test_hybrid_starts_from_the_defaults_and_adjusts(self):
|
||||
choisis = anon.choisir_modeles(
|
||||
self.TOUS,
|
||||
"hybrid",
|
||||
whitelist=["sale.order"],
|
||||
blacklist=["res.partner"],
|
||||
)
|
||||
self.assertIn("sale.order", choisis)
|
||||
self.assertIn("crm.lead", choisis) # dans les défauts
|
||||
self.assertNotIn("res.partner", choisis) # retiré
|
||||
|
||||
def test_an_unknown_mode_raises_rather_than_guessing(self):
|
||||
with self.assertRaises(ValueError):
|
||||
anon.choisir_modeles(self.TOUS, "peut-etre")
|
||||
|
||||
|
||||
class TestTheWordList(unittest.TestCase):
|
||||
def test_a_flat_list_is_used_everywhere(self):
|
||||
self.assertEqual(anon.mots_pour("name", ["a", "b"]), ("a", "b"))
|
||||
|
||||
def test_a_dictionary_can_answer_per_field(self):
|
||||
mots = {"email": ["a@b.c"], "*": ["mot"]}
|
||||
self.assertEqual(anon.mots_pour("email", mots), ("a@b.c",))
|
||||
self.assertEqual(anon.mots_pour("name", mots), ("mot",))
|
||||
|
||||
def test_an_empty_list_falls_back_to_the_built_in(self):
|
||||
self.assertEqual(anon.mots_pour("name", []), anon.MOTS_PAR_DEFAUT)
|
||||
|
||||
|
||||
class TestThereIsNoModelInTheLoop(unittest.TestCase):
|
||||
"""« sans passer par un GPT » : vérifié sur le code, pas sur parole."""
|
||||
|
||||
def test_the_engine_imports_nothing_that_could_call_out(self):
|
||||
arbre = ast.parse(MOTEUR.read_text(encoding="utf-8"))
|
||||
interdits = {
|
||||
"requests",
|
||||
"urllib",
|
||||
"urllib3",
|
||||
"http",
|
||||
"httpx",
|
||||
"socket",
|
||||
"openai",
|
||||
"anthropic",
|
||||
"xmlrpc",
|
||||
"json",
|
||||
}
|
||||
for noeud in ast.walk(arbre):
|
||||
noms = []
|
||||
if isinstance(noeud, ast.Import):
|
||||
noms = [a.name.split(".")[0] for a in noeud.names]
|
||||
elif isinstance(noeud, ast.ImportFrom) and noeud.module:
|
||||
noms = [noeud.module.split(".")[0]]
|
||||
for nom in noms:
|
||||
self.assertNotIn(nom, interdits, nom)
|
||||
|
||||
def test_the_write_is_one_transaction(self):
|
||||
"""Une collision au dixième modèle laisserait une base à moitié
|
||||
anonymisée, que rien ne rattrape sinon une restauration."""
|
||||
arbre = ast.parse(MOTEUR.read_text(encoding="utf-8"))
|
||||
fonction = [
|
||||
n
|
||||
for n in ast.walk(arbre)
|
||||
if isinstance(n, ast.FunctionDef) and n.name == "ecrire"
|
||||
]
|
||||
self.assertEqual(len(fonction), 1)
|
||||
# Le corps SANS la docstring : le texte la mentionne, l'argument
|
||||
# doit être réellement passé.
|
||||
corps = [
|
||||
n
|
||||
for n in fonction[0].body
|
||||
if not (
|
||||
isinstance(n, ast.Expr)
|
||||
and isinstance(n.value, ast.Constant)
|
||||
and isinstance(n.value.value, str)
|
||||
)
|
||||
]
|
||||
litteraux = {
|
||||
n.value
|
||||
for bloc in corps
|
||||
for n in ast.walk(bloc)
|
||||
if isinstance(n, ast.Constant) and isinstance(n.value, str)
|
||||
}
|
||||
self.assertIn("-1", litteraux)
|
||||
self.assertIn("ON_ERROR_STOP=1", litteraux)
|
||||
|
||||
|
||||
class TestTheRefusalToWrite(unittest.TestCase):
|
||||
"""Le refus doit précéder la connexion.
|
||||
|
||||
Sinon ces deux tests passent au vert parce que la base n'existe pas,
|
||||
et ne prouvent rien du garde. On lit donc le message rendu, et l'on
|
||||
vérifie qu'aucun psql n'a été appelé.
|
||||
"""
|
||||
|
||||
def _refus(self, argv):
|
||||
import contextlib
|
||||
import io as flux
|
||||
|
||||
appels = []
|
||||
vrai = anon.lib_analyse.require_odoo_database
|
||||
|
||||
def espion(*a, **k):
|
||||
appels.append(a)
|
||||
return vrai(*a, **k)
|
||||
|
||||
anon.lib_analyse.require_odoo_database = espion
|
||||
sortie = flux.StringIO()
|
||||
try:
|
||||
with contextlib.redirect_stderr(sortie):
|
||||
code = anon.main(argv)
|
||||
finally:
|
||||
anon.lib_analyse.require_odoo_database = vrai
|
||||
return code, sortie.getvalue(), appels
|
||||
|
||||
def test_apply_without_a_matching_confirm_is_refused(self):
|
||||
code, message, appels = self._refus(
|
||||
["-d", "une_base", "--apply", "--confirm", "une_autre"]
|
||||
)
|
||||
self.assertEqual(code, 2)
|
||||
self.assertIn(
|
||||
anon.t("Refusing to write: --confirm must repeat"), message
|
||||
)
|
||||
self.assertEqual(appels, [], "la base a été contactée pour refuser")
|
||||
|
||||
def test_apply_with_no_confirm_at_all_is_refused(self):
|
||||
code, message, appels = self._refus(["-d", "une_base", "--apply"])
|
||||
self.assertEqual(code, 2)
|
||||
self.assertIn(
|
||||
anon.t("Refusing to write: --confirm must repeat"), message
|
||||
)
|
||||
self.assertEqual(appels, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -63,7 +63,7 @@ class TestWhatASourceAdmits(unittest.TestCase):
|
|||
|
||||
def test_every_analysis_says_why_it_cannot_do_live(self):
|
||||
for analyse in monitoring.ANALYSES:
|
||||
self.assertTrue(analyse["needs_sql"].strip(), analyse["key"])
|
||||
self.assertTrue(analyse["why_not"].strip(), analyse["key"])
|
||||
|
||||
def test_available_and_unavailable_cover_every_analysis(self):
|
||||
for genre in (monitoring.KIND_DATABASE, monitoring.KIND_LIVE):
|
||||
|
|
|
|||
Loading…
Reference in a new issue