erplibre/test/test_database_tools.py

#!/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()