Mathieu Benoit
ec29781b7a
Answering « all » reset one copy of two and reported success. Two defects behind it, both mine. A requested key matching no finding did nothing, silently: the tool returned early on « nothing drifted » and never looked. Detection is differential — it only sees a copy whose CHILD breaks — so a stale copy without children escapes it. A key can now be reset outside detection, an unknown one is named, and the exit code is 2. And the culprit is not always the parent. On /contactus the parent was identical to its module view; the CHILD held the stale arch. Both are proposed now, parent first — it is the more common case. Measured on the migration: 33 of 33 public URLs answer. --- FR --- [FIX] migration : réinitialiser la copie vraiment périmée, et signaler une clé sans objet Répondre « toutes » réinitialisait une copie sur deux et annonçait un succès. Deux défauts derrière, tous deux à moi. Une clé demandée ne correspondant à aucun constat ne faisait rien, en silence : l'outil sortait sur « rien n'a dérivé » sans jamais chercher. La détection est différentielle — elle ne voit qu'une copie dont un ENFANT casse — donc une copie périmée sans enfant lui échappe. Une clé peut désormais être réinitialisée hors détection, une clé inconnue est nommée, et le code de sortie vaut 2. Et le coupable n'est pas toujours le parent. Sur /contactus, le parent était identique à sa vue module ; c'est l'ENFANT qui portait l'arch périmée. Les deux sont proposés, le parent d'abord — le cas le plus fréquent. Mesuré sur la migration : 33 URL publiques sur 33 répondent. Assisted-by: Claude Opus 5
383 lines
15 KiB
Python
Executable file
383 lines
15 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
|
|
|
"""Une migration peut réussir et servir quand même des 500.
|
|
|
|
Mesuré sur une vraie migration 12 → 13 : tous les modules chargés, aucune
|
|
erreur au journal, et pourtant deux pages publiques sur trente-trois
|
|
répondaient 500 — un billet de blogue et /contactus. Rien ne les distinguait
|
|
avant de les demander.
|
|
|
|
La liste vient du sitemap, celle qu'Odoo publie pour les moteurs de
|
|
recherche. Elle ne se lit PAS depuis « odoo-bin shell » : enumerate_pages()
|
|
réclame http.root.get_db_router(request.db) et lève « object unbound » sans
|
|
requête réelle. D'où le serveur démarré, interrogé, arrêté.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.insert(0, os.path.join(REPO, "script", "odoo", "migration"))
|
|
|
|
import smoke_public_url as smoke # noqa: E402
|
|
|
|
|
|
class TestTheUrlAreBroughtHome(unittest.TestCase):
|
|
"""Le sitemap porte le domaine du site ; on teste une base locale."""
|
|
|
|
def test_the_host_is_replaced(self):
|
|
# Garder le domaine ferait interroger la PRODUCTION depuis une
|
|
# migration. C'est le genre d'erreur qui ne se voit qu'après.
|
|
url = smoke.local_url(
|
|
"http://127.0.0.1:8169", "https://technolibre.ca/blog/x-3/post/y-5"
|
|
)
|
|
self.assertEqual(url, "http://127.0.0.1:8169/blog/x-3/post/y-5")
|
|
|
|
def test_a_query_string_survives(self):
|
|
url = smoke.local_url("http://127.0.0.1:8169", "https://a.ca/p?x=1")
|
|
self.assertEqual(url, "http://127.0.0.1:8169/p?x=1")
|
|
|
|
def test_a_bare_domain_becomes_the_root(self):
|
|
url = smoke.local_url("http://127.0.0.1:8169", "https://a.ca")
|
|
self.assertEqual(url, "http://127.0.0.1:8169/")
|
|
|
|
def test_a_trailing_slash_on_the_base_does_not_double(self):
|
|
url = smoke.local_url("http://127.0.0.1:8169/", "https://a.ca/p")
|
|
self.assertEqual(url, "http://127.0.0.1:8169/p")
|
|
|
|
|
|
class TestReadingTheSitemap(unittest.TestCase):
|
|
def setUp(self):
|
|
self.answers = {}
|
|
self.original = smoke.fetch
|
|
smoke.fetch = lambda url, timeout=30: self.answers.get(url, (404, ""))
|
|
self.addCleanup(setattr, smoke, "fetch", self.original)
|
|
|
|
def test_a_plain_sitemap(self):
|
|
self.answers["http://h/sitemap.xml"] = (
|
|
200,
|
|
"<urlset><loc>https://a.ca/x</loc><loc>https://a.ca/y</loc></urlset>",
|
|
)
|
|
lst_url, status = smoke.sitemap_urls("http://h")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(lst_url, ["http://h/x", "http://h/y"])
|
|
|
|
def test_an_index_is_followed(self):
|
|
# Odoo découpe le sitemap au-delà d'un certain nombre de pages : ne
|
|
# pas descendre d'un cran ferait tester deux fichiers XML au lieu du
|
|
# site, et conclure « tout va bien ».
|
|
self.answers["http://h/sitemap.xml"] = (
|
|
200,
|
|
"<sitemapindex><loc>https://a.ca/sitemap-1-1.xml</loc>"
|
|
"</sitemapindex>",
|
|
)
|
|
self.answers["http://h/sitemap-1-1.xml"] = (
|
|
200,
|
|
"<urlset><loc>https://a.ca/deep</loc></urlset>",
|
|
)
|
|
lst_url, _status = smoke.sitemap_urls("http://h")
|
|
self.assertEqual(lst_url, ["http://h/deep"])
|
|
|
|
def test_duplicates_are_dropped(self):
|
|
self.answers["http://h/sitemap.xml"] = (
|
|
200,
|
|
"<urlset><loc>https://a.ca/x</loc><loc>https://a.ca/x</loc></urlset>",
|
|
)
|
|
lst_url, _status = smoke.sitemap_urls("http://h")
|
|
self.assertEqual(lst_url, ["http://h/x"])
|
|
|
|
def test_an_unreachable_sitemap_is_reported_not_taken_as_empty(self):
|
|
# Sans cela, un sitemap injoignable se lirait comme « aucune page à
|
|
# tester », c'est-à-dire comme un succès.
|
|
lst_url, status = smoke.sitemap_urls("http://h")
|
|
self.assertEqual(lst_url, [])
|
|
self.assertEqual(status, 404)
|
|
|
|
|
|
class TestWhatCountsAsAFailure(unittest.TestCase):
|
|
def setUp(self):
|
|
self.answers = {}
|
|
self.original = smoke.fetch
|
|
smoke.fetch = lambda url, timeout=30: self.answers.get(url, (200, ""))
|
|
self.addCleanup(setattr, smoke, "fetch", self.original)
|
|
|
|
def test_a_500_fails(self):
|
|
self.answers["http://h/bad"] = (500, "")
|
|
self.assertEqual(
|
|
smoke.check_urls(["http://h/bad"]), [("http://h/bad", 500, [])]
|
|
)
|
|
|
|
def test_a_404_fails_too(self):
|
|
# Le sitemap est la liste PUBLIQUE : une page qui y figure et répond
|
|
# 404 est aussi fausse qu'un 500, juste plus discrète.
|
|
self.answers["http://h/gone"] = (404, "")
|
|
self.assertEqual(len(smoke.check_urls(["http://h/gone"])), 1)
|
|
|
|
def test_no_answer_at_all_fails(self):
|
|
self.answers["http://h/dead"] = (0, "")
|
|
self.assertEqual(
|
|
smoke.check_urls(["http://h/dead"]), [("http://h/dead", 0, [])]
|
|
)
|
|
|
|
def test_a_200_passes(self):
|
|
self.assertEqual(smoke.check_urls(["http://h/ok"]), [])
|
|
|
|
|
|
class TestTheReport(unittest.TestCase):
|
|
def setUp(self):
|
|
from script.todo import todo_i18n
|
|
|
|
self.addCleanup(
|
|
setattr, todo_i18n, "_current_lang", todo_i18n._current_lang
|
|
)
|
|
todo_i18n._current_lang = "en"
|
|
|
|
def test_all_green_says_how_many(self):
|
|
text = smoke.render(["a", "b"], [])
|
|
self.assertIn("2", text)
|
|
self.assertIn("✅", text)
|
|
|
|
def test_a_failure_shows_the_status_and_the_url(self):
|
|
text = smoke.render(["a"], [("http://h/blog/x", 500, [])])
|
|
self.assertIn("500", text)
|
|
self.assertIn("http://h/blog/x", text)
|
|
|
|
def test_an_empty_sitemap_is_not_a_success(self):
|
|
# « rien testé » ne doit pas se lire « rien de cassé ».
|
|
text = smoke.render([], [])
|
|
self.assertIn("⚠️", text)
|
|
|
|
|
|
class TestTheServerIsAlwaysStopped(unittest.TestCase):
|
|
"""Un serveur oublié tient le port et fait échouer le palier suivant."""
|
|
|
|
def test_it_is_terminated_even_when_the_sitemap_fails(self):
|
|
stopped = []
|
|
|
|
class FakeServer:
|
|
def terminate(self):
|
|
stopped.append("terminate")
|
|
|
|
def wait(self, timeout=None):
|
|
return 0
|
|
|
|
def kill(self):
|
|
stopped.append("kill")
|
|
|
|
original_start = smoke.start_server
|
|
original_wait = smoke.wait_ready
|
|
original_stop = smoke.stop_server
|
|
original_port = smoke.port_is_taken
|
|
smoke.start_server = (
|
|
lambda db, port, cfg="./config.conf", log_path=None: FakeServer()
|
|
)
|
|
smoke.wait_ready = lambda base, timeout=180, sleep=2: False
|
|
smoke.stop_server = lambda server: stopped.append("terminate")
|
|
smoke.port_is_taken = lambda port, host="127.0.0.1": False
|
|
self.addCleanup(setattr, smoke, "start_server", original_start)
|
|
self.addCleanup(setattr, smoke, "wait_ready", original_wait)
|
|
self.addCleanup(setattr, smoke, "stop_server", original_stop)
|
|
self.addCleanup(setattr, smoke, "port_is_taken", original_port)
|
|
with self.assertRaises(RuntimeError):
|
|
smoke.run("db", 8169, "./config.conf")
|
|
self.assertEqual(stopped, ["terminate"])
|
|
|
|
def test_a_distinct_port_by_default(self):
|
|
# 8069 est souvent pris par l'instance de travail : le lui voler
|
|
# ferait échouer le test pour une raison sans rapport.
|
|
self.assertNotEqual(smoke.DEFAULT_PORT, 8069)
|
|
|
|
|
|
class TestTheCulpritViewsAreNamed(unittest.TestCase):
|
|
"""Détecter sans nommer laisse relever des ids dans une trace à la main."""
|
|
|
|
def test_the_parent_is_read_from_the_error_context(self):
|
|
# C'est le PARENT le coupable : la copie figée dans laquelle
|
|
# l'enfant ne trouve plus son xpath.
|
|
line = "[view_id: 3282, xml_id: n/a, model: n/a, parent_id: 3281]"
|
|
self.assertEqual(smoke.RE_CONTEXT.findall(line), [("3282", "3281")])
|
|
|
|
def test_the_context_line_is_not_translated(self):
|
|
# La phrase qui précède l'est, celle-ci non : la lire marche donc
|
|
# sur un système français comme anglais.
|
|
self.assertEqual(
|
|
smoke.RE_CONTEXT.findall("[view_id: 1, parent_id: 2]"),
|
|
[("1", "2")],
|
|
)
|
|
|
|
def test_a_late_context_is_still_attached(self):
|
|
# Odoo vide son tampon à l'arrêt : le journal se lit APRÈS, et rien
|
|
# ne doit dépendre du moment où la ligne est apparue.
|
|
lst_failure = [("http://h/a", 500, [])]
|
|
log = ["[view_id: 3288, model: n/a, parent_id: 2841]"]
|
|
rebuilt = smoke.attach_missing_parents(lst_failure, log)
|
|
self.assertIn("2841", rebuilt[0][2])
|
|
|
|
def test_both_the_parent_and_the_child_are_proposed(self):
|
|
# LE défaut mesuré sur /contactus : le parent (3281) était IDENTIQUE
|
|
# à sa vue module, et c'est l'enfant (3282) qui portait l'arch
|
|
# périmée. Ne nommer que le parent envoyait réinitialiser une copie
|
|
# qui allait déjà bien, et la page restait en 500.
|
|
lst_failure = [("http://h/contactus", 500, [])]
|
|
log = ["[view_id: 3282, model: n/a, parent_id: 3281]"]
|
|
rebuilt = smoke.attach_missing_parents(lst_failure, log)
|
|
self.assertEqual(rebuilt[0][2], ["3281", "3282"])
|
|
|
|
def test_the_parent_comes_first(self):
|
|
# C'est le cas le plus fréquent — le blogue — donc en tête de liste.
|
|
lst_failure = [("http://h/a", 500, [])]
|
|
log = ["[view_id: 3288, model: n/a, parent_id: 2841]"]
|
|
rebuilt = smoke.attach_missing_parents(lst_failure, log)
|
|
self.assertEqual(rebuilt[0][2][0], "2841")
|
|
|
|
def test_an_already_attributed_id_is_not_duplicated(self):
|
|
lst_failure = [("http://h/a", 500, ["2841"])]
|
|
log = ["[view_id: 3288, model: n/a, parent_id: 2841]"]
|
|
rebuilt = smoke.attach_missing_parents(lst_failure, log)
|
|
self.assertEqual(rebuilt[0][2].count("2841"), 1)
|
|
|
|
|
|
class TestTheServerIsKilledForReal(unittest.TestCase):
|
|
"""« ./run.sh » est un script bash : il ne transmet pas les signaux."""
|
|
|
|
def test_it_starts_its_own_process_group(self):
|
|
# Sans cela, terminate() tue l'enveloppe et laisse odoo-bin vivant.
|
|
# Mesuré : six serveurs orphelins, un par essai, et les essais
|
|
# suivants interrogeaient sans le savoir celui d'avant.
|
|
import inspect
|
|
|
|
source = inspect.getsource(smoke.start_server)
|
|
self.assertIn("start_new_session=True", source)
|
|
|
|
def test_it_kills_the_group_not_just_the_wrapper(self):
|
|
import inspect
|
|
|
|
source = inspect.getsource(smoke.stop_server)
|
|
self.assertIn("killpg", source)
|
|
|
|
def test_a_taken_port_is_refused_not_tested(self):
|
|
# Un serveur resté d'un essai précédent répond « prêt » : on
|
|
# testerait sa base à lui en croyant tester la sienne.
|
|
original = smoke.port_is_taken
|
|
smoke.port_is_taken = lambda port, host="127.0.0.1": True
|
|
self.addCleanup(setattr, smoke, "port_is_taken", original)
|
|
with self.assertRaises(RuntimeError) as caught:
|
|
smoke.run("db", 8169, "./config.conf")
|
|
self.assertIn("8169", str(caught.exception))
|
|
|
|
|
|
class TestOfferingTheFix(unittest.TestCase):
|
|
def setUp(self):
|
|
from script.todo import todo_i18n
|
|
|
|
self.addCleanup(
|
|
setattr, todo_i18n, "_current_lang", todo_i18n._current_lang
|
|
)
|
|
todo_i18n._current_lang = "en"
|
|
self.applied = []
|
|
original = smoke.apply_reset
|
|
smoke.apply_reset = lambda db, keys: (
|
|
self.applied.append(keys),
|
|
(0, "ok"),
|
|
)[1]
|
|
self.addCleanup(setattr, smoke, "apply_reset", original)
|
|
|
|
def run_prompt(self, answer, lst_key=None):
|
|
import contextlib
|
|
import io
|
|
|
|
if lst_key is None:
|
|
lst_key = ["website_blog.blog_post_complete", "website_form.x"]
|
|
out = io.StringIO()
|
|
with contextlib.redirect_stdout(out):
|
|
done = smoke.prompt(
|
|
"db",
|
|
[("http://h/a", 500, ["2841"])],
|
|
lst_key,
|
|
ask=lambda prompt: answer,
|
|
)
|
|
return done, out.getvalue()
|
|
|
|
def test_every_key_is_numbered(self):
|
|
_done, text = self.run_prompt("")
|
|
self.assertIn("[1] website_blog.blog_post_complete", text)
|
|
self.assertIn("[a]", text)
|
|
|
|
def test_a_number_applies_only_that_one(self):
|
|
done, _text = self.run_prompt("1")
|
|
self.assertEqual(done, ["website_blog.blog_post_complete"])
|
|
self.assertEqual(self.applied, [["website_blog.blog_post_complete"]])
|
|
|
|
def test_a_applies_them_all(self):
|
|
done, _text = self.run_prompt("a")
|
|
self.assertEqual(len(done), 2)
|
|
|
|
def test_empty_applies_nothing(self):
|
|
done, text = self.run_prompt("")
|
|
self.assertEqual(done, [])
|
|
self.assertEqual(self.applied, [])
|
|
self.assertIn("Kept", text)
|
|
|
|
def test_an_unknown_answer_applies_nothing(self):
|
|
done, text = self.run_prompt("9")
|
|
self.assertEqual(self.applied, [])
|
|
self.assertIn("Unknown choice", text)
|
|
|
|
def test_no_key_found_says_so(self):
|
|
done, text = self.run_prompt("a", lst_key=[])
|
|
self.assertEqual(done, [])
|
|
self.assertIn("nothing to offer", text)
|
|
|
|
|
|
class TestTheMigrationOffersIt(unittest.TestCase):
|
|
def setUp(self):
|
|
from script.todo import todo_i18n
|
|
|
|
self.addCleanup(
|
|
setattr, todo_i18n, "_current_lang", todo_i18n._current_lang
|
|
)
|
|
todo_i18n._current_lang = "en"
|
|
|
|
def run_prompt(self, answer):
|
|
from script.todo.todo_upgrade import TodoUpgrade
|
|
|
|
upgrade = TodoUpgrade.__new__(TodoUpgrade)
|
|
upgrade.dct_progression = {}
|
|
upgrade.lst_command_executed = []
|
|
upgrade.write_config = lambda: None
|
|
lst_cmd = []
|
|
upgrade.run_on_terminal = lambda cmd: lst_cmd.append(cmd) or 0
|
|
upgrade.ask_gate = lambda prompt: answer
|
|
upgrade.prompt_smoke_public_url("db_upgrade_13")
|
|
return lst_cmd
|
|
|
|
def test_the_default_runs_nothing(self):
|
|
# Cela démarre un serveur et peut durer : pas à chaque palier sans
|
|
# qu'on l'ait demandé.
|
|
self.assertEqual(self.run_prompt(""), [])
|
|
|
|
def test_yes_runs_it_on_the_upgraded_database(self):
|
|
lst_cmd = self.run_prompt("y")
|
|
self.assertEqual(len(lst_cmd), 1)
|
|
self.assertIn("smoke_public_url.py", lst_cmd[0])
|
|
self.assertIn("-d db_upgrade_13", lst_cmd[0])
|
|
|
|
def test_it_is_asked_before_the_selenium_prompt(self):
|
|
# Après, la question arriverait une fois le navigateur ouvert : on
|
|
# aurait déjà cherché à la main ce que le test nomme.
|
|
import inspect
|
|
|
|
from script.todo.todo_upgrade import TodoUpgrade
|
|
|
|
source = inspect.getsource(TodoUpgrade.execute_odoo_upgrade)
|
|
self.assertLess(
|
|
source.index("prompt_smoke_public_url"),
|
|
source.index("Open the server with Selenium"),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|