mathben/erplibre
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity

[ADD] migration : mesurer la dérive du plan comptable entre paliers

Browse source 
Une montée de version recharge le gabarit de localisation sans rien dire :
trois des scripts l10n du noyau appellent try_loading() sans
force_create=False, et tout compte que le gabarit n'apparie pas par code
est CRÉÉ. Les groupes ajoutés reclassent alors le plan existant.

Un compte absolu ne dit rien : seul l'ÉCART entre deux paliers de la même
migration se juge — comptes sans écriture, noms homonymes, grille apparue
de rien. L'outil ne répare pas : la clé naturelle de suppression attrape
aussi des comptes réappariés par code, et des ajoutés sont accrochés en ON
DELETE SET NULL. Il dit de rejouer le palier ; 24 tests le couvrent.

--- EN ---

A version bump reloads the localisation template silently: three of the
core l10n scripts call try_loading() without force_create=False, and every
account the template cannot match by code is CREATED. The groups thus
added then reclassify the existing chart.

An absolute count says nothing: only the GAP between two steps of one
migration can be judged — accounts with no entry, homonymous names, a
grouping grid from nothing. The tool does not repair: its natural deletion
key also catches accounts re-matched by code, and added ones are pinned by
ON DELETE SET NULL. It says to replay the step; 24 tests cover it.

Assisted-by: Claude Opus 5
(cherry picked from commit e99575967e0ad5b5122a2c35df3b9eaef7a72d53)
This commit is contained in:
Mathieu Benoit 2026-08-28 03:37:57 -04:00
parent d2832f5c60
commit 17295fd7c3
4 changed files with 812 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

376
script/analyse/check_chart_drift.py Executable file
View file

