71 lines
2.7 KiB
Python
71 lines
2.7 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
from tempfile import TemporaryDirectory
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
from django.test import TestCase
|
||
|
|
|
||
|
|
from pobsync.util import write_yaml_atomic
|
||
|
|
from pobsync_backend.backup_runner import queue_backup_run
|
||
|
|
from pobsync_backend.management.commands.run_pobsync_worker import Command
|
||
|
|
from pobsync_backend.models import BackupRun, GlobalConfig, HostConfig, SnapshotRecord
|
||
|
|
|
||
|
|
|
||
|
|
class BackupWorkerTests(TestCase):
|
||
|
|
def test_queue_backup_run_records_requested_options(self) -> None:
|
||
|
|
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
||
|
|
|
||
|
|
run = queue_backup_run(
|
||
|
|
host=host,
|
||
|
|
dry_run=True,
|
||
|
|
prune=True,
|
||
|
|
prune_max_delete=3,
|
||
|
|
prune_protect_bases=True,
|
||
|
|
)
|
||
|
|
|
||
|
|
self.assertEqual(run.status, BackupRun.Status.QUEUED)
|
||
|
|
self.assertEqual(run.run_type, BackupRun.RunType.MANUAL)
|
||
|
|
self.assertEqual(
|
||
|
|
run.result["requested"],
|
||
|
|
{
|
||
|
|
"dry_run": True,
|
||
|
|
"prune": True,
|
||
|
|
"prune_max_delete": 3,
|
||
|
|
"prune_protect_bases": True,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_worker_executes_next_queued_run(self) -> None:
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
backup_root = Path(tmp) / "backups"
|
||
|
|
GlobalConfig.objects.create(name="default", backup_root=str(backup_root))
|
||
|
|
host = HostConfig.objects.create(host="web-01", address="web-01.example.test")
|
||
|
|
snapshot_dir = backup_root / host.host / "scheduled" / "20260519-021500Z__ABCDEFGH"
|
||
|
|
meta_dir = snapshot_dir / "meta"
|
||
|
|
meta_dir.mkdir(parents=True)
|
||
|
|
write_yaml_atomic(meta_dir / "meta.yaml", {"status": "success", "started_at": "2026-05-19T02:15:00Z"})
|
||
|
|
run = queue_backup_run(host=host)
|
||
|
|
|
||
|
|
with patch("pobsync_backend.backup_runner.run_scheduled") as run_scheduled:
|
||
|
|
run_scheduled.return_value = {
|
||
|
|
"ok": True,
|
||
|
|
"dry_run": False,
|
||
|
|
"host": host.host,
|
||
|
|
"snapshot": str(snapshot_dir),
|
||
|
|
"base": None,
|
||
|
|
"rsync": {"exit_code": 0},
|
||
|
|
}
|
||
|
|
count = Command()._run_once(prefix=Path(tmp) / "home")
|
||
|
|
|
||
|
|
self.assertEqual(count, 1)
|
||
|
|
run.refresh_from_db()
|
||
|
|
self.assertEqual(run.status, BackupRun.Status.SUCCESS)
|
||
|
|
self.assertEqual(SnapshotRecord.objects.count(), 1)
|
||
|
|
self.assertEqual(run.snapshot, SnapshotRecord.objects.get())
|
||
|
|
|
||
|
|
def test_worker_returns_zero_without_queued_runs(self) -> None:
|
||
|
|
count = Command()._run_once(prefix=Path("/opt/pobsync"))
|
||
|
|
|
||
|
|
self.assertEqual(count, 0)
|