Mathieu Benoit
1e731114f5
Reflect the current year in all TechnoLibre license headers across script/, test/, and docker/. Generated by Claude Code 2.1.74 model claude-sonnet-4-6 Co-Authored-By: Mathieu Benoit <mathben@technolibre.ca>
146 lines
5.6 KiB
Python
146 lines
5.6 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)
|