Initialize clean CSR inventory repository

scripts/test_backup_scripts.py

@@ -0,0 +1,67 @@
+"""Failure-path checks for backup/restore; no database or Docker required."""
+import gzip
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+
+@unittest.skipUnless(shutil.which("sh") and shutil.which("gzip"), "POSIX shell and gzip required")
+class BackupScriptTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name)
+        self.scripts = Path(__file__).resolve().parent
+        docker = self.root / "docker"
+        docker.write_text('''#!/bin/sh
+if [ "${FAIL_DOCKER:-0}" = 1 ]; then exit 9; fi
+case "$*" in
+  *pg_dump*) printf 'SELECT 1;\\n' ;;
+  *psql*) cat > "$RESTORE_CAPTURE" ;;
+esac
+''', encoding="utf-8", newline="\n")
+        docker.chmod(0o700)
+        self.env = {**os.environ, "PATH": str(self.root) + os.pathsep + os.environ["PATH"],
+                    "RESTORE_CAPTURE": str(self.root / "restored.sql")}
+
+    def run_script(self, name, *args, **env):
+        return subprocess.run(["sh", str(self.scripts / name), *map(str, args)],
+                              cwd=self.root, env={**self.env, **env}, capture_output=True, text=True)
+
+    def test_dump_failure_does_not_publish_backup_or_report_success(self):
+        result = self.run_script("backup.sh", FAIL_DOCKER="1")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertNotIn("Backup erstellt", result.stdout)
+        self.assertEqual(list((self.root / "backups").iterdir()), [])
+
+    def test_success_round_trip_and_env_is_not_executed(self):
+        (self.root / ".env").write_text("touch env_was_executed\n", encoding="utf-8")
+        result = self.run_script("backup.sh")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertFalse((self.root / "env_was_executed").exists())
+        backup, = (self.root / "backups").glob("*.sql.gz")
+        self.assertEqual(gzip.decompress(backup.read_bytes()), b"SELECT 1;\n")
+        result = self.run_script("restore.sh", backup)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual((self.root / "restored.sql").read_text(), "SELECT 1;\n")
+
+    def test_corrupt_backup_never_reaches_database(self):
+        corrupt = self.root / "corrupt.sql.gz"
+        corrupt.write_bytes(gzip.compress(b"SELECT 1;\n")[:-6])
+        result = self.run_script("restore.sh", corrupt)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertFalse((self.root / "restored.sql").exists())
+
+    def test_restore_error_is_not_reported_as_success(self):
+        backup = self.root / "valid.sql.gz"
+        backup.write_bytes(gzip.compress(b"SELECT 1;\n"))
+        result = self.run_script("restore.sh", backup, FAIL_DOCKER="1")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertNotIn("abgeschlossen", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()