|
|
@@ -0,0 +1,234 @@
|
|
|
+"""Unit tests for scheduled local backup service (#884)."""
|
|
|
+
|
|
|
+import tempfile
|
|
|
+import zipfile
|
|
|
+from datetime import datetime, timedelta, timezone
|
|
|
+from pathlib import Path
|
|
|
+from unittest.mock import AsyncMock, patch
|
|
|
+
|
|
|
+import pytest
|
|
|
+
|
|
|
+from backend.app.services.local_backup import LocalBackupService
|
|
|
+
|
|
|
+
|
|
|
+class TestCalculateNextRun:
|
|
|
+ """Tests for _calculate_next_run scheduling logic."""
|
|
|
+
|
|
|
+ def test_hourly_returns_next_full_hour(self):
|
|
|
+ service = LocalBackupService()
|
|
|
+ now = datetime(2026, 4, 12, 14, 30, 0, tzinfo=timezone.utc)
|
|
|
+ with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
|
+ mock_dt.now.return_value = now
|
|
|
+ mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
|
+ result = service._calculate_next_run("hourly", "03:00")
|
|
|
+ assert result.hour == 15
|
|
|
+ assert result.minute == 0
|
|
|
+
|
|
|
+ def test_daily_before_target_time_schedules_today(self):
|
|
|
+ service = LocalBackupService()
|
|
|
+ now = datetime(2026, 4, 12, 2, 0, 0, tzinfo=timezone.utc)
|
|
|
+ with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
|
+ mock_dt.now.return_value = now
|
|
|
+ mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
|
+ result = service._calculate_next_run("daily", "03:00")
|
|
|
+ assert result.day == 12
|
|
|
+ assert result.hour == 3
|
|
|
+
|
|
|
+ def test_daily_after_target_time_schedules_tomorrow(self):
|
|
|
+ service = LocalBackupService()
|
|
|
+ now = datetime(2026, 4, 12, 4, 0, 0, tzinfo=timezone.utc)
|
|
|
+ with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
|
+ mock_dt.now.return_value = now
|
|
|
+ mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
|
+ result = service._calculate_next_run("daily", "03:00")
|
|
|
+ assert result.day == 13
|
|
|
+ assert result.hour == 3
|
|
|
+
|
|
|
+ def test_weekly_adds_full_week(self):
|
|
|
+ service = LocalBackupService()
|
|
|
+ now = datetime(2026, 4, 12, 2, 0, 0, tzinfo=timezone.utc)
|
|
|
+ with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
|
+ mock_dt.now.return_value = now
|
|
|
+ mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
|
+ result = service._calculate_next_run("weekly", "03:00")
|
|
|
+ expected = datetime(2026, 4, 19, 3, 0, 0, tzinfo=timezone.utc)
|
|
|
+ assert result == expected
|
|
|
+
|
|
|
+ def test_weekly_after_target_time_adds_full_week_from_tomorrow(self):
|
|
|
+ service = LocalBackupService()
|
|
|
+ now = datetime(2026, 4, 12, 4, 0, 0, tzinfo=timezone.utc)
|
|
|
+ with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
|
+ mock_dt.now.return_value = now
|
|
|
+ mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
|
+ result = service._calculate_next_run("weekly", "03:00")
|
|
|
+ expected = datetime(2026, 4, 20, 3, 0, 0, tzinfo=timezone.utc)
|
|
|
+ assert result == expected
|
|
|
+
|
|
|
+ def test_invalid_time_defaults_to_0300(self):
|
|
|
+ service = LocalBackupService()
|
|
|
+ now = datetime(2026, 4, 12, 2, 0, 0, tzinfo=timezone.utc)
|
|
|
+ with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
|
+ mock_dt.now.return_value = now
|
|
|
+ mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
|
+ result = service._calculate_next_run("daily", "invalid")
|
|
|
+ assert result.hour == 3
|
|
|
+ assert result.minute == 0
|
|
|
+
|
|
|
+ def test_unknown_schedule_type_defaults_to_daily(self):
|
|
|
+ service = LocalBackupService()
|
|
|
+ now = datetime(2026, 4, 12, 2, 0, 0, tzinfo=timezone.utc)
|
|
|
+ with patch("backend.app.services.local_backup.datetime") as mock_dt:
|
|
|
+ mock_dt.now.return_value = now
|
|
|
+ mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
|
|
|
+ result = service._calculate_next_run("every_5_min", "03:00")
|
|
|
+ # Should fall through to daily behavior (time-based)
|
|
|
+ assert result.hour == 3
|
|
|
+
|
|
|
+
|
|
|
+class TestPruneBackups:
|
|
|
+ """Tests for backup retention pruning."""
|
|
|
+
|
|
|
+ def test_prune_keeps_retention_count(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ # Create 5 backup files
|
|
|
+ for i in range(5):
|
|
|
+ f = tmp_path / f"bambuddy-backup-20260412-{i:06d}.zip"
|
|
|
+ f.write_text(f"backup{i}")
|
|
|
+ service._prune_backups(tmp_path, retention=3)
|
|
|
+ remaining = list(tmp_path.glob("bambuddy-backup-*.zip"))
|
|
|
+ assert len(remaining) == 3
|
|
|
+
|
|
|
+ def test_prune_noop_when_under_retention(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ for i in range(2):
|
|
|
+ f = tmp_path / f"bambuddy-backup-20260412-{i:06d}.zip"
|
|
|
+ f.write_text(f"backup{i}")
|
|
|
+ service._prune_backups(tmp_path, retention=5)
|
|
|
+ remaining = list(tmp_path.glob("bambuddy-backup-*.zip"))
|
|
|
+ assert len(remaining) == 2
|
|
|
+
|
|
|
+ def test_prune_only_touches_matching_files(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ # Create backup files and a non-backup file
|
|
|
+ for i in range(3):
|
|
|
+ f = tmp_path / f"bambuddy-backup-20260412-{i:06d}.zip"
|
|
|
+ f.write_text(f"backup{i}")
|
|
|
+ other = tmp_path / "other_file.txt"
|
|
|
+ other.write_text("keep me")
|
|
|
+ service._prune_backups(tmp_path, retention=1)
|
|
|
+ assert other.exists()
|
|
|
+ remaining = list(tmp_path.glob("bambuddy-backup-*.zip"))
|
|
|
+ assert len(remaining) == 1
|
|
|
+
|
|
|
+
|
|
|
+class TestResolveBackupFile:
|
|
|
+ """Tests for backup file resolution with path traversal protection."""
|
|
|
+
|
|
|
+ def test_valid_filename(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ f = tmp_path / "bambuddy-backup-20260412-120000.zip"
|
|
|
+ f.write_text("data")
|
|
|
+ result = service.resolve_backup_file(str(tmp_path), "bambuddy-backup-20260412-120000.zip")
|
|
|
+ assert result == f
|
|
|
+
|
|
|
+ def test_path_traversal_blocked(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ result = service.resolve_backup_file(str(tmp_path), "../etc/passwd")
|
|
|
+ assert result is None
|
|
|
+
|
|
|
+ def test_backslash_blocked(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ result = service.resolve_backup_file(str(tmp_path), "..\\etc\\passwd")
|
|
|
+ assert result is None
|
|
|
+
|
|
|
+ def test_dotdot_blocked(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ result = service.resolve_backup_file(str(tmp_path), "..bambuddy-backup.zip")
|
|
|
+ assert result is None
|
|
|
+
|
|
|
+ def test_wrong_prefix_blocked(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ f = tmp_path / "evil-file.zip"
|
|
|
+ f.write_text("data")
|
|
|
+ result = service.resolve_backup_file(str(tmp_path), "evil-file.zip")
|
|
|
+ assert result is None
|
|
|
+
|
|
|
+ def test_nonexistent_file(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ result = service.resolve_backup_file(str(tmp_path), "bambuddy-backup-20260412-120000.zip")
|
|
|
+ assert result is None
|
|
|
+
|
|
|
+
|
|
|
+class TestDeleteBackup:
|
|
|
+ """Tests for backup deletion."""
|
|
|
+
|
|
|
+ def test_delete_valid_backup(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ f = tmp_path / "bambuddy-backup-20260412-120000.zip"
|
|
|
+ f.write_text("data")
|
|
|
+ result = service.delete_backup(str(tmp_path), "bambuddy-backup-20260412-120000.zip")
|
|
|
+ assert result["success"] is True
|
|
|
+ assert not f.exists()
|
|
|
+
|
|
|
+ def test_delete_nonexistent_backup(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ result = service.delete_backup(str(tmp_path), "bambuddy-backup-20260412-120000.zip")
|
|
|
+ assert result["success"] is False
|
|
|
+
|
|
|
+ def test_delete_path_traversal_blocked(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ result = service.delete_backup(str(tmp_path), "../important.zip")
|
|
|
+ assert result["success"] is False
|
|
|
+
|
|
|
+
|
|
|
+class TestListBackups:
|
|
|
+ """Tests for backup listing."""
|
|
|
+
|
|
|
+ def test_list_empty_dir(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ result = service.list_backups(str(tmp_path))
|
|
|
+ assert result == []
|
|
|
+
|
|
|
+ def test_list_nonexistent_dir(self):
|
|
|
+ service = LocalBackupService()
|
|
|
+ result = service.list_backups("/nonexistent/path/12345")
|
|
|
+ assert result == []
|
|
|
+
|
|
|
+ def test_list_only_matching_files(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ (tmp_path / "bambuddy-backup-20260412-120000.zip").write_text("a")
|
|
|
+ (tmp_path / "bambuddy-backup-20260412-130000.zip").write_text("bb")
|
|
|
+ (tmp_path / "other-file.txt").write_text("ccc")
|
|
|
+ result = service.list_backups(str(tmp_path))
|
|
|
+ assert len(result) == 2
|
|
|
+ assert all(r["filename"].startswith("bambuddy-backup-") for r in result)
|
|
|
+
|
|
|
+ def test_list_sorted_newest_first(self, tmp_path):
|
|
|
+ import time
|
|
|
+
|
|
|
+ service = LocalBackupService()
|
|
|
+ f1 = tmp_path / "bambuddy-backup-20260412-120000.zip"
|
|
|
+ f1.write_text("a")
|
|
|
+ time.sleep(0.05)
|
|
|
+ f2 = tmp_path / "bambuddy-backup-20260412-130000.zip"
|
|
|
+ f2.write_text("b")
|
|
|
+ result = service.list_backups(str(tmp_path))
|
|
|
+ assert result[0]["filename"] == "bambuddy-backup-20260412-130000.zip"
|
|
|
+
|
|
|
+ def test_list_includes_size(self, tmp_path):
|
|
|
+ service = LocalBackupService()
|
|
|
+ (tmp_path / "bambuddy-backup-20260412-120000.zip").write_bytes(b"x" * 1024)
|
|
|
+ result = service.list_backups(str(tmp_path))
|
|
|
+ assert result[0]["size"] == 1024
|
|
|
+
|
|
|
+
|
|
|
+class TestGetStatus:
|
|
|
+ """Tests for status reporting."""
|
|
|
+
|
|
|
+ def test_initial_status(self):
|
|
|
+ service = LocalBackupService()
|
|
|
+ status = service.get_status()
|
|
|
+ assert status["is_running"] is False
|
|
|
+ assert status["last_backup_at"] is None
|
|
|
+ assert status["last_status"] is None
|
|
|
+ assert status["next_run"] is None
|