From ad632363751e8f5bcb6cfd9c6f115e8b303668db Mon Sep 17 00:00:00 2001 From: Mathieu Benoit Date: Mon, 3 Aug 2026 08:11:20 +0000 Subject: [PATCH] [ADD] analyse: le socle des outils d'analyse d'une base Odoo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Premier fichier de script/analyse/, le paquet d'outils qui répondent à « qu'y a-t-il dans cette base, et qu'est-ce qui cassera à la montée de version ». Ce commit ne livre que ce que le premier outil consommera, pas une bibliothèque spéculative. psql en sous-processus, pas psycopg2. La raison est déjà écrite dans reset_stale_cow_views.py : ces outils tournent sur des bases dont le registre Odoo ne charge PAS — une base 12.0 sur un checkout 18.0 — et c'est précisément le moment où on veut les analyser. Accessoirement psycopg2 n'est pas dans .venv.erplibre, qui est l'interpréteur de ces outils. La lecture seule est une garantie du serveur, pas une promesse du code. PGOPTIONS porte default_transaction_read_only=on, donc PostgreSQL refuse toute écriture sur la connexion. Un SET dans le même -c n'aurait rien donné : psql -c ouvre une transaction implicite unique, et default_transaction_read_only ne vaut que pour les transactions suivantes. PGOPTIONS borne aussi statement_timeout — un scan parti en vrille ne doit pas figer un menu. Le résultat voyage en JSON, pas en champs séparés. Une arch de vue contient des sauts de ligne et des « | » : tout séparateur maison finit par couper au mauvais endroit. json_agg côté PostgreSQL, json.loads côté Python, et le problème n'existe plus. C'est le même choix que snapshot_cow_views.py. Les sondes lisent pg_attribute, pas information_schema, qui est filtré par les droits : avec un rôle non propriétaire elle rendrait un ensemble vide, et l'analyse concluerait « aucune colonne website, donc aucune vue COW » sans le moindre avertissement. Faux, et silencieux. tr_col décide sur le TYPE réel de la colonne, pas sur un numéro de version. Un champ traduit est du jsonb à partir de 16.0 et du texte avant ; une base à moitié migrée porte les deux, et un numéro de version mentirait. Une colonne inconnue rend NULL::text plutôt que du SQL invalide, ce qui permet de sonder et d'interroger d'un trait. MODEL_TABLE_OVERRIDE est dérivée des sources, pas écrite de mémoire : parcours AST des 27 843 .py de odoo18.0/ et addons/, en ne gardant que les classes dont le _table diffère du défaut. Onze entrées. Beaucoup de modules déclarent un _table égal au défaut — les compter ferait croire à 22 surcharges là où il y en a 11. Un modèle absent de la table et dont la table est introuvable est classé « table inconnue », un fait, jamais « table orpheline », une anomalie. Le nom de base est validé au lieu d'être échappé : il finit dans une commande lancée avec shell=True, et l'alphabet qu'une base Odoo utilise vraiment est court. Ce contrôle attrape aussi « Traceback (most recent call last): », que la liste des bases pouvait offrir jusqu'au commit précédent. Pas de __init__.py : script/todo/, script/database/ et script/odoo/migration/ n'en ont pas et sont importés tous les jours. Paquets d'espace de noms. Vérifié. 31 tests sans base — fonctions pures, construction de SQL, « False » de config.conf traité comme non défini, nom de base hostile refusé. Puis à la main sur une base synthétique dont le SQL est dans le docstring du test et a été rejoué tel quel : jsonb et text distingués, le « | » et le saut de ligne préservés à travers json_query, ir.actions.act_window résolu en ir_act_window, une base sans ir_module_module refusée, et un CREATE TABLE refusé par le serveur. La suite passe de 262 à 293 tests, sans nouvel échec. Co-Authored-By: Claude Opus 5 (1M context) --- script/analyse/lib_analyse.py | 391 ++++++++++++++++++++++++++++++++++ test/test_analyse_lib.py | 289 +++++++++++++++++++++++++ 2 files changed, 680 insertions(+) create mode 100644 script/analyse/lib_analyse.py create mode 100644 test/test_analyse_lib.py diff --git a/script/analyse/lib_analyse.py b/script/analyse/lib_analyse.py new file mode 100644 index 0000000..55542a1 --- /dev/null +++ b/script/analyse/lib_analyse.py @@ -0,0 +1,391 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""Socle commun des outils d'analyse d'une base Odoo, en lecture seule. + +Pourquoi psql en sous-processus plutôt que psycopg2 +--------------------------------------------------- +La raison est déjà écrite dans le dépôt, dans ``reset_stale_cow_views.py`` : +« Plain psql on purpose: this runs on databases whose Odoo registry does not +load, which is precisely when it is needed ». Une base 12.0 sur un checkout +18.0 ne charge pas son registre, et c'est exactement le moment où on veut +l'analyser. Accessoirement, psycopg2 n'est pas dans ``.venv.erplibre``, qui est +l'interpréteur de ces outils. + +La lecture seule est une garantie, pas une promesse +-------------------------------------------------- +``PGOPTIONS`` porte ``default_transaction_read_only=on`` : c'est **le serveur** +qui refuse toute écriture, pour toutes les transactions de la connexion. Un +``SET`` glissé dans le même ``-c`` ne suffirait pas — ``psql -c`` ouvre une +transaction implicite unique, et ``default_transaction_read_only`` ne vaut que +pour les transactions *suivantes*. + +Ne jamais deviner la forme du schéma +------------------------------------ +Douze versions d'Odoo se partagent ces tables. Les colonnes apparaissent, +changent de type (``text`` puis ``jsonb`` à partir de 16.0), ou n'existent que +si un module est installé. D'où ``existing_columns()`` et ``tr_col()`` : on +sonde avant d'écrire une requête, on ne date pas les colonnes de mémoire. + +Les sondes lisent ``pg_attribute``, pas ``information_schema`` : +``information_schema`` est filtré par les droits. Avec un rôle non +propriétaire, elle renverrait un ensemble vide, et l'analyse concluerait +« aucune colonne website, donc aucune vue COW » sans le moindre avertissement. +""" + +import configparser +import json +import os +import re +import subprocess +import sys + +new_path = os.path.normpath( + os.path.join(os.path.dirname(__file__), "..", "..") +) +sys.path.append(new_path) + +try: + from script.todo.todo_i18n import t +except Exception: # pragma: no cover - repli si i18n indisponible + + def t(key: str) -> str: + return key + + +REPO_ROOT = new_path + +# Une valeur littérale « False » dans config.conf veut dire « non défini » : +# c'est ainsi qu'Odoo écrit l'absence de valeur dans son fichier de config. +CONFIG_UNSET = ("false", "none", "") + +# Un nom de base voyage jusqu'à une commande shell (`odoo_bin.sh shell -d …`, +# lancée avec shell=True par execute.py). On le valide au lieu de compter sur +# l'échappement : la liste de caractères qu'une base Odoo utilise réellement +# est courte, et tout le reste est soit une erreur de frappe, soit une +# injection. +RE_DATABASE_NAME = re.compile(r"[A-Za-z0-9_.-]+") + +# Modèles dont la table N'EST PAS `_name.replace('.', '_')`. +# +# Dérivé des sources, pas écrit de mémoire : parcours AST de tous les `.py` de +# `odoo18.0/` et `addons/` (27 843 fichiers), en gardant les classes dont le +# `_table` diffère du défaut. Beaucoup de modules déclarent un `_table` égal au +# défaut — ce sont des déclarations sans effet, à ne pas confondre avec une +# surcharge. +# +# Sans cette table, `replace('.', '_')` échoue précisément sur les modèles les +# plus fréquents dans `ir_model_data` : `ir.actions.act_window` chercherait +# `ir_actions_act_window`, qui n'existe pas. +# +# La liste n'est pas la vérité, seulement ce qu'on sait : un modèle absent +# d'ici et dont la table est introuvable est classé « table inconnue » (un +# fait), jamais « table orpheline » (une anomalie). Pour la régénérer, refaire +# le parcours AST sur l'arbre courant. +MODEL_TABLE_OVERRIDE = { + "ir.actions.act_multi": "ir_actions", + "ir.actions.act_url": "ir_act_url", + "ir.actions.act_window": "ir_act_window", + "ir.actions.act_window.message": "ir_actions", + "ir.actions.act_window.view": "ir_act_window_view", + "ir.actions.act_window_close": "ir_actions", + "ir.actions.actions": "ir_actions", + "ir.actions.client": "ir_act_client", + "ir.actions.report": "ir_act_report_xml", + "ir.actions.server": "ir_act_server", + "project.task.stage.personal": "project_task_user_rel", + # `ir.actions.report.xml` est le nom d'avant 11.0 : les lignes + # `ir_model_data` d'une base ancienne le portent encore. + "ir.actions.report.xml": "ir_act_report_xml", +} + + +class AnalyseError(Exception): + """Échec de l'outil, pas un constat d'analyse. + + Distinction qui porte le code de retour : 2 pour « je n'ai pas pu + analyser », réservé à cette exception ; 1 pour « j'ai analysé et j'ai + trouvé des constats ». Les confondre rendrait une analyse en échec + indistinguable d'une base à problèmes. + """ + + +def valid_database_name(name): + """Le nom est-il un nom de base plausible, sûr à mettre dans une commande ?""" + return bool(name) and RE_DATABASE_NAME.fullmatch(name) is not None + + +def read_config(config_path=None): + """Lire config.conf, en repliant sur /etc/odoo/odoo.conf comme run.sh. + + Renvoie un dict des options, vide si aucun fichier n'est trouvé — l'absence + de config n'est pas une erreur : sur une installation native, psql se + connecte très bien par le socket unix sans aucun paramètre. + """ + lst_candidate = ( + [config_path] + if config_path + else [ + os.path.join(REPO_ROOT, "config.conf"), + "/etc/odoo/odoo.conf", + ] + ) + for path in lst_candidate: + if path and os.path.isfile(path): + parser = configparser.RawConfigParser() + try: + parser.read(path) + except configparser.Error: + continue + if parser.has_section("options"): + return dict(parser.items("options")) + return {} + + +def pg_env(config_path=None, timeout=120, overrides=None): + """Variables d'environnement pour psql : connexion + lecture seule. + + Les paramètres viennent de config.conf, pas d'une hypothèse « socket unix + et rôle = utilisateur système » : le dépôt lui-même livre + ``db_user = erplibre`` et un docker-compose.yml avec un mot de passe. + + ``PGOPTIONS`` est ce qui rend l'analyse incapable d'écrire, et borne la + durée d'une requête — un scan qui part en vrille ne bloque pas un menu. + """ + config = read_config(config_path) + my_env = os.environ.copy() + + dct_map = { + "db_host": "PGHOST", + "db_port": "PGPORT", + "db_user": "PGUSER", + "db_password": "PGPASSWORD", + "db_sslmode": "PGSSLMODE", + } + for key, var in dct_map.items(): + value = str(config.get(key, "")).strip() + if value.lower() not in CONFIG_UNSET: + my_env[var] = value + for var, value in (overrides or {}).items(): + if value: + my_env[var] = str(value) + + my_env["PGOPTIONS"] = ( + f"-c default_transaction_read_only=on -c statement_timeout={timeout}s" + ) + # Un ~/.psqlrc avec \timing ou \pset ajoute des lignes à la sortie et casse + # le parsing. -X l'ignore, mais PSQLRC vide protège aussi les appels qui + # oublieraient -X. + my_env["PSQLRC"] = "" + return my_env + + +def run_psql(database, sql, timeout=120, config_path=None, overrides=None): + """Exécuter du SQL et rendre la sortie brute, une ligne par enregistrement. + + ``-X`` ignore ~/.psqlrc, ``-w`` interdit l'invite de mot de passe (sans + lui, un mot de passe manquant bloque le menu TODO sans rien afficher), + ``ON_ERROR_STOP=1`` fait échouer au premier problème plutôt que de rendre + une sortie partielle qu'on prendrait pour un résultat. + """ + if not valid_database_name(database): + raise AnalyseError(f"{t('Invalid database name: ')}{database!r}") + cmd = [ + "psql", + "-X", + "-w", + "-v", + "ON_ERROR_STOP=1", + "-d", + database, + "-tAc", + sql, + ] + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout + 30, + cwd=REPO_ROOT, + env=pg_env(config_path, timeout=timeout, overrides=overrides), + ) + except FileNotFoundError as exc: + raise AnalyseError( + f"{t('psql is not installed or not in PATH.')}" + ) from exc + except subprocess.TimeoutExpired as exc: + raise AnalyseError( + f"{t('Query exceeded the timeout (s): ')}{timeout}" + ) from exc + if result.returncode: + raise AnalyseError( + f"{t('Cannot read from the database: ')}" + f"{result.stderr.strip() or result.returncode}" + ) + return result.stdout + + +def json_query(database, sql, **kwargs): + """Rendre le résultat d'un SELECT comme une liste de dicts. + + Le SQL est enveloppé côté PostgreSQL plutôt que découpé côté Python : une + arch de vue contient des retours de ligne et des « | », donc tout + séparateur maison finirait par couper au mauvais endroit. C'est le même + choix que ``snapshot_cow_views.py``. + + ``sql`` est un SELECT SANS point-virgule final : il devient une + sous-requête. + """ + inner = sql.strip().rstrip(";") + wrapped = ( + "SELECT COALESCE(json_agg(row_to_json(t))::text, '[]')" + f" FROM ({inner}) t;" + ) + raw = run_psql(database, wrapped, **kwargs).strip() + if not raw: + return [] + try: + return json.loads(raw) + except ValueError as exc: + raise AnalyseError(f"{t('Unreadable JSON from psql: ')}{exc}") from exc + + +def scalar_query(database, sql, **kwargs): + """Première valeur de la première ligne, ou None si aucune ligne.""" + raw = run_psql(database, sql, **kwargs).strip() + if not raw: + return None + return raw.splitlines()[0].strip() or None + + +def require_odoo_database(database, **kwargs): + """Refuser tout ce qui n'est pas une base Odoo, avant d'aller plus loin. + + Sans ce contrôle, ``-d postgres`` ou une base vide remonte un + « relation "ir_ui_view" does not exist » brut, qui ressemble à un bogue de + l'outil alors que c'est une erreur de saisie. + """ + found = scalar_query( + database, "SELECT to_regclass('public.ir_module_module');", **kwargs + ) + if not found: + raise AnalyseError(f"'{database}' {t('is not an Odoo database.')}") + return True + + +def database_version(database, **kwargs): + """Version Odoo de la BASE, qui n'est pas celle du checkout. + + Comparer les deux est ce qui évite d'ouvrir un shell Odoo pour rien : sur + une base 13.0 avec un checkout 18.0, le registre ne chargera pas, et mieux + vaut le dire tout de suite qu'après trente secondes de chargement. + """ + return scalar_query( + database, + "SELECT latest_version FROM ir_module_module WHERE name = 'base';", + **kwargs, + ) + + +def existing_columns(database, table, **kwargs): + """Colonnes réellement présentes, via pg_attribute (pas information_schema). + + Renvoie un ensemble vide si la table n'existe pas — les deux cas se + distinguent avec ``to_regclass`` si l'appelant en a besoin. + """ + sql = ( + "SELECT a.attname FROM pg_attribute a" + " JOIN pg_class c ON c.oid = a.attrelid" + " JOIN pg_namespace n ON n.oid = c.relnamespace" + f" WHERE n.nspname = 'public' AND c.relname = {quote_literal(table)}" + " AND a.attnum > 0 AND NOT a.attisdropped;" + ) + return { + line.strip() + for line in run_psql(database, sql, **kwargs).splitlines() + if line.strip() + } + + +def column_types(database, table, **kwargs): + """{colonne: type PostgreSQL} — dit `jsonb` là où 15.0 disait `text`.""" + sql = ( + "SELECT a.attname, format_type(a.atttypid, a.atttypmod)" + " FROM pg_attribute a" + " JOIN pg_class c ON c.oid = a.attrelid" + " JOIN pg_namespace n ON n.oid = c.relnamespace" + f" WHERE n.nspname = 'public' AND c.relname = {quote_literal(table)}" + " AND a.attnum > 0 AND NOT a.attisdropped;" + ) + dct_type = {} + for line in run_psql(database, sql, **kwargs).splitlines(): + if "|" in line: + name, _, kind = line.partition("|") + dct_type[name.strip()] = kind.strip() + return dct_type + + +def tr_col(table, column, dct_type, lang="en_US"): + """Fragment SQL lisant un champ traduit, quelle que soit la version. + + À partir de 16.0 un champ traduit est un ``jsonb`` ``{"en_US": "…"}`` ; + jusqu'à 15.0 c'est du texte. Un seul endroit décide, et il décide sur le + type réel de la colonne — pas sur un numéro de version, qu'il faudrait + connaître et qui mentirait sur une base à moitié migrée. + + ``dct_type`` vient de ``column_types()``. Une colonne inconnue rend NULL + plutôt que du SQL invalide : l'appelant verra un champ vide, pas une + requête qui explose. + """ + kind = (dct_type or {}).get(column) + if kind is None: + return "NULL::text" + qualified = f'"{table}"."{column}"' if table else f'"{column}"' + if kind == "jsonb": + return f"{qualified}->>{quote_literal(lang)}" + return f"{qualified}::text" + + +def quote_literal(value): + """Littéral SQL sûr : les quotes simples sont doublées. + + Nécessaire parce que ces requêtes sont assemblées en texte pour psql, sans + paramètres liés. Les seules valeurs concernées ici sont des noms de tables + et de colonnes venant du catalogue, mais un nom de table hérité peut + parfaitement porter une apostrophe. + """ + return "'" + str(value).replace("'", "''") + "'" + + +def model_table(model, known_tables=None): + """Table d'un modèle, ou None si elle est introuvable. + + None veut dire « je ne sais pas », jamais « il n'y en a pas » : c'est ce + qui empêche de classer un modèle à `_table` surchargé comme une anomalie. + ``known_tables`` est l'ensemble des tables réelles, quand l'appelant l'a. + """ + table = MODEL_TABLE_OVERRIDE.get(model, model.replace(".", "_")) + if known_tables is not None and table not in known_tables: + return None + return table + + +def public_tables(database, **kwargs): + """Tables réelles du schéma public, vues par le catalogue. + + ``relkind IN ('r', 'p')`` : 'r' pour une table ordinaire, 'p' pour une + table partitionnée. Odoo n'en partitionne pas, mais le jour où cela + changera, le compte ne doit pas devenir faux en silence. + """ + sql = ( + "SELECT c.relname FROM pg_class c" + " JOIN pg_namespace n ON n.oid = c.relnamespace" + " WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p');" + ) + return { + line.strip() + for line in run_psql(database, sql, **kwargs).splitlines() + if line.strip() + } diff --git a/test/test_analyse_lib.py b/test/test_analyse_lib.py new file mode 100644 index 0000000..43322f0 --- /dev/null +++ b/test/test_analyse_lib.py @@ -0,0 +1,289 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""Socle d'analyse : ce qui se teste SANS base de données. + +Tout ici est une fonction pure ou une fonction dont le seul effet est de +construire du SQL ou un environnement. C'est délibéré : ces tests doivent +tourner sur une machine sans PostgreSQL, donc dans la suite du dépôt. + +Ce qui exige une vraie base — `require_odoo_database`, `database_version`, +`existing_columns`, `column_types`, `public_tables` — n'est pas ici : le test +ne peut pas fabriquer sa base, puisque créer une base est une écriture et que +l'outillage est en lecture seule par construction. + +Ces cinq fonctions s'éprouvent à la main sur une base synthétique, ce qui +prend une minute et n'exige aucun Odoo installé : + + createdb erplibre_analyse_selftest + psql -d erplibre_analyse_selftest -c " + CREATE TABLE ir_module_module (id serial PRIMARY KEY, name varchar, + state varchar, latest_version varchar); + INSERT INTO ir_module_module (name, state, latest_version) + VALUES ('base', 'installed', '18.0.1.3'); + -- arch_db en jsonb : la forme >= 16.0. Le « | » et le saut de ligne + -- sont là exprès : ils prouvent que json_query ne coupe pas dessus. + CREATE TABLE ir_ui_view (id serial PRIMARY KEY, name jsonb, + arch_db jsonb, website_id integer, arch_fs varchar); + INSERT INTO ir_ui_view (name, arch_db) VALUES + ('{\"en_US\":\"Avec un | et un saut\\nde ligne\"}', + '{\"en_US\":\"
\"}'); + -- une colonne text : la forme <= 15.0, pour éprouver tr_col des deux côtés + CREATE TABLE res_partner (id serial PRIMARY KEY, ref text); + -- table dont le nom NE dérive PAS du modèle ir.actions.act_window + CREATE TABLE ir_act_window (id serial PRIMARY KEY, res_model varchar);" + +Attendu : `database_version` rend « 18.0.1.3 » ; `column_types` distingue +`jsonb` de `text` ; `tr_col` produit `->>'en_US'` d'un côté et `::text` de +l'autre ; `model_table('ir.actions.act_window')` rend `ir_act_window` ; +`require_odoo_database` refuse une base sans `ir_module_module` ; et un +`CREATE TABLE` passé à `run_psql` est refusé par le serveur, pas par nous. +Puis `dropdb erplibre_analyse_selftest`. +""" + +import os +import tempfile +import unittest + +from script.analyse import lib_analyse as L + + +class TestValidDatabaseName(unittest.TestCase): + def test_accepts_real_names(self): + for name in ("test", "prod_18", "client.prod", "a-b_c.1"): + self.assertTrue(L.valid_database_name(name), name) + + def test_rejects_a_traceback_line(self): + # Le cas qui a motivé le contrôle : un PostgreSQL injoignable faisait + # remonter sa trace d'appel comme si c'était une liste de bases. + self.assertFalse( + L.valid_database_name("Traceback (most recent call last):") + ) + + def test_rejects_shell_injection(self): + for name in ("a; DROP DATABASE b", "a b", "$(id)", "a|b", "a'b"): + self.assertFalse(L.valid_database_name(name), name) + + def test_rejects_empty(self): + self.assertFalse(L.valid_database_name("")) + self.assertFalse(L.valid_database_name(None)) + + +class TestReadConfig(unittest.TestCase): + def _write(self, body): + handle = tempfile.NamedTemporaryFile( + "w", suffix=".conf", delete=False, encoding="utf-8" + ) + handle.write(body) + handle.close() + self.addCleanup(os.unlink, handle.name) + return handle.name + + def test_reads_the_options_section(self): + path = self._write("[options]\ndb_user = erplibre\ndb_port = 5433\n") + config = L.read_config(path) + self.assertEqual(config["db_user"], "erplibre") + self.assertEqual(config["db_port"], "5433") + + def test_missing_file_is_not_an_error(self): + # Sur une installation native, psql se connecte par le socket unix sans + # aucun paramètre : l'absence de config est un cas normal. + self.assertEqual(L.read_config("/nowhere/absent.conf"), {}) + + def test_file_without_options_section(self): + path = self._write("[other]\nfoo = bar\n") + self.assertEqual(L.read_config(path), {}) + + +class TestPgEnv(unittest.TestCase): + def _write(self, body): + handle = tempfile.NamedTemporaryFile( + "w", suffix=".conf", delete=False, encoding="utf-8" + ) + handle.write(body) + handle.close() + self.addCleanup(os.unlink, handle.name) + return handle.name + + def test_read_only_and_timeout_are_always_set(self): + env = L.pg_env("/nowhere/absent.conf", timeout=42) + self.assertIn("default_transaction_read_only=on", env["PGOPTIONS"]) + self.assertIn("statement_timeout=42s", env["PGOPTIONS"]) + + def test_psqlrc_is_neutralised(self): + # Un ~/.psqlrc avec \timing ajoute des lignes à la sortie et casse le + # parsing JSON. + self.assertEqual(L.pg_env("/nowhere/absent.conf")["PSQLRC"], "") + + def test_config_values_become_pg_variables(self): + path = self._write( + "[options]\ndb_host = pg.example.org\ndb_port = 5433\n" + "db_user = erplibre\ndb_password = s3cret\ndb_sslmode = require\n" + ) + env = L.pg_env(path) + self.assertEqual(env["PGHOST"], "pg.example.org") + self.assertEqual(env["PGPORT"], "5433") + self.assertEqual(env["PGUSER"], "erplibre") + self.assertEqual(env["PGPASSWORD"], "s3cret") + self.assertEqual(env["PGSSLMODE"], "require") + + def test_literal_false_means_unset(self): + # Odoo écrit « False » dans config.conf pour dire « pas de valeur ». + # L'exporter tel quel ferait chercher un hôte nommé « False ». + path = self._write( + "[options]\ndb_host = False\ndb_port = False\n" + "db_password = False\ndb_user = erplibre\n" + ) + env = L.pg_env(path) + self.assertNotIn("PGHOST", env) + self.assertNotIn("PGPORT", env) + self.assertNotIn("PGPASSWORD", env) + self.assertEqual(env["PGUSER"], "erplibre") + + def test_overrides_win_over_the_config(self): + path = self._write("[options]\ndb_host = pg.example.org\n") + env = L.pg_env(path, overrides={"PGHOST": "127.0.0.1"}) + self.assertEqual(env["PGHOST"], "127.0.0.1") + + +class TestQuoteLiteral(unittest.TestCase): + def test_wraps_in_single_quotes(self): + self.assertEqual(L.quote_literal("ir_ui_view"), "'ir_ui_view'") + + def test_doubles_embedded_quotes(self): + self.assertEqual(L.quote_literal("a'b"), "'a''b'") + + +class TestTrCol(unittest.TestCase): + """Le fragment SQL d'un champ traduit, décidé sur le TYPE réel. + + Pas sur un numéro de version : une base à moitié migrée porte les deux + formes, et un numéro de version mentirait. + """ + + def test_jsonb_column_is_unpacked(self): + # La forme >= 16.0 : {"en_US": "..."} + got = L.tr_col("ir_ui_view", "arch_db", {"arch_db": "jsonb"}) + self.assertEqual(got, '"ir_ui_view"."arch_db"->>\'en_US\'') + + def test_text_column_is_cast(self): + # La forme <= 15.0 : du texte + got = L.tr_col("ir_ui_view", "arch_db", {"arch_db": "text"}) + self.assertEqual(got, '"ir_ui_view"."arch_db"::text') + + def test_character_varying_is_cast_too(self): + got = L.tr_col("ir_model", "name", {"name": "character varying"}) + self.assertEqual(got, '"ir_model"."name"::text') + + def test_unknown_column_yields_null_not_broken_sql(self): + # Une colonne absente doit donner un champ vide, pas une requête qui + # explose : c'est ce qui permet de sonder puis d'interroger d'un trait. + self.assertEqual(L.tr_col("ir_ui_view", "absente", {}), "NULL::text") + self.assertEqual(L.tr_col("ir_ui_view", "absente", None), "NULL::text") + + def test_other_language(self): + got = L.tr_col("ir_model", "name", {"name": "jsonb"}, lang="fr_CA") + self.assertEqual(got, '"ir_model"."name"->>\'fr_CA\'') + + +class TestModelTable(unittest.TestCase): + def test_default_derivation(self): + self.assertEqual(L.model_table("res.partner"), "res_partner") + + def test_override_is_applied(self): + # replace('.', '_') donnerait ir_actions_act_window, qui n'existe pas. + self.assertEqual( + L.model_table("ir.actions.act_window"), "ir_act_window" + ) + + def test_unknown_table_returns_none_not_a_guess(self): + # None veut dire « je ne sais pas », jamais « il n'y en a pas » : c'est + # ce qui empêche de classer un modèle abstrait comme une anomalie. + self.assertIsNone( + L.model_table("mail.thread", known_tables={"res_partner"}) + ) + + def test_known_table_passes_through(self): + self.assertEqual( + L.model_table("res.partner", known_tables={"res_partner"}), + "res_partner", + ) + + +class TestModelTableOverrideIntegrity(unittest.TestCase): + """La table de surcharges ne doit pas accumuler d'entrées inutiles. + + Beaucoup de modules déclarent un `_table` égal au défaut. Recopier une + telle déclaration ici serait du poids mort qui donne l'illusion d'une + surcharge — et c'est exactement la confusion qui a fait annoncer 22 + surcharges là où il y en a 11. + """ + + def test_no_entry_equals_the_default(self): + for model, table in L.MODEL_TABLE_OVERRIDE.items(): + self.assertNotEqual( + table, + model.replace(".", "_"), + f"'{model}' n'est pas une surcharge : son _table est le défaut", + ) + + def test_tables_look_like_table_names(self): + for model, table in L.MODEL_TABLE_OVERRIDE.items(): + self.assertRegex(table, r"^[a-z][a-z0-9_]*$", model) + + +class TestJsonQuery(unittest.TestCase): + """L'enveloppe json_agg, sans base : on intercepte run_psql.""" + + def setUp(self): + self.seen = [] + self.reply = "[]" + self.original = L.run_psql + + def fake_run_psql(database, sql, **kwargs): + self.seen.append((database, sql)) + return self.reply + + L.run_psql = fake_run_psql + self.addCleanup(setattr, L, "run_psql", self.original) + + def test_wraps_the_select(self): + L.json_query("db", "SELECT id FROM ir_ui_view") + _, sql = self.seen[0] + self.assertIn("json_agg(row_to_json(t))", sql) + self.assertIn("FROM (SELECT id FROM ir_ui_view) t", sql) + + def test_strips_a_trailing_semicolon(self): + # Un « ; » resté dans la sous-requête produirait du SQL invalide. + L.json_query("db", "SELECT id FROM ir_ui_view;") + _, sql = self.seen[0] + self.assertNotIn(";) t", sql) + + def test_empty_output_is_an_empty_list(self): + self.reply = " \n" + self.assertEqual(L.json_query("db", "SELECT 1"), []) + + def test_pipes_and_newlines_survive(self): + # La raison d'être de l'enveloppe : une arch XML contient des « | » et + # des sauts de ligne, donc tout séparateur maison couperait au mauvais + # endroit. + self.reply = '[{"arch": "a | b\\nc"}]' + rows = L.json_query("db", "SELECT 1") + self.assertEqual(rows, [{"arch": "a | b\nc"}]) + + def test_unreadable_output_raises_analyse_error(self): + self.reply = "ERREUR: quelque chose" + with self.assertRaises(L.AnalyseError): + L.json_query("db", "SELECT 1") + + +class TestRunPsqlGuards(unittest.TestCase): + def test_hostile_database_name_never_reaches_psql(self): + with self.assertRaises(L.AnalyseError) as caught: + L.run_psql("a; DROP DATABASE b", "SELECT 1;") + self.assertIn("Invalid database name", str(caught.exception)) + + +if __name__ == "__main__": + unittest.main()