82 lines
3.7 KiB
Python
82 lines
3.7 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"backup_transfer", Path(__file__).resolve().parents[1] / "deploy" / "backup_transfer.py")
|
|
transfer = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(transfer)
|
|
|
|
|
|
class BackupTransferTests(unittest.TestCase):
|
|
def backup(self, directory: Path, name="guestops-20261001T021700Z.tar.gpg") -> Path:
|
|
path = directory / name
|
|
path.write_bytes(b"encrypted-backup")
|
|
return path
|
|
|
|
def config(self, directory: Path):
|
|
return {
|
|
"directory": directory, "host": "store.example.invalid", "user": "guestops_upload",
|
|
"remote": "/restricted/guestops", "identity": Path("/safe/key"),
|
|
"known_hosts": Path("/safe/known_hosts"),
|
|
}
|
|
|
|
def test_existing_matching_remote_is_marked_without_upload(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
backup = self.backup(Path(folder)); checksum = transfer.digest(backup)
|
|
with patch.object(transfer, "remote_digest", return_value=checksum), \
|
|
patch.object(transfer, "run") as run:
|
|
transfer.transfer_one(self.config(Path(folder)), backup)
|
|
run.assert_not_called()
|
|
marker = json.loads(transfer.marker_path(backup).read_text(encoding="utf-8"))
|
|
self.assertEqual(marker["sha256"], checksum)
|
|
|
|
def test_existing_mismatched_remote_is_never_overwritten(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
backup = self.backup(Path(folder))
|
|
with patch.object(transfer, "remote_digest", return_value="9" * 64), \
|
|
patch.object(transfer, "run") as run:
|
|
with self.assertRaisesRegex(RuntimeError, "different checksum"):
|
|
transfer.transfer_one(self.config(Path(folder)), backup)
|
|
run.assert_not_called()
|
|
self.assertFalse(transfer.marker_path(backup).exists())
|
|
|
|
def test_new_remote_is_uploaded_verified_and_marked(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
backup = self.backup(Path(folder)); checksum = transfer.digest(backup)
|
|
with patch.object(transfer, "remote_digest", side_effect=[None, checksum, checksum]), \
|
|
patch.object(transfer, "run", return_value=b"") as run, \
|
|
patch.object(transfer.subprocess, "run"):
|
|
transfer.transfer_one(self.config(Path(folder)), backup)
|
|
self.assertTrue(any(call.args[0][0] == "rsync" for call in run.call_args_list))
|
|
self.assertTrue(transfer.marker_path(backup).exists())
|
|
|
|
def test_prune_removes_only_old_checksum_verified_backups(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
directory = Path(folder)
|
|
verified = self.backup(directory)
|
|
checksum = transfer.digest(verified)
|
|
transfer.write_marker(verified, checksum)
|
|
unverified = self.backup(directory, "guestops-20261002T021700Z.tar.gpg")
|
|
old = time.time() - 8 * 86400
|
|
os.utime(verified, (old, old)); os.utime(unverified, (old, old))
|
|
removed = transfer.prune_verified(self.config(directory), 7, now=time.time())
|
|
self.assertEqual(removed, 1)
|
|
self.assertFalse(verified.exists())
|
|
self.assertTrue(unverified.exists())
|
|
|
|
def test_retention_bounds_are_enforced(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
with self.assertRaisesRegex(RuntimeError, "between 1 and 365"):
|
|
transfer.prune_verified(self.config(Path(folder)), 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|