Test whole remote backups through a fake ssh command
Real rsync runs over fake ssh and sshpass scripts that execute the remote side locally, covering --link-dest, removal, key file, password and --remote-sudo without a server.
This commit is contained in:
+23
-1
@@ -1,7 +1,8 @@
|
||||
"""Shared fixtures for the simple_backup tests"""
|
||||
|
||||
import sys
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Callable, Iterator
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -18,6 +19,27 @@ def no_side_effects(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(sb, '_notify', lambda text: None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_now(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Give each backup a new timestamp, as real runs within the same second would collide"""
|
||||
|
||||
def times() -> Iterator[datetime]:
|
||||
t = datetime(2026, 10, 14, 10, 0, 0)
|
||||
|
||||
while True:
|
||||
yield t
|
||||
t += timedelta(hours=1)
|
||||
|
||||
it = times()
|
||||
|
||||
class FakeDatetime:
|
||||
@staticmethod
|
||||
def now() -> datetime:
|
||||
return next(it)
|
||||
|
||||
monkeypatch.setattr(sb, 'datetime', FakeDatetime)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def run_cli(monkeypatch: pytest.MonkeyPatch) -> Callable[..., int]:
|
||||
"""Run simple_backup() with the given command line arguments"""
|
||||
|
||||
+2
-21
@@ -3,8 +3,7 @@
|
||||
import errno
|
||||
import io
|
||||
import os
|
||||
from collections.abc import Callable, Iterator
|
||||
from datetime import datetime, timedelta
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
from shutil import which
|
||||
from typing import Any
|
||||
@@ -16,25 +15,7 @@ from simple_backup import simple_backup as sb
|
||||
OPTIONS = '-r -v --ignore-missing-args --mkpath -a -h -H -X -s'
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def fake_now(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Give each backup a new timestamp, as real runs within the same second would collide"""
|
||||
|
||||
def times() -> Iterator[datetime]:
|
||||
t = datetime(2026, 10, 14, 10, 0, 0)
|
||||
|
||||
while True:
|
||||
yield t
|
||||
t += timedelta(hours=1)
|
||||
|
||||
it = times()
|
||||
|
||||
class FakeDatetime:
|
||||
@staticmethod
|
||||
def now() -> datetime:
|
||||
return next(it)
|
||||
|
||||
monkeypatch.setattr(sb, 'datetime', FakeDatetime)
|
||||
pytestmark = pytest.mark.usefixtures('fake_now')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@@ -603,3 +603,108 @@ class TestLoadKey:
|
||||
|
||||
assert remote_backup(ssh_keyfile=str(keyfile))._load_ssh_key() is None
|
||||
assert 'Unable to load SSH key' in caplog.text
|
||||
|
||||
|
||||
FAKE_SSH = """#!/bin/sh
|
||||
# Record the arguments, then run the remote command locally in the fake remote home
|
||||
for a in "$@"; do printf '%s\\n' "$a"; done >> "{log}"
|
||||
echo --- >> "{log}"
|
||||
|
||||
while [ $# -gt 0 ]; do
|
||||
case $1 in
|
||||
-i|-l|-o|-p) shift 2 ;;
|
||||
-*) shift ;;
|
||||
*) break ;;
|
||||
esac
|
||||
done
|
||||
|
||||
# The host
|
||||
shift
|
||||
cd "{home}" && HOME="{home}" exec sh -c "$*"
|
||||
"""
|
||||
|
||||
FAKE_SSHPASS = """#!/bin/sh
|
||||
printf '%s' "${{SSHPASS-unset}}" > "{log}"
|
||||
# -e
|
||||
shift
|
||||
exec "$@"
|
||||
"""
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('connect', 'fake_now')
|
||||
class TestRsyncOverSsh:
|
||||
"""Whole remote backups: real rsync, with fake ssh and sshpass commands that run the remote side locally"""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def fake_ssh(self, ssh: FakeSSH, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
self.ssh_log = tmp_path / 'ssh.log'
|
||||
self.sshpass_log = tmp_path / 'sshpass.log'
|
||||
|
||||
for name, script, log in [('ssh', FAKE_SSH, self.ssh_log), ('sshpass', FAKE_SSHPASS, self.sshpass_log)]:
|
||||
path = ssh.bin_dir / name
|
||||
path.write_text(script.format(log=log, home=ssh.home))
|
||||
path.chmod(0o755)
|
||||
|
||||
monkeypatch.setenv('PATH', f'{ssh.bin_dir}:{os.environ["PATH"]}')
|
||||
monkeypatch.delenv('SSHPASS', raising=False)
|
||||
|
||||
self.src = tmp_path / 'src dir'
|
||||
self.src.mkdir()
|
||||
(self.src / 'same.txt').write_text('unchanged')
|
||||
(self.src / 'changed.txt').write_text('first')
|
||||
|
||||
def run(self, keep: int = -1, password: str | None = None, **kwargs: Any) -> int:
|
||||
backup = connected_backup(keep=keep, **kwargs)
|
||||
backup.inputs = [str(self.src)]
|
||||
backup._password = password
|
||||
|
||||
return backup.run()
|
||||
|
||||
def ssh_calls(self) -> list[list[str]]:
|
||||
return [c.strip('\n').split('\n') for c in self.ssh_log.read_text().split('---\n') if c != '']
|
||||
|
||||
def copied(self, remote_home: Path, backup: str, name: str) -> Path:
|
||||
return remote_home / 'backups' / 'simple_backup' / backup / str(self.src).lstrip('/') / name
|
||||
|
||||
def test_backups(self, remote_home: Path) -> None:
|
||||
assert self.run() == 0
|
||||
(self.src / 'changed.txt').write_text('second')
|
||||
assert self.run(keep=1) == 0
|
||||
assert self.run(keep=1) == 0
|
||||
|
||||
old, new = '2026-10-14_11-00-00', '2026-10-14_12-00-00'
|
||||
assert remote_dirs(remote_home / 'backups' / 'simple_backup') == [old, new]
|
||||
assert self.copied(remote_home, new, 'changed.txt').read_text() == 'second'
|
||||
# Unchanged files are hard-linked to the previous backup (--link-dest on the server)
|
||||
assert self.copied(remote_home, old, 'same.txt').stat().st_ino == \
|
||||
self.copied(remote_home, new, 'same.txt').stat().st_ino
|
||||
|
||||
for call in self.ssh_calls():
|
||||
assert call[:5] == ['-o', 'StrictHostKeyChecking=accept-new', '-l', 'user', 'server']
|
||||
# rsync passes the remote command as separate arguments
|
||||
assert call[5:7] == ['rsync', '--server']
|
||||
|
||||
def test_keyfile(self, remote_home: Path) -> None:
|
||||
assert self.run(ssh_keyfile='/keys/my key') == 0
|
||||
|
||||
(call,) = self.ssh_calls()
|
||||
assert call[:4] == ['-i', '/keys/my key', '-o', 'StrictHostKeyChecking=accept-new']
|
||||
|
||||
def test_password(self, remote_home: Path) -> None:
|
||||
assert self.run(password='secret') == 0
|
||||
|
||||
assert self.sshpass_log.read_text() == 'secret'
|
||||
assert 'SSHPASS' not in os.environ
|
||||
assert len(remote_dirs(remote_home / 'backups' / 'simple_backup')) == 1
|
||||
|
||||
def test_remote_sudo(self, remote_home: Path) -> None:
|
||||
assert self.run() == 0
|
||||
assert self.run(keep=0, remote_sudo=True) == 0
|
||||
|
||||
assert remote_dirs(remote_home / 'backups' / 'simple_backup') == ['2026-10-14_11-00-00']
|
||||
assert self.ssh_calls()[-1][5:7] == ['sudo rsync', '--server']
|
||||
|
||||
def test_progress(self, remote_home: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
assert self.run(progress=True) == 0
|
||||
|
||||
assert '100%' in capsys.readouterr().err
|
||||
|
||||
Reference in New Issue
Block a user