| 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() |
|
|