@ -0,0 +1,376 @@
#!/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 montée de version ajoute au référentiel comptable, sans le dire.
Le cas qui a fait écrire cet outil, mesuré sur une migration 12 → 18 :
palier 17 : 96 comptes, 32 taxes, 14 positions, 0 groupe
palier 18 : 427 comptes, 64 taxes, 28 positions, 168 groupes
Le script de migration de `l10n_ca` recharge le plan canadien `ca_2023`
par `try_loading(...)` SANS `force_create=False` — 31 des 33 scripts de
localisation du noyau le passent, trois l'oublient. Les codes du client
(sa propre numérotation) ne recouvrent celui du gabarit que sur 15 codes
sur 341 : les 326 autres ont été CRÉÉS.
Rien n'est détruit — les écritures ne bougent pas, les journaux gardent
leurs comptes. Mais trois écrans se remplissent de doublons, et surtout
les 168 groupes RECLASSENT les comptes du client : un compte fournisseurs
de 1067 écritures apparaît sous « Residential Mortgage Loans ».
Pourquoi une comparaison, et non un compte absolu
-------------------------------------------------
« 427 comptes » ne dit rien : un plan peut légitimement en compter mille.
Ce qui se juge, c'est l'ÉCART entre deux paliers de la même migration, et
la signature qui l'accompagne — des comptes ajoutés sans une écriture,
des noms qui deviennent homonymes, une grille de regroupement qui
apparaît là où il n'y en avait aucune.
L'outil ne répare rien. Il ne peut pas : la clé naturelle de suppression
attrape aussi des comptes du client réappariés par code, et douze des
comptes ajoutés sont accrochés en ON DELETE SET NULL. La bonne réponse
est de rejouer le palier avec le correctif, pas d'opérer la base livrée.
"""
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 check_migration_quality as quality # noqa: E402
from script.analyse import lib_analyse # noqa: E402
COULEURS = {
"broken": "\033[31m",
"watch": "\033[33m",
"ok": "\033[32m",
"step": "\033[36m",
"dim": "\033[90m",
}
RESET = "\033[0m"
def paint(texte, genre, colour):
"""Teinter, ou rendre le texte tel quel quand la couleur est coupée."""
if not colour:
return texte
return f"{COULEURS.get(genre, '')}{texte}{RESET}"
# Ce que le référentiel comptable porte, et ce qu'un gonflement y veut
# dire. `gravity` suit la règle du dépôt : `broken` pour ce qui est faux
# en soi, `watch` pour ce qui mérite un regard.
TABLES = (
{
"key": "account",
"table": "account_account",
"title": "Accounts in the chart",
"why": "A reloaded chart template adds every account it could not"
" match by code. They carry no entry, and they bury the"
" customer's own chart.",
"gravity": "broken",
},
{
"key": "tax",
"table": "account_tax",
"title": "Taxes",
"why": "Reloaded taxes are not renamed '[old]': the duplicates are"
" indistinguishable in the interface.",
"gravity": "broken",
},
{
"key": "fiscal_position",
"table": "account_fiscal_position",
"title": "Fiscal positions",
"why": "Each province then appears twice, under the same name.",
"gravity": "broken",
},
{
"key": "group",
"table": "account_group",
"title": "Account groups",
"why": "Groups are matched to accounts by code PREFIX: a grid that"
" appears reclassifies the customer's own accounts, and the trial"
" balance in hierarchy mode follows.",
"gravity": "broken",
},
{
"key": "journal",
"table": "account_journal",
"title": "Journals",
"why": "A journal added by a template reload competes with the"
" customer's own.",
"gravity": "watch",
},
)
def compte(database, table, **kwargs):
"""Combien de lignes, ou None si la table n'existe pas."""
existe = lib_analyse.scalar_query(
database,
f"SELECT (to_regclass('public.{table}') IS NOT NULL)::text;",
**kwargs,
)
if (existe or "").strip().lower() != "true":
return None
brut = lib_analyse.scalar_query(
database, f"SELECT count(*) FROM {table};", **kwargs
)
try:
return int(brut)
except (TypeError, ValueError):
return None
def comptes_sans_ecriture(database, **kwargs):
"""(comptes, ceux qui ne portent aucune écriture).
C'est la signature d'un gabarit rechargé : un plan vécu a des comptes
vides, mais pas trois cents d'un coup.
"""
lignes = lib_analyse.json_query(
database,
"SELECT count(*) AS total,"
" count(*) FILTER (WHERE NOT EXISTS ("
" SELECT 1 FROM account_move_line l WHERE l.account_id = a.id"
" )) AS sans_ecriture"
" FROM account_account a",
**kwargs,
)
if not lignes:
return None, None
return lignes[0].get("total"), lignes[0].get("sans_ecriture")
def homonymes(database, table, **kwargs):
"""Combien de noms portés par plus d'une ligne.
Le rechargement ne renomme pas ce qu'il double : dans l'interface,
deux lignes du même nom ne se distinguent pas.
"""
brut = lib_analyse.scalar_query(
database,
"SELECT coalesce(sum(n - 1), 0) FROM ("
f" SELECT count(*) AS n FROM {table}"
" GROUP BY name->>'en_US' HAVING count(*) > 1) x;",
**kwargs,
)
try:
return int(brut)
except (TypeError, ValueError):
return None
def reclasses(database, **kwargs):
"""Combien de comptes tombent sous un groupe, par préfixe de code.
Rejoue ce que fait `_compute_account_group` : le groupe dont le
préfixe encadre le code, le plus spécifique d'abord.
"""
brut = lib_analyse.scalar_query(
database,
"SELECT count(*) FROM ("
" SELECT a.id, a.code_store->>'1' AS code FROM account_account a"
") c JOIN LATERAL ("
" SELECT ag.id FROM account_group ag"
" WHERE ag.code_prefix_start <="
" left(c.code, char_length(ag.code_prefix_start))"
" AND ag.code_prefix_end >="
" left(c.code, char_length(ag.code_prefix_end))"
" ORDER BY char_length(ag.code_prefix_start) DESC, ag.id"
" LIMIT 1) g ON true;",
**kwargs,
)
try:
return int(brut)
except (TypeError, ValueError):
return None
def previous_database(database, dct=None):
"""Le palier qui précède celui-ci, lu dans la chaîne de la migration.
On ne le devine pas depuis le nom : la base de DÉPART ne porte pas de
suffixe, et retrancher 1 à « _upgrade_18 » donnerait « _upgrade_17 »
même quand la chaîne n'est pas contiguë.
"""
dct = quality.read_progression() if dct is None else dct
chaine = quality.chain(dct)
for rang, (_version, base) in enumerate(chaine):
if base == database and rang:
return chaine[rang - 1][1]
return None
def inspect(after, before, **kwargs):
"""Ce que le dernier palier ajoute au référentiel du précédent."""
rapport = {"before": before, "after": after, "tables": [], "signes": {}}
for entree in TABLES:
avant = compte(before, entree["table"], **kwargs)
apres = compte(after, entree["table"], **kwargs)
rapport["tables"].append(
{
"key": entree["key"],
"table": entree["table"],
"before": avant,
"after": apres,
"delta": (
None if avant is None or apres is None else apres - avant
),
}
)
total, vides = comptes_sans_ecriture(after, **kwargs)
rapport["signes"]["accounts"] = total
rapport["signes"]["accounts_without_entry"] = vides
for table, clef in (
("account_tax", "tax_homonyms"),
("account_fiscal_position", "fiscal_position_homonyms"),
):
rapport["signes"][clef] = homonymes(after, table, **kwargs)
if compte(after, "account_group", **kwargs):
rapport["signes"]["accounts_regrouped"] = reclasses(after, **kwargs)
return rapport
def judge(rapport):
"""[(entrée, écart)] pour ce qui a gonflé, du plus grave au moins."""
par_clef = {e["key"]: e for e in TABLES}
trouve = [
(par_clef[ligne["key"]], ligne)
for ligne in rapport["tables"]
if ligne["delta"] and ligne["delta"] > 0
]
ordre = {"broken": 0, "watch": 1}
trouve.sort(
key=lambda paire: (
ordre.get(paire[0]["gravity"], 2),
-paire[1]["delta"],
)
)
return trouve
def render(rapport, colour=True):
"""Le rapport lisible. L'écart d'abord, ce qu'il signifie ensuite."""
lignes = [
f"📚 {t('Accounting reference data, one step to the next')}",
paint(f" {rapport['before']} → {rapport['after']}", "dim", colour),
"",
]
trouve = judge(rapport)
if not trouve:
lignes.append(
paint(
f"✅ {t('Nothing was added to the reference data.')}",
"ok",
colour,
)
)
return "\n".join(lignes)
for entree, ligne in trouve:
icone = "❌" if entree["gravity"] == "broken" else "⚠"
lignes.append(
paint(
f"{icone} {('+' + str(ligne['delta'])).rjust(6)}"
f" {t(entree['title'])}"
f" ({ligne['before']} → {ligne['after']})",
entree["gravity"],
colour,
)
)
lignes.append(paint(f" {t(entree['why'])}", "dim", colour))
lignes.append("")
signes = rapport["signes"]
lignes.append(paint(f" {t('What it looks like')}", "step", colour))
vides = signes.get("accounts_without_entry")
if vides:
lignes.append(
f" {vides} / {signes.get('accounts')}"
f" {t('accounts carry no entry at all')}"
)
for clef, phrase in (
("tax_homonyms", "taxes share a name with another"),
(
"fiscal_position_homonyms",
"fiscal positions share a name with another",
),
):
if signes.get(clef):
lignes.append(f" {signes[clef]} {t(phrase)}")
if signes.get("accounts_regrouped"):
lignes.append(
f" {signes['accounts_regrouped']}"
f" {t('accounts fall under a group by code prefix')}"
)
lignes.append("")
lignes.append(
paint(
f" {t('Replay the step with the fix; do not operate the')}"
f" {t('delivered database.')}",
"dim",
colour,
)
)
return "\n".join(lignes).rstrip()
def main(argv=None):
import argparse
import json
parser = argparse.ArgumentParser(
description=t("What a version bump adds to the accounting data."),
)
parser.add_argument("-d", "--database", required=True)
parser.add_argument(
"--before",
help=t("the step to compare against (default: the previous one)"),
)
parser.add_argument("-c", "--config", help="odoo config file")
parser.add_argument("--json", action="store_true")
parser.add_argument("--no-color", action="store_true")
args = parser.parse_args(argv)
before = args.before or previous_database(args.database)
if not before:
print(
f"❌ {t('No previous step found: name it with --before.')}",
file=sys.stderr,
)
return 2
for base in (before, args.database):
try:
lib_analyse.require_odoo_database(base, config_path=args.config)
except Exception as exc: # noqa: BLE001
print(f"❌ {exc}", file=sys.stderr)
return 2
rapport = inspect(args.database, before, config_path=args.config)
if args.json:
print(json.dumps(rapport, indent=2, ensure_ascii=False))
else:
colour = sys.stdout.isatty() and not args.no_color
print(render(rapport, colour))
return 1 if judge(rapport) else 0
if __name__ == "__main__":
sys.exit(main())

