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()