diff --git a/script/analyse/check_instance_state.py b/script/analyse/check_instance_state.py new file mode 100755 index 0000000..ae3c74e --- /dev/null +++ b/script/analyse/check_instance_state.py @@ -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) + +"""Dans quel état est cette instance — et pour quel usage on la destine. + +Le même chiffre veut dire deux choses opposées. Zéro cron actif est le +SUCCÈS attendu d'une copie de développement, et une panne totale sur une +production. Zéro serveur de courriel rassure sur l'une, condamne l'autre. +Un rapport qui ignore cette différence crie au loup sur ce que l'on vient +de demander, et l'on cesse alors de le lire — c'est la faute que ce dépôt +a déjà corrigée trois fois. + +L'attente est donc DÉCLARÉE, `--expect copy` ou `--expect live`, et +chaque contrôle dit ce qu'il juge sous l'une et sous l'autre. Ce qui n'a +pas de sens sous l'attente courante n'est pas affiché en vert : il est +affiché comme non jugé, avec la raison. + +Mesuré, et c'est ce qui a fixé la conception +-------------------------------------------- +Sur la base 12 d'ORIGINE et sur sa migrée 18, jamais démarrées : + + crons en retard de plus d'un cycle 11 et 8 + lignes db_backup 0 et 0 + +Le retard n'a rien à voir avec la migration : personne ne fait tourner le +cadenceur d'une base restaurée. Sous `--expect copy`, ces deux contrôles +ne sont donc pas jugés du tout. Les afficher en rouge aurait été du bruit +pur ; les afficher en vert, un mensonge. + +Ce qui se juge SOUS LES DEUX +---------------------------- +La neutralisation, elle, se mesure. Sur sept bases dont le nom portait +« neutralize », `database.is_neutralized` était absent des sept, avec +jusqu'à 35 crons actifs et le domaine de courriel du CLIENT en place. + +On ne lit que des BOOLÉENS de présence pour tout ce qui touche à un +secret. Une clé de paiement vivante a été trouvée dans une base de test ; +un rapport finit dans un billet ou devant un agent, et n'a aucune raison +de la porter. + +Lecture seule : `default_transaction_read_only=on`, imposé par le serveur. +""" + +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 + +COPY = "copy" +LIVE = "live" +ATTENTES = (COPY, LIVE) + +COULEURS = { + "broken": "\033[31m", + "watch": "\033[33m", + "ok": "\033[32m", + "dim": "\033[90m", +} +RESET = "\033[0m" + + +def paint(texte, genre, colour): + if not colour: + return texte + return f"{COULEURS.get(genre, '')}{texte}{RESET}" + + +# Une politique dit COMMENT lire le nombre sous une attente donnée : +# ("zero", gravité) il doit valoir zéro +# ("nonzero", gravité) il doit être non nul +# ("info",) on montre, on ne juge pas +# ("skip", raison) sans objet ici, et l'on DIT pourquoi +CONTROLES = ( + { + "key": "neutralized", + "section": "Neutralisation", + "title": "database.is_neutralized", + "sql": "SELECT count(*) FROM ir_config_parameter" + " WHERE key='database.is_neutralized'" + " AND value IN ('true','True','1')", + "copy": ("nonzero", "broken"), + "live": ("zero", "broken"), + "why_copy": "Odoo does not consider this database neutralised;" + " modules that neutralise themselves never ran.", + "why_live": "A production marked as neutralised has had its" + " scheduled actions and outgoing mail disabled.", + }, + { + "key": "mail_server_open", + "section": "Neutralisation", + "title": "Mail servers that could actually send", + "sql": "SELECT count(*) FROM ir_mail_server WHERE active" + " AND coalesce(smtp_host,'') NOT IN ('invalid','localhost.invalid')", + "copy": ("zero", "broken"), + "live": ("info",), + "why_copy": "A copy that can send reaches the customer's real" + " contacts.", + "why_live": "", + }, + { + "key": "mail_server_total", + "section": "Neutralisation", + "title": "Mail servers declared at all", + "sql": "SELECT count(*) FROM ir_mail_server", + "copy": ("nonzero", "watch"), + "live": ("nonzero", "watch"), + "why_copy": "With NO server, Odoo falls back to smtp_server from" + " the config file — which is why Odoo's own neutralize.sql" + " INSERTS a blocking one instead of deleting them all.", + "why_live": "With no server at all, Odoo silently uses the" + " smtp_server from the config file.", + }, + { + "key": "payment_live", + "section": "Neutralisation", + "title": "Payment providers neither disabled nor in test", + "sql": "SELECT count(*) FROM payment_provider" + " WHERE state NOT IN ('disabled','test')", + "copy": ("zero", "broken"), + "live": ("info",), + "why_copy": "A copy can charge real cards.", + "why_live": "", + }, + { + "key": "url_mismatch", + "section": "Neutralisation", + "title": "Mail domain that does not match the base URL", + "sql": "SELECT CASE WHEN EXISTS (SELECT 1 FROM ir_config_parameter c" + " JOIN ir_config_parameter u ON u.key='web.base.url'" + " WHERE c.key='mail.catchall.domain'" + " AND position(c.value in u.value) = 0) THEN 1 ELSE 0 END", + "copy": ("info",), + "live": ("zero", "watch"), + "why_copy": "", + "why_live": "Portal links, invoice QR codes and reply addresses" + " point somewhere else than the instance itself.", + }, + { + "key": "cron_active", + "section": "Scheduler", + "title": "Scheduled actions active", + "sql": "SELECT count(*) FROM ir_cron WHERE active", + "copy": ("zero", "watch"), + "live": ("nonzero", "broken"), + "why_copy": "A copy that still runs them sends mail and posts" + " payments.", + "why_live": "Nothing at all runs on a schedule.", + }, + { + "key": "cron_late", + "section": "Scheduler", + "title": "Scheduled actions late by more than one cycle", + "sql": "SELECT count(*) FROM ir_cron c WHERE c.active" + " AND c.nextcall <" + " now() - (c.interval_number || ' ' || c.interval_type)::interval", + "copy": ( + "skip", + "Nobody runs the scheduler on a restored copy:" + " measured 11 late on an untouched source database.", + ), + "live": ("zero", "broken"), + "why_copy": "", + "why_live": "A fixed threshold misses them — a ten-minute job one" + " hour late matters, a monthly one does not.", + }, + { + "key": "backup_rows", + "section": "Backups", + "title": "Backup configurations recorded", + "sql": "SELECT count(*) FROM db_backup", + "copy": ( + "skip", + "update_prod_to_dev deletes them on purpose;" + " their absence here proves nothing.", + ), + "live": ("nonzero", "broken"), + "why_copy": "", + "why_live": "The backup scheduler runs and backs up nothing.", + }, + { + "key": "mail_stuck", + "section": "Queues", + "title": "Messages stuck in the outgoing queue", + "sql": "SELECT count(*) FROM mail_mail" + " WHERE state IN ('outgoing','exception')", + "copy": ("info",), + "live": ("zero", "watch"), + "why_copy": "", + "why_live": "Visible to the customer, invisible to the operator.", + }, + { + "key": "all_internal_are_admin", + "section": "Users", + "title": "Internal users who are ALL system administrators", + "sql": "SELECT CASE WHEN (SELECT count(*) FROM res_users u" + " WHERE u.active AND EXISTS (SELECT 1 FROM res_groups_users_rel r" + " JOIN ir_model_data d ON d.model='res.groups' AND d.res_id=r.gid" + " WHERE r.uid=u.id AND d.module='base' AND d.name='group_user'))" + " = (SELECT count(*) FROM res_users u WHERE u.active" + " AND EXISTS (SELECT 1 FROM res_groups_users_rel r" + " JOIN ir_model_data d ON d.model='res.groups' AND d.res_id=r.gid" + " WHERE r.uid=u.id AND d.module='base' AND d.name='group_system'))" + " THEN 1 ELSE 0 END", + "copy": ("zero", "watch"), + "live": ("zero", "watch"), + "why_copy": "No ordinary user exists to test visibility with — any" + " access-rights check run here proves nothing.", + "why_live": "Everyone can change everything, and no rule is ever" + " exercised.", + }, +) + +SECTIONS = ("Neutralisation", "Scheduler", "Backups", "Queues", "Users") + + +def inspect(database, config_path=None): + """Passer chaque contrôle. Un contrôle illisible n'est PAS un zéro.""" + resultats = {} + for controle in CONTROLES: + try: + brut = lib_analyse.run_psql( + database, controle["sql"], config_path=config_path + ).strip() + resultats[controle["key"]] = int(brut.splitlines()[0]) + except Exception as exc: # noqa: BLE001 - on rapporte, on ne meurt pas + resultats[controle["key"]] = { + "error": str(exc).splitlines()[0][:120] + } + return resultats + + +def verdict(controle, valeur, attente): + """(genre, gravité, raison) pour ce contrôle sous cette attente. + + `genre` vaut ok, bad, info, skip ou unreadable — jamais autre chose, + et surtout jamais « ok » par défaut : un contrôle sans politique est un + oubli, pas une bonne nouvelle. + """ + if isinstance(valeur, dict): + return "unreadable", "watch", valeur.get("error", "") + politique = controle.get(attente) + if not politique: + return "skip", "dim", "" + genre = politique[0] + if genre == "skip": + return "skip", "dim", t(politique[1]) + if genre == "info": + return "info", "dim", "" + raison = t(controle.get(f"why_{attente}", "") or "") + if genre == "zero": + return ( + ("ok", "ok", "") if not valeur else ("bad", politique[1], raison) + ) + if genre == "nonzero": + return ("ok", "ok", "") if valeur else ("bad", politique[1], raison) + return "skip", "dim", "" + + +def findings(resultats, attente): + """Ce qui ne va pas, et rien d'autre — pour le code de sortie.""" + mauvais = [] + for controle in CONTROLES: + genre, gravite, raison = verdict( + controle, resultats.get(controle["key"]), attente + ) + if genre == "bad": + mauvais.append((controle, gravite, raison)) + return mauvais + + +def render(database, resultats, attente, colour=True): + """Le rapport, section par section, dans l'ordre déclaré.""" + entete = ( + t("a development copy") if attente == COPY else t("a live instance") + ) + lignes = [ + f"🩺 {t('State of')} {database} — {t('read as')} {entete}", + "", + ] + for section in SECTIONS: + corps = [] + for controle in CONTROLES: + if controle["section"] != section: + continue + valeur = resultats.get(controle["key"]) + genre, gravite, raison = verdict(controle, valeur, attente) + nombre = "?" if isinstance(valeur, dict) else str(valeur) + icone = { + "ok": "✅", + "bad": "❌" if gravite == "broken" else "⚠", + "info": "ℹ️ ", + "skip": "·", + "unreadable": "❔", + }[genre] + corps.append( + paint( + f" {icone} {nombre.rjust(5)} {t(controle['title'])}", + gravite if genre in ("bad", "ok") else "dim", + colour, + ) + ) + if raison: + corps.append(paint(f" {raison}", "dim", colour)) + if corps: + lignes.append(t(section)) + lignes.extend(corps) + lignes.append("") + return "\n".join(lignes).rstrip() + + +def main(argv=None): + import argparse + import json + + parser = argparse.ArgumentParser( + description=t("What state an instance is in, for its intended use."), + ) + parser.add_argument("-d", "--database", required=True) + parser.add_argument("-c", "--config", help="odoo config file") + parser.add_argument( + "--expect", + choices=ATTENTES, + default=COPY, + help="copy: a restored development copy. live: a running instance.", + ) + parser.add_argument("--json", action="store_true") + parser.add_argument("--no-color", action="store_true") + args = parser.parse_args(argv) + + 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 + + resultats = inspect(args.database, args.config) + mauvais = findings(resultats, args.expect) + + if args.json: + print( + json.dumps( + { + "database": args.database, + "expect": args.expect, + "checks": resultats, + "bad": [c["key"] for c, _, _ in mauvais], + }, + indent=2, + ensure_ascii=False, + ) + ) + else: + colour = sys.stdout.isatty() and not args.no_color + print(render(args.database, resultats, args.expect, colour)) + return 1 if mauvais else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/script/analyse/monitoring.py b/script/analyse/monitoring.py index 1f8afa6..cba79aa 100644 --- a/script/analyse/monitoring.py +++ b/script/analyse/monitoring.py @@ -71,6 +71,20 @@ ANALYSES = ( "needs_sql": "It reads pg_catalog — indexes and real tables — which" " no RPC session exposes.", }, + { + "key": "instance_state", + "title": "State of the instance", + "why": "Neutralisation, scheduler, backups, queues and who is an" + " 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," + " 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": "cow_views", "title": "Customised views, website copies included", diff --git a/script/todo/todo.py b/script/todo/todo.py index 1c4e651..0249191 100755 --- a/script/todo/todo.py +++ b/script/todo/todo.py @@ -3553,7 +3553,34 @@ class TODO( if kind not in analyse["kinds"]: print(f"✖ {t('Not available for this source.')}") return - monitoring.run_analysis(analyse, target) + extra = None + if analyse.get("asks_expect"): + extra = ["--expect", self._monitoring_expect(kind)] + monitoring.run_analysis(analyse, target, extra=extra) + + def _monitoring_expect(self, kind): + """Copie de développement, ou instance en service ? + + Zéro cron actif est le SUCCÈS attendu d'une copie et une panne + totale sur une production ; zéro serveur de courriel rassure sur + l'une et condamne l'autre. Deviner à la place de l'utilisateur, + c'est afficher du rouge sur ce qu'il vient de demander — et l'on + cesse alors de lire le rapport. + """ + from script.analyse import check_instance_state, monitoring + + if kind == monitoring.KIND_LIVE: + return check_instance_state.LIVE + print() + print(f"[1] {t('A development copy (restored, neutralised)')}") + print(f"[2] {t('An instance in service')}") + answer = click.prompt(t("Command:")) + print() + return ( + check_instance_state.LIVE + if answer == "2" + else check_instance_state.COPY + ) def _monitoring_select_source(self): """(genre, cible), ou None si l'on renonce. @@ -3565,17 +3592,25 @@ class TODO( from script.analyse import monitoring print() - print(f"[1] {t('A local backup .zip')}") - print(f"[2] {t('A remote backup (https + master password)')}") - print(f"[3] {t('A live remote instance')}") + # La base locale d'abord : c'est la provenance la plus directe, et + # `_analyse_select_source` du même menu range déjà la base avant la + # sauvegarde. Deux ordres différents dans un même menu se paient en + # hésitation à chaque usage. + print(f"[1] {t('A local database')}") + print(f"[2] {t('A local backup .zip')}") + print(f"[3] {t('A remote backup (https + master password)')}") + print(f"[4] {t('A live remote instance')}") print(f"[0] {t('Back')}") answer = click.prompt(t("Command:")) print() if answer == "1": + database = self.db_manager.select_database() + return (monitoring.KIND_DATABASE, database) if database else None + if answer == "2": path = self.db_manager.select_backup_path() database = self._monitoring_restore(path) if path else None return (monitoring.KIND_DATABASE, database) if database else None - if answer == "2": + if answer == "3": status, path, _name = ( self.db_manager.download_database_backup_cli() ) @@ -3584,7 +3619,7 @@ class TODO( return None database = self._monitoring_restore(path) return (monitoring.KIND_DATABASE, database) if database else None - if answer == "3": + if answer == "4": return self._monitoring_live() return None diff --git a/script/todo/todo_i18n.py b/script/todo/todo_i18n.py index 3046315..51c76b0 100644 --- a/script/todo/todo_i18n.py +++ b/script/todo/todo_i18n.py @@ -9584,6 +9584,30 @@ TRANSLATIONS = { "fr": "Odoo lève un KeyError rien qu'en les contrôlant, et plus rien ne les lira jamais.", "en": "Odoo raises a KeyError merely checking them, and nothing will ever read them again.", }, + "A local database": { + "fr": "Une base de données locale", + "en": "A local database", + }, + "State of the instance": { + "fr": "État de l'instance", + "en": "State of the instance", + }, + "Neutralisation, scheduler, backups, queues and who is an administrator — read for the use you intend.": { + "fr": "Neutralisation, cadenceur, sauvegardes, files et qui est administrateur — lu pour l'usage que vous en faites.", + "en": "Neutralisation, scheduler, backups, queues and who is an administrator — read for the use you intend.", + }, + "Several checks read tables no RPC session exposes, and the lateness of a job is computed in SQL.": { + "fr": "Plusieurs contrôles lisent des tables qu'aucune session RPC n'expose, et le retard d'une tâche se calcule en SQL.", + "en": "Several checks read tables no RPC session exposes, and the lateness of a job is computed in SQL.", + }, + "A development copy (restored, neutralised)": { + "fr": "Une copie de développement (restaurée, neutralisée)", + "en": "A development copy (restored, neutralised)", + }, + "An instance in service": { + "fr": "Une instance en service", + "en": "An instance in service", + }, } diff --git a/test/test_check_instance_state.py b/test/test_check_instance_state.py new file mode 100644 index 0000000..4166708 --- /dev/null +++ b/test/test_check_instance_state.py @@ -0,0 +1,221 @@ +#!/usr/bin/env python3 +# © 2021-2026 TechnoLibre (http://www.technolibre.ca) +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl) + +"""La polarité : le même chiffre, deux lectures opposées. + +Zéro cron actif est le succès attendu d'une copie de développement et une +panne totale sur une production. C'est la propriété centrale de cet outil, +et c'est aussi celle qu'une refonte casserait sans bruit : il suffirait +qu'un contrôle perde une de ses deux politiques pour qu'il se taise — +sans erreur, sans rouge, en donnant l'impression d'avoir été vérifié. + +Le second sujet de ce fichier est le secret. Une clé de paiement VIVANTE +a été mesurée dans une base de test. Un rapport finit dans un billet ou +devant un agent : aucune requête ne doit lire la valeur d'un secret, et +c'est vérifié sur le texte des requêtes, pas sur l'intention. +""" + +import os +import sys +import unittest + +sys.path.insert( + 0, os.path.normpath(os.path.join(os.path.dirname(__file__), "..")) +) + +from script.analyse import check_instance_state as etat # noqa: E402 + + +class TestEveryCheckDeclaresBothReadings(unittest.TestCase): + """Un contrôle sans politique se tairait — sans erreur ni rouge.""" + + def test_each_check_has_a_policy_for_each_expectation(self): + for controle in etat.CONTROLES: + for attente in etat.ATTENTES: + self.assertIn(attente, controle, controle["key"]) + self.assertTrue(controle[attente], controle["key"]) + + def test_each_policy_is_one_we_know_how_to_read(self): + connus = ("zero", "nonzero", "info", "skip") + for controle in etat.CONTROLES: + for attente in etat.ATTENTES: + self.assertIn(controle[attente][0], connus, controle["key"]) + + def test_a_judging_policy_carries_a_gravity_we_understand(self): + for controle in etat.CONTROLES: + for attente in etat.ATTENTES: + politique = controle[attente] + if politique[0] in ("zero", "nonzero"): + self.assertIn(politique[1], ("broken", "watch")) + + def test_a_judging_policy_explains_itself(self): + """Un rouge sans raison n'apprend rien à qui le lit.""" + for controle in etat.CONTROLES: + for attente in etat.ATTENTES: + if controle[attente][0] in ("zero", "nonzero"): + self.assertTrue( + controle.get(f"why_{attente}", "").strip(), + f"{controle['key']}/{attente}", + ) + + def test_a_skipped_check_says_why(self): + for controle in etat.CONTROLES: + for attente in etat.ATTENTES: + politique = controle[attente] + if politique[0] == "skip": + self.assertGreater(len(politique), 1, controle["key"]) + self.assertTrue(politique[1].strip()) + + def test_every_section_declared_is_actually_used(self): + utilisees = {c["section"] for c in etat.CONTROLES} + self.assertEqual(set(etat.SECTIONS), utilisees) + + def test_keys_are_unique(self): + cles = [c["key"] for c in etat.CONTROLES] + self.assertEqual(len(cles), len(set(cles))) + + +class TestThePolarityActuallyInverts(unittest.TestCase): + """La propriété centrale, épinglée sur un cas réel mesuré.""" + + def _controle(self, cle): + for controle in etat.CONTROLES: + if controle["key"] == cle: + return controle + raise AssertionError(cle) + + def test_active_crons_are_a_fault_on_a_copy_and_normal_when_live(self): + controle = self._controle("cron_active") + copie = etat.verdict(controle, 35, etat.COPY) + vivante = etat.verdict(controle, 35, etat.LIVE) + self.assertEqual(copie[0], "bad") + self.assertEqual(vivante[0], "ok") + + def test_no_cron_at_all_is_the_opposite(self): + controle = self._controle("cron_active") + self.assertEqual(etat.verdict(controle, 0, etat.COPY)[0], "ok") + self.assertEqual(etat.verdict(controle, 0, etat.LIVE)[0], "bad") + + def test_late_jobs_are_not_judged_on_a_copy(self): + """Mesuré : 11 en retard sur la base d'ORIGINE, jamais démarrée.""" + controle = self._controle("cron_late") + genre, _, raison = etat.verdict(controle, 11, etat.COPY) + self.assertEqual(genre, "skip") + self.assertTrue(raison) + self.assertEqual(etat.verdict(controle, 11, etat.LIVE)[0], "bad") + + def test_missing_backups_are_not_judged_on_a_copy(self): + """update_prod_to_dev les efface : leur absence ne prouve rien.""" + controle = self._controle("backup_rows") + self.assertEqual(etat.verdict(controle, 0, etat.COPY)[0], "skip") + self.assertEqual(etat.verdict(controle, 0, etat.LIVE)[0], "bad") + + def test_having_no_mail_server_warns_under_both_readings(self): + """Zéro serveur NE prouve PAS la sûreté : Odoo retombe sur la conf. + + C'est pourquoi le `neutralize.sql` d'Odoo INSÈRE un serveur bouchon + `invalid:1025` au lieu de tout supprimer, avec le commentaire + « prevent using fallback servers ». Notre `disable_mail_server`, lui, + fait `unlink()` — il rouvre la porte qu'Odoo ferme. + """ + controle = self._controle("mail_server_total") + for attente in etat.ATTENTES: + self.assertEqual(etat.verdict(controle, 0, attente)[0], "bad") + + def test_a_check_without_a_policy_is_skipped_never_approved(self): + """L'oubli doit se voir comme un trou, pas comme un feu vert.""" + orphelin = {"key": "x", "section": "Users", "title": "x", "sql": ""} + for attente in etat.ATTENTES: + self.assertEqual(etat.verdict(orphelin, 99, attente)[0], "skip") + + def test_an_unreadable_check_is_not_a_zero(self): + controle = self._controle("cron_active") + genre, _, raison = etat.verdict( + controle, {"error": "relation absente"}, etat.COPY + ) + self.assertEqual(genre, "unreadable") + self.assertIn("relation", raison) + + +class TestWhatTheReportSaysAndCounts(unittest.TestCase): + def _tout(self, valeur=0): + return {c["key"]: valeur for c in etat.CONTROLES} + + def test_the_header_names_the_reading(self): + for attente, mot in ( + (etat.COPY, "a development copy"), + (etat.LIVE, "a live instance"), + ): + texte = etat.render("b", self._tout(), attente, colour=False) + self.assertIn(etat.t(mot), texte) + + def test_only_faults_reach_the_exit_code(self): + resultats = self._tout() + resultats["cron_active"] = 35 # faute sous copy, normal sous live + self.assertTrue(etat.findings(resultats, etat.COPY)) + self.assertFalse( + [ + f + for f in etat.findings(resultats, etat.LIVE) + if f[0]["key"] == "cron_active" + ] + ) + + def test_a_skipped_check_shows_its_number_and_its_reason(self): + resultats = self._tout() + resultats["cron_late"] = 11 + texte = etat.render("b", resultats, etat.COPY, colour=False) + self.assertIn("11", texte) + self.assertIn( + etat.t( + "Nobody runs the scheduler on a restored copy:" + " measured 11 late on an untouched source" + " database." + ), + texte, + ) + + def test_an_unreadable_check_is_shown_as_unknown(self): + resultats = self._tout() + resultats["mail_stuck"] = {"error": "relation absente"} + texte = etat.render("b", resultats, etat.LIVE, colour=False) + self.assertIn("❔", texte) + + def test_sections_come_out_in_the_declared_order(self): + texte = etat.render("b", self._tout(), etat.COPY, colour=False) + positions = [texte.find(etat.t(s)) for s in etat.SECTIONS] + self.assertEqual(positions, sorted(positions)) + self.assertNotIn(-1, positions) + + +class TestNoQueryEverReadsASecret(unittest.TestCase): + """Une clé Stripe VIVANTE a été mesurée dans une base de test.""" + + SECRETS = ( + "secret_key", + "publishable_key", + "smtp_pass", + "password", + "sftp_password", + "database.secret", + "api_key", + ) + + def test_no_secret_column_is_selected(self): + for controle in etat.CONTROLES: + minuscule = controle["sql"].lower() + for secret in self.SECRETS: + self.assertNotIn(secret, minuscule, controle["key"]) + + def test_every_query_reads_and_returns_one_number(self): + for controle in etat.CONTROLES: + sql = " ".join(controle["sql"].split()) + self.assertTrue(sql.upper().startswith("SELECT"), controle["key"]) + majuscule = f" {sql.upper()} " + for interdit in ("INSERT", "UPDATE", "DELETE", "DROP", "ALTER"): + self.assertNotIn(f" {interdit} ", majuscule, controle["key"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_monitoring.py b/test/test_monitoring.py index 4d1d756..f40ef8c 100644 --- a/test/test_monitoring.py +++ b/test/test_monitoring.py @@ -300,3 +300,49 @@ class TestTheScreenAnswersKeys(unittest.TestCase): def test_q_gives_up(self): self.assertIsNone(self._presser(monitoring.KIND_DATABASE, ["q"])) + + +class TestTheSourceMenuIsWrittenTwice(unittest.TestCase): + """Les numéros affichés mènent-ils où ils le disent ? + + `_monitoring_select_source` imprime « [2] Une sauvegarde .zip » d'un + côté et teste `answer == "2"` de l'autre. Rien ne relie les deux : + insérer « une base locale » en tête décale tout le reste à la main. + C'est le piège exact que `MenuCoherence` garde pour les autres menus, + et celui-ci n'entre pas dans son moule — il s'imprime, il ne se + déclare pas. + """ + + def setUp(self): + import re + from pathlib import Path + + todo = ( + Path(__file__).resolve().parent.parent + / "script" + / "todo" + / "todo.py" + ) + source = todo.read_text(encoding="utf-8") + debut = source.index("def _monitoring_select_source(self):") + fin = source.index("def _monitoring_live(self):", debut) + self.corps = source[debut:fin] + self.affiches = re.findall(r'print\(f"\[(\d+)\] \{t\(', self.corps) + self.branches = re.findall(r'if answer == "(\d+)":', self.corps) + + def test_the_menu_was_actually_parsed(self): + """Sur des listes vides, tout passe : mieux vaut tomber ici.""" + self.assertGreaterEqual(len(self.affiches), 4) + + def test_every_shown_entry_has_a_branch(self): + montres = [n for n in self.affiches if n != "0"] + self.assertEqual(sorted(montres), sorted(self.branches)) + + def test_the_numbering_is_contiguous_from_one(self): + montres = sorted(int(n) for n in self.affiches if n != "0") + self.assertEqual(montres, list(range(1, len(montres) + 1))) + + def test_a_local_database_is_offered_first(self): + """La provenance la plus directe, et l'ordre du menu d'à côté.""" + self.assertEqual(self.affiches[0], "1") + self.assertIn("A local database", self.corps.split("if answer")[0])