File size: 2,995 Bytes
8059bf0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | import io
import tarfile
import tempfile
import unittest
from pathlib import Path
from deploy.huggingface.backup_manager import (
BackupArchiveError,
build_backup_filename,
create_backup_archive,
extract_backup_archive,
select_backups_to_delete,
)
class BackupManagerTests(unittest.TestCase):
def test_build_backup_filename_uses_utc_suffix(self):
name = build_backup_filename("20260321T102030Z")
self.assertEqual(name, "backup-20260321T102030Z.tar.gz")
def test_select_backups_to_delete_keeps_latest_pointer_and_newest_archives(self):
backups = [
"backups/latest.tar.gz",
"backups/backup-20260320T010101Z.tar.gz",
"backups/backup-20260321T010101Z.tar.gz",
"backups/backup-20260322T010101Z.tar.gz",
"notes/readme.txt",
]
to_delete = select_backups_to_delete(backups, keep=2)
self.assertEqual(to_delete, ["backups/backup-20260320T010101Z.tar.gz"])
def test_create_and_extract_backup_archive_round_trips_directories(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "source"
data_dir = source / "data"
postgres_dir = source / "postgres"
redis_dir = source / "redis"
data_dir.mkdir(parents=True)
postgres_dir.mkdir()
redis_dir.mkdir()
(data_dir / "config.yaml").write_text("hello", encoding="utf-8")
(postgres_dir / "PG_VERSION").write_text("16", encoding="utf-8")
(redis_dir / "appendonly.aof").write_text("redis", encoding="utf-8")
archive = root / "backup.tar.gz"
create_backup_archive(
archive,
{
"data": data_dir,
"postgres": postgres_dir,
"redis": redis_dir,
},
)
restore_root = root / "restore"
restore_root.mkdir()
extract_backup_archive(archive, restore_root)
self.assertEqual((restore_root / "data" / "config.yaml").read_text(encoding="utf-8"), "hello")
self.assertEqual((restore_root / "postgres" / "PG_VERSION").read_text(encoding="utf-8"), "16")
self.assertEqual((restore_root / "redis" / "appendonly.aof").read_text(encoding="utf-8"), "redis")
def test_extract_backup_archive_rejects_path_traversal(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
archive = root / "bad.tar.gz"
with tarfile.open(archive, "w:gz") as tar:
payload = b"oops"
info = tarfile.TarInfo("../escape.txt")
info.size = len(payload)
tar.addfile(info, io.BytesIO(payload))
with self.assertRaises(BackupArchiveError):
extract_backup_archive(archive, root / "restore")
if __name__ == "__main__":
unittest.main()
|