80 lines
2.8 KiB
Python
80 lines
2.8 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
from tempfile import TemporaryDirectory
|
||
|
|
from unittest.mock import patch
|
||
|
|
|
||
|
|
from django.test import SimpleTestCase
|
||
|
|
|
||
|
|
from pobsync.commands.run_scheduled import run_scheduled
|
||
|
|
from pobsync.rsync import RsyncResult
|
||
|
|
|
||
|
|
|
||
|
|
class FakeConfigSource:
|
||
|
|
def __init__(self, backup_root: str = "/tmp/pobsync-test-backups") -> None:
|
||
|
|
self.backup_root = backup_root
|
||
|
|
|
||
|
|
def effective_config_for_host(self, host: str) -> dict:
|
||
|
|
return {
|
||
|
|
"backup_root": self.backup_root,
|
||
|
|
"host": host,
|
||
|
|
"address": "example.test",
|
||
|
|
"ssh": {"user": "root", "port": 22, "options": []},
|
||
|
|
"rsync": {
|
||
|
|
"binary": "rsync",
|
||
|
|
"args_effective": ["--archive"],
|
||
|
|
"timeout_seconds": 0,
|
||
|
|
"bwlimit_kbps": 0,
|
||
|
|
},
|
||
|
|
"source_root": "/",
|
||
|
|
"includes": [],
|
||
|
|
"excludes_effective": [],
|
||
|
|
"retention": {"daily": 7, "weekly": 4, "monthly": 3, "yearly": 1},
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class RunScheduledConfigSourceTests(SimpleTestCase):
|
||
|
|
def test_dry_run_uses_injected_config_source(self) -> None:
|
||
|
|
with patch("pobsync.commands.run_scheduled.run_rsync") as run_rsync:
|
||
|
|
run_rsync.return_value = RsyncResult(exit_code=0, command=["rsync", "--archive"])
|
||
|
|
|
||
|
|
result = run_scheduled(
|
||
|
|
prefix=Path("/missing-prefix"),
|
||
|
|
host="web-01",
|
||
|
|
dry_run=True,
|
||
|
|
config_source=FakeConfigSource(),
|
||
|
|
)
|
||
|
|
|
||
|
|
self.assertTrue(result["ok"])
|
||
|
|
self.assertEqual(result["host"], "web-01")
|
||
|
|
run_rsync.assert_called_once()
|
||
|
|
|
||
|
|
def test_successful_real_run_applies_prune_when_requested(self) -> None:
|
||
|
|
with TemporaryDirectory() as tmp:
|
||
|
|
prefix = Path(tmp) / "home"
|
||
|
|
with (
|
||
|
|
patch("pobsync.commands.run_scheduled.run_rsync") as run_rsync,
|
||
|
|
patch("pobsync.commands.retention_apply.run_retention_plan") as plan,
|
||
|
|
):
|
||
|
|
run_rsync.return_value = RsyncResult(exit_code=0, command=["rsync", "--archive"])
|
||
|
|
plan.return_value = {
|
||
|
|
"ok": True,
|
||
|
|
"delete": [],
|
||
|
|
"keep": [],
|
||
|
|
"reasons": {},
|
||
|
|
"protect_bases": False,
|
||
|
|
}
|
||
|
|
|
||
|
|
result = run_scheduled(
|
||
|
|
prefix=prefix,
|
||
|
|
host="web-01",
|
||
|
|
dry_run=False,
|
||
|
|
prune=True,
|
||
|
|
prune_max_delete=10,
|
||
|
|
config_source=FakeConfigSource(backup_root=str(Path(tmp) / "backups")),
|
||
|
|
)
|
||
|
|
|
||
|
|
self.assertTrue(result["ok"])
|
||
|
|
self.assertIsNotNone(result["prune"])
|
||
|
|
self.assertEqual(result["prune"]["deleted"], [])
|