fix(backup): harden Google Drive freshness readback
This commit is contained in:
@@ -0,0 +1,95 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
SCRIPT = ROOT / "scripts" / "backup" / "verify-offsite-full-sync.sh"
|
||||
|
||||
|
||||
def run_verifier(tmp_path: Path, mode: str) -> tuple[subprocess.CompletedProcess[str], str, Path]:
|
||||
bin_dir = tmp_path / "bin"
|
||||
bin_dir.mkdir()
|
||||
runtime_home = tmp_path / "runtime-home"
|
||||
config = runtime_home / ".config" / "rclone" / "rclone.conf"
|
||||
config.parent.mkdir(parents=True)
|
||||
config.write_text("test fixture only\n", encoding="utf-8")
|
||||
event_log = tmp_path / "rclone.events"
|
||||
|
||||
getent = bin_dir / "getent"
|
||||
getent.write_text(
|
||||
"#!/bin/sh\nprintf 'wooo:x:1000:1000::%s:/bin/bash\\n' \"$TEST_RUNTIME_HOME\"\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
getent.chmod(0o755)
|
||||
|
||||
rclone = bin_dir / "rclone"
|
||||
rclone.write_text(
|
||||
"""#!/bin/sh
|
||||
printf 'CMD=%s CONFIG=%s\n' "$*" "${RCLONE_CONFIG:-}" >> "$TEST_EVENT_LOG"
|
||||
case "$1" in
|
||||
listremotes)
|
||||
[ "$TEST_RCLONE_MODE" = config_missing ] && printf '%s\n' 'other:' || printf '%s\n' 'gdrive:'
|
||||
;;
|
||||
lsf)
|
||||
if [ "$TEST_RCLONE_MODE" = remote_down ]; then
|
||||
exit 9
|
||||
fi
|
||||
case "$2" in
|
||||
*/snapshots) printf '%s\n' snapshot-file ;;
|
||||
*) printf '%s\n' awoooi/ ;;
|
||||
esac
|
||||
;;
|
||||
esac
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
rclone.chmod(0o755)
|
||||
|
||||
env = os.environ.copy()
|
||||
env.pop("RCLONE_CONFIG", None)
|
||||
env.update(
|
||||
{
|
||||
"BACKUP_BASE": str(tmp_path / "backup"),
|
||||
"HOME": "/",
|
||||
"OFFSITE_REPOS": "awoooi",
|
||||
"PATH": f"{bin_dir}:{env['PATH']}",
|
||||
"TEST_EVENT_LOG": str(event_log),
|
||||
"TEST_RCLONE_MODE": mode,
|
||||
"TEST_RUNTIME_HOME": str(runtime_home),
|
||||
}
|
||||
)
|
||||
result = subprocess.run(
|
||||
["bash", str(SCRIPT), "--no-color"],
|
||||
env=env,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
return result, event_log.read_text(encoding="utf-8"), config
|
||||
|
||||
|
||||
def test_verifier_binds_rclone_config_to_runtime_account(tmp_path: Path) -> None:
|
||||
result, events, config = run_verifier(tmp_path, "ok")
|
||||
|
||||
assert result.returncode == 1 # The full-sync marker is intentionally absent.
|
||||
assert "rclone provider readback succeeded" in result.stdout
|
||||
assert f"CONFIG={config}" in events
|
||||
|
||||
|
||||
def test_verifier_distinguishes_missing_config_without_remote_calls(tmp_path: Path) -> None:
|
||||
result, events, _ = run_verifier(tmp_path, "config_missing")
|
||||
|
||||
assert result.returncode == 1
|
||||
assert "rclone remote configuration unavailable" in result.stdout
|
||||
assert " lsf " not in f" {events} "
|
||||
|
||||
|
||||
def test_verifier_distinguishes_provider_failure_and_stops_fanout(tmp_path: Path) -> None:
|
||||
result, events, _ = run_verifier(tmp_path, "remote_down")
|
||||
|
||||
assert result.returncode == 1
|
||||
assert "rclone provider readback failed" in result.stdout
|
||||
assert events.count("CMD=lsf ") == 1
|
||||
Reference in New Issue
Block a user