mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 5.5
21eba1e88f
baudolo 7.1.x writes volume generations with `rsync -aHAX`, so the producing host keeps `user.*` extended attributes (OpenCloud keeps its storage metadata in `user.oc.*`). Every later copy dropped them again, because none of the other rsync calls passed `-X` (nor `-A`/`-H`): - NFS export snapshots (svc-bkp-nfs-2-local `script.sh`), - the backup host's pull (svc-bkp-remote-2-local `pull_specific_host.py`), - the device sync (svc-bkp-local-2-device `script.py`), - every recover into a live target (`utils/recovery/base.py`, used by the volume, NFS and device recover scripts), - the `recover full` and `recover remote` tree pulls. A disaster recovery through the documented chain therefore restored OpenCloud without `user.oc.*`, silently: the copies succeeded and the dirval stamp does not cover attributes. All of these now run `rsync -aHAX` (`-aHAXP` for the device sync). A target filesystem without xattr support makes rsync exit non-zero as soon as a source file carries attributes, so the hop fails loudly instead of dropping them; there is no per-device opt-out. The single-image pull in `recover remote` keeps `-a`. The provider's ssh-wrapper needs no change: its glob accepts any server flags and its character guard allows them. Two tests write a `user.oc.id` attribute and assert it survives the NFS snapshot and the recover; with the old `-a` both fail with `OSError: [Errno 61] No data available`. Closes #1021 Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
import unittest.mock as mock
|
|
from pathlib import Path
|
|
|
|
from utils.recovery.base import DirectoryRecovery
|
|
|
|
|
|
class _Recovery(DirectoryRecovery):
|
|
unit_pattern = "svc-bkp-test*.service"
|
|
|
|
|
|
class TestDirectoryRecovery(unittest.TestCase):
|
|
def _mk(self, td: str) -> tuple[Path, Path]:
|
|
root = Path(td)
|
|
source = root / "source"
|
|
target = root / "target"
|
|
source.mkdir()
|
|
target.mkdir()
|
|
(source / "restored.txt").write_text("from-snapshot")
|
|
(target / "live.txt").write_text("pre-recover")
|
|
return source, target
|
|
|
|
def test_run_starts_backup_unit_then_mirrors_source(self) -> None:
|
|
with tempfile.TemporaryDirectory() as td:
|
|
source, target = self._mk(td)
|
|
recovery = _Recovery(str(source), str(target))
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
calls.append(cmd)
|
|
result = mock.Mock()
|
|
if cmd[:2] == ["systemctl", "list-unit-files"]:
|
|
result.stdout = "svc-bkp-test.1.example.service enabled\n"
|
|
return result
|
|
if cmd[:2] == ["systemctl", "start"]:
|
|
return result
|
|
return mock.DEFAULT
|
|
|
|
with mock.patch(
|
|
"utils.recovery.base.subprocess.run", side_effect=fake_run
|
|
) as run:
|
|
run.side_effect = fake_run
|
|
recovery.backup_target()
|
|
|
|
self.assertEqual(
|
|
calls[1], ["systemctl", "start", "svc-bkp-test.1.example.service"]
|
|
)
|
|
|
|
recovery.restore()
|
|
restored = target / "restored.txt"
|
|
text = restored.read_text() # nocheck: cache-read - tempdir fixture
|
|
self.assertEqual(text, "from-snapshot")
|
|
self.assertFalse((target / "live.txt").exists())
|
|
|
|
def test_restore_keeps_user_xattrs(self) -> None:
|
|
with tempfile.TemporaryDirectory() as td:
|
|
source, target = self._mk(td)
|
|
os.setxattr(source / "restored.txt", "user.oc.id", b"space-1")
|
|
_Recovery(str(source), str(target)).restore()
|
|
self.assertEqual(
|
|
os.getxattr(target / "restored.txt", "user.oc.id"), b"space-1"
|
|
)
|
|
|
|
def test_missing_unit_refuses(self) -> None:
|
|
with tempfile.TemporaryDirectory() as td:
|
|
source, target = self._mk(td)
|
|
recovery = _Recovery(str(source), str(target))
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
result = mock.Mock()
|
|
result.stdout = ""
|
|
return result
|
|
|
|
with (
|
|
mock.patch("utils.recovery.base.subprocess.run", side_effect=fake_run),
|
|
self.assertRaises(SystemExit),
|
|
):
|
|
recovery.backup_target()
|
|
|
|
def test_no_safety_backup_skips_unit_run(self) -> None:
|
|
with tempfile.TemporaryDirectory() as td:
|
|
source, target = self._mk(td)
|
|
recovery = _Recovery(str(source), str(target), service_backup=False)
|
|
with mock.patch.object(recovery, "backup_target") as backup:
|
|
recovery.run()
|
|
backup.assert_not_called()
|
|
restored = target / "restored.txt"
|
|
text = restored.read_text() # nocheck: cache-read - tempdir fixture
|
|
self.assertEqual(text, "from-snapshot")
|
|
|
|
def test_missing_target_refuses(self) -> None:
|
|
with tempfile.TemporaryDirectory() as td:
|
|
source, target = self._mk(td)
|
|
with self.assertRaises(SystemExit):
|
|
_Recovery(str(source), str(target / "absent"))
|
|
|
|
def test_unit_pattern_is_mandatory(self) -> None:
|
|
class _NoUnit(DirectoryRecovery):
|
|
pass
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
source, target = self._mk(td)
|
|
with self.assertRaises(ValueError):
|
|
_NoUnit(str(source), str(target))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|