68 lines
2.9 KiB
Python
68 lines
2.9 KiB
Python
"""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()
|