5
script/analyse/check_migration_quality.py
View file

@ -390,6 +390,11 @@ REVUE = (
"script/analyse/check_view_type_tree.py",
"viewtype",
),
(
"Did the last step add to the accounting data?",
"script/analyse/check_chart_drift.py -d {db}",
"chartdrift",
),
)

93
script/todo/todo_i18n.py
View file

@ -6979,6 +6979,99 @@ TRANSLATIONS = {
},
# --- Écran de qualité : Verdicts, Validation, Revue ---
# --- Journal d'étape et bascule de version ---
# --- Dérive du référentiel comptable ---
"Did the last step add to the accounting data?": {
"fr": "Le dernier palier a-t-il gonflé le référentiel comptable ?",
"en": "Did the last step add to the accounting data?",
},
"Accounting reference data, one step to the next": {
"fr": "Référentiel comptable, d'un palier au suivant",
"en": "Accounting reference data, one step to the next",
},
"Nothing was added to the reference data.": {
"fr": "Rien n'a été ajouté au référentiel.",
"en": "Nothing was added to the reference data.",
},
"Accounts in the chart": {
"fr": "Comptes du plan comptable",
"en": "Accounts in the chart",
},
"Taxes": {
"fr": "Taxes",
"en": "Taxes",
},
"Fiscal positions": {
"fr": "Positions fiscales",
"en": "Fiscal positions",
},
"Account groups": {
"fr": "Groupes de comptes",
"en": "Account groups",
},
"Journals": {
"fr": "Journaux",
"en": "Journals",
},
"What it looks like": {
"fr": "À quoi cela ressemble",
"en": "What it looks like",
},
"accounts carry no entry at all": {
"fr": "comptes ne portent aucune écriture",
"en": "accounts carry no entry at all",
},
"taxes share a name with another": {
"fr": "taxes portent le nom d'une autre",
"en": "taxes share a name with another",
},
"fiscal positions share a name with another": {
"fr": "positions fiscales portent le nom d'une autre",
"en": "fiscal positions share a name with another",
},
"accounts fall under a group by code prefix": {
"fr": "comptes tombent sous un groupe par préfixe de code",
"en": "accounts fall under a group by code prefix",
},
"Replay the step with the fix; do not operate the": {
"fr": "Rejouer le palier avec le correctif ; ne pas opérer la",
"en": "Replay the step with the fix; do not operate the",
},
"delivered database.": {
"fr": "base livrée.",
"en": "delivered database.",
},
"What a version bump adds to the accounting data.": {
"fr": "Ce qu'une montée de version ajoute au référentiel comptable.",
"en": "What a version bump adds to the accounting data.",
},
"the step to compare against (default: the previous one)": {
"fr": "le palier de comparaison (défaut : le précédent)",
"en": "the step to compare against (default: the previous one)",
},
"No previous step found: name it with --before.": {
"fr": "Aucun palier précédent trouvé : le nommer avec --before.",
"en": "No previous step found: name it with --before.",
},
"A reloaded chart template adds every account it could not match by code. They carry no entry, and they bury the customer's own chart.": {
"fr": "Un gabarit de plan rechargé ajoute chaque compte qu'il n'a pu apparier par son code. Ils ne portent aucune écriture, et noient le plan du client.",
"en": "A reloaded chart template adds every account it could not match by code. They carry no entry, and they bury the customer's own chart.",
},
"Reloaded taxes are not renamed '[old]': the duplicates are indistinguishable in the interface.": {
"fr": "Les taxes rechargées ne sont pas renommées « [old] » : les doublons ne se distinguent pas dans l'interface.",
"en": "Reloaded taxes are not renamed '[old]': the duplicates are indistinguishable in the interface.",
},
"Each province then appears twice, under the same name.": {
"fr": "Chaque province apparaît alors deux fois, sous le même nom.",
"en": "Each province then appears twice, under the same name.",
},
"Groups are matched to accounts by code PREFIX: a grid that appears reclassifies the customer's own accounts, and the trial balance in hierarchy mode follows.": {
"fr": "Les groupes s'apparient aux comptes par PRÉFIXE de code : une grille qui apparaît reclasse les comptes du client, et la balance en mode hiérarchie suit.",
"en": "Groups are matched to accounts by code PREFIX: a grid that appears reclassifies the customer's own accounts, and the trial balance in hierarchy mode follows.",
},
"A journal added by a template reload competes with the customer's own.": {
"fr": "Un journal ajouté par un rechargement concurrence celui du client.",
"en": "A journal added by a template reload competes with the customer's own.",
},
"findings": {
"fr": "des trouvailles",
"en": "findings",

338
test/test_check_chart_drift.py Normal file
View file

@ -0,0 +1,338 @@
#!/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 montée de version ajoute au référentiel comptable.
Mesuré sur une migration 12 → 18 réelle, entre les deux derniers paliers :
comptes 96 → 427 taxes 32 → 64
groupes 0 → 168 positions 14 → 28
Le script de migration de `l10n_ca` recharge le plan `ca_2023` sans
`force_create=False` ; les codes du client ne recouvrent le gabarit que
sur 15 des 341, et les 326 autres sont créés. Rien n'est détruit — mais
les 168 groupes reclassent les comptes du client par PRÉFIXE de code, et
un compte fournisseurs de 1067 écritures se retrouve sous « Residential
Mortgage Loans ».
Ce qui se teste ici n'est pas le comptage — psql le fait — mais le
JUGEMENT : un compte absolu ne dit rien (un plan peut légitimement porter
mille comptes), seul l'écart entre deux paliers de la même migration se
juge. Et la comparaison doit rester muette quand rien n'a bougé, sinon
personne ne la lira deux fois.
"""
import io
import os
import sys
import unittest
from contextlib import redirect_stdout
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, REPO)
from script.analyse import check_chart_drift as drift # noqa: E402
from script.analyse import lib_analyse # noqa: E402
class Base(unittest.TestCase):
"""psql bouchonné : chaque base rend les comptes qu'on lui prête."""
def setUp(self):
self.vues = []
self.ancien = lib_analyse.run_psql
self.addCleanup(setattr, lib_analyse, "run_psql", self.ancien)
self.tables = {}
self.sans_ecriture = (0, 0)
self.homonymes = 0
self.reclasses = 0
lib_analyse.run_psql = self.repondre
def poser(self, base, **compteurs):
self.tables[base] = compteurs
def repondre(self, database, sql, **kwargs):
self.vues.append((database, sql))
compteurs = self.tables.get(database, {})
if "to_regclass" in sql and "count(*)" not in sql:
table = sql.split("public.")[1].split("'")[0]
return "true" if table in compteurs else "false"
if "account_move_line l WHERE l.account_id" in sql:
total, vides = self.sans_ecriture
return '[{"total": %d, "sans_ecriture": %d}]' % (total, vides)
if "HAVING count(*) > 1" in sql:
return str(self.homonymes)
if "code_prefix_start" in sql:
return str(self.reclasses)
for table, valeur in compteurs.items():
if f"FROM {table};" in sql:
return str(valeur)
return ""
class TestWhatItCompares(Base):
def test_it_reads_both_steps(self):
self.poser("avant", account_account=96)
self.poser("apres", account_account=427)
drift.inspect("apres", "avant")
bases = {base for base, _sql in self.vues}
self.assertEqual({"avant", "apres"}, bases)
def test_a_table_that_does_not_exist_is_not_a_zero(self):
# Sans le module account, la table manque : dire « 0 → 0 » ferait
# croire à une mesure, et « 0 → 400 » à une explosion.
self.poser("avant")
self.poser("apres", account_account=427)
rapport = drift.inspect("apres", "avant")
ligne = [x for x in rapport["tables"] if x["key"] == "account"][0]
self.assertIsNone(ligne["before"])
self.assertIsNone(ligne["delta"])
def test_every_declared_table_is_measured(self):
self.poser("avant", account_account=1)
self.poser("apres", account_account=1)
rapport = drift.inspect("apres", "avant")
self.assertEqual(
{e["key"] for e in drift.TABLES},
{x["key"] for x in rapport["tables"]},
)
class TestWhatItJudges(Base):
def rapport(self, avant, apres):
self.poser("avant", **avant)
self.poser("apres", **apres)
return drift.inspect("apres", "avant")
def test_nothing_added_is_nothing_to_report(self):
rapport = self.rapport(
{"account_account": 96}, {"account_account": 96}
)
self.assertEqual([], drift.judge(rapport))
def test_a_chart_that_quadruples_is_a_finding(self):
rapport = self.rapport(
{"account_account": 96}, {"account_account": 427}
)
trouve = drift.judge(rapport)
self.assertEqual(1, len(trouve))
self.assertEqual(331, trouve[0][1]["delta"])
def test_a_chart_that_SHRINKS_is_not_reported_here(self):
# Ce n'est pas le sujet de cet outil, et le dire ferait deux
# constats pour deux causes différentes sous un seul nom.
rapport = self.rapport(
{"account_account": 427}, {"account_account": 96}
)
self.assertEqual([], drift.judge(rapport))
def test_groups_appearing_from_nothing_are_judged(self):
# Zéro groupe puis cent soixante-huit : c'est la grille du gabarit,
# et elle reclasse les comptes du client.
rapport = self.rapport({"account_group": 0}, {"account_group": 168})
self.assertEqual("group", drift.judge(rapport)[0][0]["key"])
def test_gravity_wins_over_size(self):
rapport = self.rapport(
{"account_journal": 8, "account_account": 96},
{"account_journal": 28, "account_account": 99},
)
trouve = drift.judge(rapport)
cles = [entree["key"] for entree, _ in trouve]
self.assertEqual("account", cles[0], "la gravite doit primer")
self.assertEqual("journal", cles[-1])
# Et l'écart le plus GROS est bien celui qu'on relègue.
deltas = dict(zip(cles, [ligne["delta"] for _e, ligne in trouve]))
self.assertGreater(deltas["journal"], deltas["account"])
def test_at_equal_gravity_the_bigger_gap_comes_first(self):
rapport = self.rapport(
{"account_account": 96, "account_tax": 32},
{"account_account": 427, "account_tax": 64},
)
cles = [entree["key"] for entree, _ in drift.judge(rapport)]
self.assertEqual(["account", "tax"], cles)
def test_every_entry_says_why_it_matters(self):
for entree in drift.TABLES:
self.assertTrue(entree["why"].strip(), entree["key"])
self.assertIn(entree["gravity"], ("broken", "watch"))
class TestTheReport(Base):
def test_it_names_both_steps(self):
self.poser("base_upgrade_17", account_account=96)
self.poser("base_upgrade_18", account_account=427)
rapport = drift.inspect("base_upgrade_18", "base_upgrade_17")
texte = drift.render(rapport, colour=False)
self.assertIn("base_upgrade_17", texte)
self.assertIn("base_upgrade_18", texte)
def test_it_shows_the_before_and_the_after_not_just_the_delta(self):
# « +331 » seul ne dit pas si l'on part de 96 ou de 4000.
self.poser("avant", account_account=96)
self.poser("apres", account_account=427)
texte = drift.render(drift.inspect("apres", "avant"), colour=False)
self.assertIn("+331", texte)
self.assertIn("96", texte)
self.assertIn("427", texte)
def test_a_quiet_migration_says_so_and_stops(self):
self.poser("avant", account_account=96)
self.poser("apres", account_account=96)
texte = drift.render(drift.inspect("apres", "avant"), colour=False)
self.assertIn(
drift.t("Nothing was added to the reference data."), texte
)
self.assertNotIn("❌", texte)
def test_it_points_at_the_replay_not_at_surgery(self):
# La clé naturelle de suppression attrape aussi des comptes du
# client, et douze des ajoutés sont accrochés en SET NULL.
self.poser("avant", account_account=96)
self.poser("apres", account_account=427)
texte = drift.render(drift.inspect("apres", "avant"), colour=False)
self.assertIn(
drift.t("Replay the step with the fix; do not operate the"),
texte,
)
def test_the_shape_is_shown_when_there_is_something_to_show(self):
self.sans_ecriture = (427, 404)
self.homonymes = 15
self.reclasses = 422
self.poser("avant", account_account=96)
self.poser("apres", account_account=427, account_group=168)
texte = drift.render(drift.inspect("apres", "avant"), colour=False)
self.assertIn("404", texte)
self.assertIn("422", texte)
class TestFindingThePreviousStep(unittest.TestCase):
"""Le palier précédent se LIT dans la chaîne, il ne se devine pas."""
def chaine(self):
return {
"config_database_name": "base",
"target_odoo_version": "18.0",
"state_4_upgrade_odoo_lst": [1, 2, 3, 4, 5, 6],
}
def test_it_reads_the_chain_the_migration_wrote(self):
self.assertEqual(
"base_upgrade_17",
drift.previous_database("base_upgrade_18", self.chaine()),
)
def test_the_first_step_has_no_predecessor(self):
# La base de DÉPART ne porte pas de suffixe : lui retrancher 1
# donnerait un nom qui n'existe pas.
self.assertIsNone(drift.previous_database("base", self.chaine()))
def test_a_database_outside_the_chain_is_not_guessed(self):
self.assertIsNone(
drift.previous_database("une_autre_base", self.chaine())
)
class TestTheCommand(Base):
def lancer(self, argv):
tampon = io.StringIO()
with redirect_stdout(tampon):
code = drift.main(argv)
return code, tampon.getvalue()
def setUp(self):
super().setUp()
self.ancien_require = lib_analyse.require_odoo_database
self.addCleanup(
setattr, lib_analyse, "require_odoo_database", self.ancien_require
)
lib_analyse.require_odoo_database = lambda *a, **k: None
def test_without_a_previous_step_it_is_a_tool_failure(self):
# Sortir 0 laisserait croire que la comparaison a eu lieu.
ancien = drift.previous_database
drift.previous_database = lambda *a, **k: None
self.addCleanup(setattr, drift, "previous_database", ancien)
code, _ = self.lancer(["-d", "base_upgrade_18"])
self.assertEqual(2, code)
def test_no_drift_exits_zero(self):
self.poser("avant", account_account=96)
self.poser("apres", account_account=96)
code, _ = self.lancer(
["-d", "apres", "--before", "avant", "--no-color"]
)
self.assertEqual(0, code)
def test_drift_exits_one(self):
# 1 = des trouvailles, la convention des outils du dépôt.
self.poser("avant", account_account=96)
self.poser("apres", account_account=427)
code, _ = self.lancer(
["-d", "apres", "--before", "avant", "--no-color"]
)
self.assertEqual(1, code)
def test_the_json_carries_both_counts(self):
import json
self.poser("avant", account_account=96)
self.poser("apres", account_account=427)
_code, texte = self.lancer(
["-d", "apres", "--before", "avant", "--json"]
)
dct = json.loads(texte)
ligne = [x for x in dct["tables"] if x["key"] == "account"][0]
self.assertEqual(96, ligne["before"])
self.assertEqual(427, ligne["after"])
self.assertEqual(331, ligne["delta"])
class TestItNeverWrites(unittest.TestCase):
"""On compare deux bases de palier : l'une d'elles est parfois la
seule copie d'un état."""
def test_no_statement_of_the_tool_can_write(self):
"""Sur le CODE, jamais sur la prose.
Les docstrings de l'outil expliquent justement pourquoi il ne
faut pas opérer la base — « accrochés en ON DELETE SET NULL » —
et une garde qui lit le texte échoue sur le texte qui l'explique.
"""
import ast
import inspect
arbre = ast.parse(inspect.getsource(drift))
for noeud in ast.walk(arbre):
if isinstance(
noeud, (ast.Module, ast.FunctionDef, ast.ClassDef)
) and ast.get_docstring(noeud):
noeud.body = noeud.body[1:]
code = ast.unparse(arbre).upper()
for interdit in ("UPDATE ", "DELETE ", "INSERT ", "DROP ", "ALTER "):
self.assertNotIn(interdit, code, interdit)
def test_it_only_ever_selects(self):
import ast
import inspect
arbre = ast.parse(inspect.getsource(drift))
requetes = [
n.value
for n in ast.walk(arbre)
if isinstance(n, ast.Constant)
and isinstance(n.value, str)
and "FROM " in n.value.upper()
]
self.assertTrue(requetes)
for requete in requetes:
self.assertTrue(
requete.upper().lstrip().startswith("SELECT"), requete[:40]
)
if __name__ == "__main__":
unittest.main()