Mathieu Benoit
48a1d5ce20
Un invité sans magasin de confiance ne peut RIEN recevoir, et sur un hôte Proxmox c'est l'HÔTE qu'il faut excepter : imbriqué, l'invité sort masqué derrière lui et le pont ne voit jamais sa propre adresse. Mesuré depuis l'invité : code 000 et vérification SSL 19, puis 200 et 0. Sans cela le gestionnaire de paquets se rabattait sur 564 dérivations à construire. Le journal par VM ne portait AUCUN message du menu : la cause vivait sur une console qui défile, et le fichier qu'on rouvre après l'échec n'avait que le symptôme. Et « db_drop_all » annonçait détruites des bases qui ne l'étaient pas, son code de retour jeté. --- EN --- A guest with no trust store can receive NOTHING, and on a Proxmox host it is the HOST that must be exempted: nested, the guest leaves masqueraded behind it and the bridge never sees its own address. Measured from the guest: code 000 and SSL verification 19, then 200 and 0. Without it the package manager fell back to building 564 derivations. The per-VM log carried NO message from the menu: the cause lived on a console that scrolls away, and the file reopened after failure held only the symptom. And "db_drop_all" announced databases as dropped that were not, its exit status discarded. Assisted-by: Claude Opus 5
219 lines
8.1 KiB
Python
219 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
# © 2026 TechnoLibre (http://www.technolibre.ca)
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
|
|
|
|
import csv
|
|
import io
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from unittest.mock import patch
|
|
|
|
from script.database.migrate.process_backup_file import process_zip
|
|
|
|
|
|
class TestProcessZip(unittest.TestCase):
|
|
"""Test ZIP file processing to remove lines containing a keyword."""
|
|
|
|
def _create_zip(self, path, filename, content):
|
|
with zipfile.ZipFile(path, "w") as zf:
|
|
zf.writestr(filename, content)
|
|
|
|
def test_removes_matching_lines(self):
|
|
tmpdir = tempfile.mkdtemp()
|
|
input_zip = os.path.join(tmpdir, "input.zip")
|
|
output_zip = os.path.join(tmpdir, "output.zip")
|
|
self._create_zip(
|
|
input_zip,
|
|
"dump.sql",
|
|
"keep this line\ndelete secret line\nkeep also\n",
|
|
)
|
|
process_zip(input_zip, output_zip, "secret", "dump.sql")
|
|
with zipfile.ZipFile(output_zip) as zf:
|
|
content = zf.read("dump.sql").decode()
|
|
self.assertIn("keep this line", content)
|
|
self.assertIn("keep also", content)
|
|
self.assertNotIn("secret", content)
|
|
|
|
def test_preserves_all_when_no_match(self):
|
|
tmpdir = tempfile.mkdtemp()
|
|
input_zip = os.path.join(tmpdir, "input.zip")
|
|
output_zip = os.path.join(tmpdir, "output.zip")
|
|
original = "line1\nline2\nline3\n"
|
|
self._create_zip(input_zip, "data.sql", original)
|
|
process_zip(input_zip, output_zip, "nomatch", "data.sql")
|
|
with zipfile.ZipFile(output_zip) as zf:
|
|
content = zf.read("data.sql").decode()
|
|
self.assertIn("line1", content)
|
|
self.assertIn("line2", content)
|
|
self.assertIn("line3", content)
|
|
|
|
def test_removes_all_matching(self):
|
|
tmpdir = tempfile.mkdtemp()
|
|
input_zip = os.path.join(tmpdir, "input.zip")
|
|
output_zip = os.path.join(tmpdir, "output.zip")
|
|
self._create_zip(
|
|
input_zip,
|
|
"dump.sql",
|
|
"bad line\nbad again\nbad too\n",
|
|
)
|
|
process_zip(input_zip, output_zip, "bad", "dump.sql")
|
|
with zipfile.ZipFile(output_zip) as zf:
|
|
content = zf.read("dump.sql").decode()
|
|
self.assertEqual(content.strip(), "")
|
|
|
|
def test_other_files_untouched(self):
|
|
tmpdir = tempfile.mkdtemp()
|
|
input_zip = os.path.join(tmpdir, "input.zip")
|
|
output_zip = os.path.join(tmpdir, "output.zip")
|
|
with zipfile.ZipFile(input_zip, "w") as zf:
|
|
zf.writestr("target.sql", "keep\nremove secret\n")
|
|
zf.writestr("other.txt", "secret stays here\n")
|
|
process_zip(input_zip, output_zip, "secret", "target.sql")
|
|
with zipfile.ZipFile(output_zip) as zf:
|
|
target = zf.read("target.sql").decode()
|
|
other = zf.read("other.txt").decode()
|
|
self.assertNotIn("secret", target)
|
|
self.assertIn("secret", other)
|
|
|
|
|
|
class TestCompareDatabaseApplicationLogic(unittest.TestCase):
|
|
"""Test CSV set comparison logic used by compare_database_application."""
|
|
|
|
def _write_csv(self, path, rows, fieldnames):
|
|
with open(path, "w", newline="") as f:
|
|
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
def test_identical_csvs(self):
|
|
tmpdir = tempfile.mkdtemp()
|
|
csv1 = os.path.join(tmpdir, "a.csv")
|
|
csv2 = os.path.join(tmpdir, "b.csv")
|
|
rows = [{"name": "mod1"}, {"name": "mod2"}]
|
|
self._write_csv(csv1, rows, ["name"])
|
|
self._write_csv(csv2, rows, ["name"])
|
|
with open(csv1) as f1, open(csv2) as f2:
|
|
r1 = csv.DictReader(f1)
|
|
r2 = csv.DictReader(f2)
|
|
s1 = {a["name"] for a in r1}
|
|
s2 = {a["name"] for a in r2}
|
|
self.assertEqual(s1, s2)
|
|
self.assertEqual(len(s1.difference(s2)), 0)
|
|
|
|
def test_different_csvs(self):
|
|
tmpdir = tempfile.mkdtemp()
|
|
csv1 = os.path.join(tmpdir, "a.csv")
|
|
csv2 = os.path.join(tmpdir, "b.csv")
|
|
self._write_csv(csv1, [{"name": "mod1"}, {"name": "mod2"}], ["name"])
|
|
self._write_csv(csv2, [{"name": "mod2"}, {"name": "mod3"}], ["name"])
|
|
with open(csv1) as f1, open(csv2) as f2:
|
|
r1 = csv.DictReader(f1)
|
|
r2 = csv.DictReader(f2)
|
|
s1 = {a["name"] for a in r1}
|
|
s2 = {a["name"] for a in r2}
|
|
self.assertEqual(s1.intersection(s2), {"mod2"})
|
|
self.assertEqual(s1.difference(s2), {"mod1"})
|
|
self.assertEqual(s2.difference(s1), {"mod3"})
|
|
|
|
def test_empty_csvs(self):
|
|
tmpdir = tempfile.mkdtemp()
|
|
csv1 = os.path.join(tmpdir, "a.csv")
|
|
csv2 = os.path.join(tmpdir, "b.csv")
|
|
self._write_csv(csv1, [], ["name"])
|
|
self._write_csv(csv2, [], ["name"])
|
|
with open(csv1) as f1, open(csv2) as f2:
|
|
r1 = csv.DictReader(f1)
|
|
r2 = csv.DictReader(f2)
|
|
s1 = {a["name"] for a in r1}
|
|
s2 = {a["name"] for a in r2}
|
|
self.assertEqual(len(s1.union(s2)), 0)
|
|
|
|
def test_one_empty_csv(self):
|
|
tmpdir = tempfile.mkdtemp()
|
|
csv1 = os.path.join(tmpdir, "a.csv")
|
|
csv2 = os.path.join(tmpdir, "b.csv")
|
|
self._write_csv(csv1, [{"name": "mod1"}], ["name"])
|
|
self._write_csv(csv2, [], ["name"])
|
|
with open(csv1) as f1, open(csv2) as f2:
|
|
r1 = csv.DictReader(f1)
|
|
r2 = csv.DictReader(f2)
|
|
s1 = {a["name"] for a in r1}
|
|
s2 = {a["name"] for a in r2}
|
|
self.assertEqual(s1.difference(s2), {"mod1"})
|
|
self.assertEqual(len(s2.difference(s1)), 0)
|
|
|
|
|
|
class UneDestructionNAnnonceQueCeQuElleAFait(unittest.TestCase):
|
|
"""« make db_drop_all » composait une commande « parallel », jetait son
|
|
code de retour et imprimait la liste des bases comme détruites.
|
|
|
|
Le cas s'atteint dès que « parallel » manque du PATH : le shell rend 127,
|
|
pas une base n'est touchée, et l'opérateur passe à la suite en croyant
|
|
ses bases parties. Une destruction qui annonce un succès qu'elle n'a pas
|
|
obtenu est pire que celle qui échoue.
|
|
"""
|
|
|
|
def _module(self):
|
|
import importlib.util
|
|
|
|
chemin = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
|
"script/database/db_drop_all.py",
|
|
)
|
|
spec = importlib.util.spec_from_file_location("db_drop_all", chemin)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
def _courir(self, code_destruction):
|
|
import contextlib
|
|
|
|
mod = self._module()
|
|
|
|
def faux_shell(cmd):
|
|
if "--list" in cmd:
|
|
return 0, "test_alpha\ntest_beta"
|
|
return code_destruction, (
|
|
"" if not code_destruction else "parallel: command not found"
|
|
)
|
|
|
|
mod.execute_shell = faux_shell
|
|
|
|
class Config:
|
|
database = ""
|
|
test_only = True
|
|
|
|
mod.get_config = lambda: Config()
|
|
sortie, erreur = io.StringIO(), io.StringIO()
|
|
with contextlib.redirect_stdout(sortie):
|
|
with contextlib.redirect_stderr(erreur):
|
|
code = mod.main()
|
|
return code, sortie.getvalue(), erreur.getvalue()
|
|
|
|
def test_a_failed_drop_is_not_announced_as_done(self):
|
|
code, sortie, erreur = self._courir(127)
|
|
self.assertEqual(127, code)
|
|
self.assertNotIn("Database deleted", sortie)
|
|
self.assertNotIn("test_alpha", sortie)
|
|
|
|
def test_the_cause_reaches_the_operator(self):
|
|
"""Le code de retour seul laisserait chercher : la sortie du shell
|
|
nomme ce qui manque."""
|
|
_code, _sortie, erreur = self._courir(127)
|
|
self.assertIn("NOT deleted", erreur)
|
|
self.assertIn("parallel", erreur)
|
|
|
|
def test_a_real_drop_is_still_announced(self):
|
|
"""Le cas ordinaire ne change pas : les bases détruites se disent."""
|
|
code, sortie, _erreur = self._courir(0)
|
|
self.assertEqual(0, code)
|
|
self.assertIn("Database deleted", sortie)
|
|
self.assertIn("test_alpha", sortie)
|
|
self.assertIn("test_beta", sortie)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|