Add tests for remote backups

Use a fake SSH client that runs commands locally to test remote path resolution, removing old backups, finding the last backup and the rsync arguments, plus the login paths with a fake paramiko client and real keys made by ssh-keygen.
2026-10-14 08:23:25 +02:00
parent 28ecff2306
commit 8d08032b5d
1 changed files with 566 additions and 0 deletions
@@ -0,0 +1,566 @@
+"""Tests for remote backups, with a fake SSH client that runs commands locally"""
+
+import importlib.util
+import io
+import os
+import shlex
+import socket
+import subprocess
+import warnings
+from getpass import GetPassWarning
+from pathlib import Path
+from shutil import which
+from typing import Any, Optional
+
+import pytest
+
+from simple_backup import simple_backup as sb
+
+OPTIONS = '-r -v --ignore-missing-args --mkpath -a -h -H -X -s'
+
+needs_paramiko = pytest.mark.skipif(importlib.util.find_spec('paramiko') is None,
+                                    reason='paramiko not installed')
+needs_ssh_keygen = pytest.mark.skipif(which('ssh-keygen') is None, reason='ssh-keygen not installed')
+
+
+class FakeSSH:
+    """SSHClient stand-in: runs commands with sh in a fake remote home"""
+
+    def __init__(self, home: Path, bin_dir: Path) -> None:
+        self.home = home
+        self.bin_dir = bin_dir
+        self.commands: list[str] = []
+        self.fail_on: Optional[str] = None
+        self.closed = False
+
+    def exec_command(self, command: str) -> tuple[None, io.BytesIO, io.BytesIO]:
+        self.commands.append(command)
+
+        if self.fail_on is not None and self.fail_on in command:
+            return None, io.BytesIO(b''), io.BytesIO(b'Permission denied\n')
+
+        env = {'HOME': str(self.home), 'PATH': f'{self.bin_dir}:{os.environ["PATH"]}'}
+        p = subprocess.run(['sh', '-c', command], cwd=self.home, env=env, capture_output=True, check=False)
+
+        return None, io.BytesIO(p.stdout), io.BytesIO(p.stderr)
+
+    def close(self) -> None:
+        self.closed = True
+
+
+@pytest.fixture
+def remote_home(tmp_path: Path) -> Path:
+    home = tmp_path / 'remote home'
+    (home / 'backups').mkdir(parents=True)
+
+    return home
+
+
+@pytest.fixture
+def ssh(remote_home: Path, tmp_path: Path) -> FakeSSH:
+    # A 'sudo' that just runs the command, for --remote-sudo
+    bin_dir = tmp_path / 'bin'
+    bin_dir.mkdir()
+    sudo = bin_dir / 'sudo'
+    sudo.write_text('#!/bin/sh\nexec "$@"\n')
+    sudo.chmod(0o755)
+
+    return FakeSSH(remote_home, bin_dir)
+
+
+@pytest.fixture
+def connect(monkeypatch: pytest.MonkeyPatch, ssh: FakeSSH) -> FakeSSH:
+    monkeypatch.setattr(sb.Backup, '_ssh_connect', lambda self, homedir='': ssh)
+
+    return ssh
+
+
+def remote_backup(output: str = '~/backups', keep: int = -1, **kwargs: Any) -> sb.Backup:
+    return sb.Backup(['/x'], output, [], keep, OPTIONS, ssh_host='server', ssh_user='user', **kwargs)
+
+
+def connected_backup(output: str = '~/backups', keep: int = -1, **kwargs: Any) -> sb.Backup:
+    backup = remote_backup(output, keep, **kwargs)
+
+    assert backup.check_params() == 0
+
+    return backup
+
+
+def make_backups(remote_home: Path, *names: str) -> Path:
+    base = remote_home / 'backups' / 'simple_backup'
+    base.mkdir()
+
+    for name in names:
+        (base / name).mkdir()
+
+    return base
+
+
+def remote_dirs(base: Path) -> list[str]:
+    return sorted(d.name for d in base.iterdir() if d.is_dir())
+
+
+@pytest.mark.usefixtures('connect')
+class TestRemotePath:
+    @pytest.mark.parametrize('output', ['~/backups', 'backups', '~/backups/', '{home}/backups'])
+    def test_resolved(self, remote_home: Path, output: str) -> None:
+        backup = connected_backup(output.format(home=remote_home))
+
+        assert backup.output == f'{remote_home}/backups'
+
+    def test_home(self, remote_home: Path) -> None:
+        assert connected_backup('~').output == str(remote_home)
+
+    def test_shell_characters_not_expanded(self, remote_home: Path) -> None:
+        name = 'a $HOME `touch pwned` b'
+        (remote_home / name).mkdir()
+
+        assert connected_backup(f'~/{name}').output == f'{remote_home}/{name}'
+        assert not (remote_home / 'pwned').exists()
+
+    def test_missing(self, caplog: pytest.LogCaptureFixture) -> None:
+        assert remote_backup('~/missing').check_params() == 2
+        assert 'Output path for backup does not exist' in caplog.text
+
+    def test_needs_host_and_user(self, tmp_path: Path) -> None:
+        # Only one of them: a local backup
+        backup = sb.Backup(['/x'], str(tmp_path), [], -1, OPTIONS, ssh_host='server')
+
+        assert backup.check_params() == 0
+        assert not backup._remote
+
+
+class TestConnectionErrors:
+    def test_connection_failed(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(sb.Backup, '_ssh_connect', lambda self, homedir='': None)
+
+        assert remote_backup().check_params() == 5
+
+    @pytest.mark.parametrize('error', [socket.gaierror(-2, 'Name or service not known'),
+                                       ConnectionRefusedError(111, 'Connection refused'),
+                                       TimeoutError('timed out')])
+    def test_os_error(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture,
+                      error: OSError) -> None:
+        def fail(self: sb.Backup, homedir: str = '') -> None:
+            raise error
+
+        monkeypatch.setattr(sb.Backup, '_ssh_connect', fail)
+
+        assert remote_backup().check_params() == 5
+        assert 'Can\'t connect to the server server' in caplog.text
+
+    def test_paramiko_missing(self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
+        monkeypatch.delattr(sb, 'paramiko', raising=False)
+
+        assert remote_backup().check_params() == 5
+        assert 'Install paramiko' in caplog.text
+
+
+@pytest.mark.usefixtures('connect')
+class TestRemoveOldBackups:
+    def test_keep(self, remote_home: Path) -> None:
+        base = make_backups(remote_home, '2026-10-01 10:00:00', '2026-10-02_10-00-00', '2026-10-03_10-00-00')
+        (base / 'notes.txt').touch()
+
+        # The newest one stands for the backup just made
+        connected_backup(keep=1).remove_old_backups()
+
+        assert remote_dirs(base) == ['2026-10-02_10-00-00', '2026-10-03_10-00-00']
+        assert (base / 'notes.txt').exists()
+
+    def test_remove_before(self, remote_home: Path) -> None:
+        base = make_backups(remote_home, '2026-10-01_10-00-00', '2026-10-02_10-00-00')
+
+        connected_backup(keep=0, remove_before=True).remove_old_backups()
+
+        assert remote_dirs(base) == []
+        assert base.is_dir()
+
+    def test_remote_sudo(self, remote_home: Path, ssh: FakeSSH) -> None:
+        base = make_backups(remote_home, '2026-10-01_10-00-00', '2026-10-02_10-00-00')
+
+        connected_backup(keep=0, remote_sudo=True).remove_old_backups()
+
+        assert remote_dirs(base) == ['2026-10-02_10-00-00']
+        assert [c for c in ssh.commands if 'rm -r' in c][0].startswith('sudo rm -r ')
+
+    @pytest.mark.parametrize('names', [[], None])
+    def test_nothing_to_remove(self, remote_home: Path, names: Optional[list[str]]) -> None:
+        if names is not None:
+            make_backups(remote_home, *names)
+
+        connected_backup(keep=0, remove_before=True).remove_old_backups()
+
+        # The backup folder itself must never be removed
+        assert (remote_home / 'backups').is_dir()
+        assert (remote_home / 'backups' / 'simple_backup').is_dir() == (names is not None)
+
+    def test_list_error(self, remote_home: Path, ssh: FakeSSH, caplog: pytest.LogCaptureFixture) -> None:
+        base = make_backups(remote_home, '2026-10-01_10-00-00', '2026-10-02_10-00-00')
+        backup = connected_backup(keep=0)
+        ssh.fail_on = 'find'
+
+        backup.remove_old_backups()
+
+        assert remote_dirs(base) == ['2026-10-01_10-00-00', '2026-10-02_10-00-00']
+        assert 'Cannot list old backups on the server' in caplog.text
+
+    def test_remove_error(self, remote_home: Path, ssh: FakeSSH, caplog: pytest.LogCaptureFixture) -> None:
+        make_backups(remote_home, '2026-10-01_10-00-00', '2026-10-02_10-00-00')
+        backup = connected_backup(keep=0)
+        ssh.fail_on = 'rm -r'
+
+        backup.remove_old_backups()
+
+        assert 'Error while removing backup 2026-10-01_10-00-00' in caplog.text
+        assert backup._removed_count == 0
+
+
+@pytest.mark.usefixtures('connect')
+class TestFindLastBackup:
+    def test_newest(self, remote_home: Path) -> None:
+        base = make_backups(remote_home, '2026-10-01 10:00:00', '2026-10-02_10-00-00')
+        (base / 'zzz.txt').touch()
+        backup = connected_backup()
+
+        backup.find_last_backup()
+
+        assert backup._last_backup == f'{base}/2026-10-02_10-00-00'
+
+    @pytest.mark.parametrize('create', [True, False])
+    def test_none(self, remote_home: Path, caplog: pytest.LogCaptureFixture, create: bool) -> None:
+        if create:
+            make_backups(remote_home)
+
+        backup = connected_backup()
+        backup.find_last_backup()
+
+        assert backup._last_backup == ''
+        assert 'No previous backups available' in caplog.text
+
+    def test_no_connection(self) -> None:
+        backup = connected_backup()
+        backup._ssh = None
+
+        with pytest.raises(SystemExit) as e:
+            backup.find_last_backup()
+
+        assert e.value.code == 5
+
+
+class RecordPopen:
+    """Records the rsync arguments instead of running rsync"""
+
+    returncode = 0
+    calls: list[list[str]] = []
+
+    def __init__(self, args: list[str], **kwargs: Any) -> None:
+        RecordPopen.calls.append(args)
+
+    def __enter__(self) -> 'RecordPopen':
+        return self
+
+    def __exit__(self, *args: Any) -> None:
+        pass
+
+    def communicate(self) -> tuple[bytes, None]:
+        return b'sent 1 bytes  received 2 bytes\ntotal size is 3\n', None
+
+
+@pytest.mark.usefixtures('connect')
+class TestRsyncArgs:
+    @pytest.fixture(autouse=True)
+    def record_popen(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+        RecordPopen.returncode = 0
+        RecordPopen.calls = []
+        monkeypatch.setattr(sb, 'Popen', RecordPopen)
+        # The input must exist locally
+        (tmp_path / 'input').mkdir()
+
+    def run(self, tmp_path: Path, **kwargs: Any) -> tuple[int, list[str]]:
+        backup = connected_backup(**kwargs)
+        backup.inputs = [str(tmp_path / 'input')]
+        returncode = backup.run()
+        (args,) = RecordPopen.calls
+
+        return returncode, args
+
+    @staticmethod
+    def ssh_command(args: list[str]) -> str:
+        return args[args.index('-e') + 1]
+
+    def test_default(self, tmp_path: Path, remote_home: Path) -> None:
+        returncode, args = self.run(tmp_path)
+
+        assert returncode == 0
+        assert args[args.index('/') + 1].startswith(f'user@server:{remote_home}/backups/simple_backup/')
+        assert self.ssh_command(args) == 'ssh -o StrictHostKeyChecking=accept-new'
+        assert not any(a.startswith('--rsync-path') for a in args)
+        assert not any(a.startswith('--link-dest') for a in args)
+
+    def test_link_dest(self, tmp_path: Path, remote_home: Path) -> None:
+        base = make_backups(remote_home, '2026-10-01_10-00-00')
+
+        _, args = self.run(tmp_path)
+
+        assert f'--link-dest={base}/2026-10-01_10-00-00' in args
+
+    def test_keyfile_with_spaces(self, tmp_path: Path) -> None:
+        _, args = self.run(tmp_path, ssh_keyfile='/keys/my key')
+
+        # rsync splits the -e command like a shell
+        assert shlex.split(self.ssh_command(args)) == ['ssh', '-i', '/keys/my key', '-o',
+                                                       'StrictHostKeyChecking=accept-new']
+
+    def test_remote_sudo(self, tmp_path: Path) -> None:
+        _, args = self.run(tmp_path, remote_sudo=True)
+
+        assert '--rsync-path=sudo rsync' in args
+
+    def test_password_auth(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(sb, 'which', lambda cmd: f'/usr/bin/{cmd}')
+        monkeypatch.setenv('SSHPASS', 'secret')
+        backup = connected_backup()
+        backup.inputs = [str(tmp_path / 'input')]
+        backup._password_auth = True
+
+        assert backup.run() == 0
+
+        (args,) = RecordPopen.calls
+        assert shlex.split(self.ssh_command(args)) == ['sshpass', '-e', 'ssh', '-l', 'user', '-o',
+                                                       'StrictHostKeyChecking=accept-new']
+        # The password must not stay in the environment
+        assert 'SSHPASS' not in os.environ
+
+    def test_failure(self, tmp_path: Path, remote_home: Path) -> None:
+        base = make_backups(remote_home, '2026-10-01_10-00-00')
+        RecordPopen.returncode = 23
+
+        returncode, _ = self.run(tmp_path, keep=0)
+
+        assert returncode == 4
+        assert remote_dirs(base) == ['2026-10-01_10-00-00']
+
+
+class FakeClient:
+    """paramiko.SSHClient stand-in; connect() takes its outcomes from a list"""
+
+    def __init__(self, outcomes: list[Optional[BaseException]]) -> None:
+        self.outcomes = outcomes
+        self.connects: list[dict[str, Any]] = []
+        self.policies: list[str] = []
+        self.known_hosts: Optional[str] = None
+        self.closed = False
+
+    def load_host_keys(self, filename: str) -> None:
+        self.known_hosts = filename
+
+        if not os.path.isfile(filename):
+            raise FileNotFoundError(filename)
+
+    def set_missing_host_key_policy(self, policy: Any) -> None:
+        self.policies.append(type(policy).__name__)
+
+    def connect(self, hostname: str, **kwargs: Any) -> None:
+        self.connects.append(kwargs)
+        outcome = self.outcomes.pop(0)
+
+        if isinstance(outcome, UserWarning):
+            # Like WarningPolicy for an unknown host key
+            warnings.warn(outcome)
+        elif outcome is not None:
+            raise outcome
+
+    def close(self) -> None:
+        self.closed = True
+
+
+@pytest.fixture
+def client(monkeypatch: pytest.MonkeyPatch) -> FakeClient:
+    fake = FakeClient([])
+    monkeypatch.setattr(sb.paramiko, 'SSHClient', lambda: fake)
+
+    def no_terminal(prompt: str = '') -> str:
+        raise AssertionError(f'Unexpected prompt: {prompt}')
+
+    monkeypatch.setattr(sb, 'getpass', no_terminal)
+    monkeypatch.setattr('builtins.input', no_terminal)
+
+    return fake
+
+
+def ssh_error() -> Exception:
+    return sb.paramiko.SSHException('Authentication failed')
+
+
+@needs_paramiko
+class TestLogin:
+    def test_agent(self, client: FakeClient, tmp_path: Path) -> None:
+        client.outcomes = [None]
+
+        assert remote_backup()._ssh_connect(str(tmp_path)) is client
+        assert client.known_hosts == f'{tmp_path}/.ssh/known_hosts'
+        assert client.connects[0]['timeout'] == sb.SSH_TIMEOUT
+        assert not client.closed
+
+    @pytest.mark.parametrize('answer, connected', [('y', True), (' Yes', True), ('n', False), ('', False)])
+    def test_unknown_host_key(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch,
+                              answer: str, connected: bool) -> None:
+        client.outcomes = [UserWarning('Unknown host key'), None]
+        monkeypatch.setattr('builtins.input', lambda prompt: answer)
+
+        result = remote_backup()._ssh_connect()
+
+        assert (result is client) == connected
+        assert client.closed != connected
+
+        if connected:
+            assert client.policies == ['WarningPolicy', 'AutoAddPolicy']
+
+    def test_unknown_host_key_no_terminal(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch) -> None:
+        def eof(prompt: str) -> str:
+            raise EOFError
+
+        client.outcomes = [UserWarning('Unknown host key')]
+        monkeypatch.setattr('builtins.input', eof)
+
+        assert remote_backup()._ssh_connect() is None
+        assert client.closed
+
+    def test_bad_host_key(self, client: FakeClient, caplog: pytest.LogCaptureFixture) -> None:
+        class Key:
+            def get_base64(self) -> str:
+                return 'AAAA'
+
+        client.outcomes = [sb.paramiko.BadHostKeyException('server', Key(), Key())]
+
+        assert remote_backup()._ssh_connect() is None
+        assert client.closed
+        assert 'Can\'t connect to the server' in caplog.text
+
+    def test_password(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch) -> None:
+        client.outcomes = [ssh_error(), ssh_error(), None]
+        monkeypatch.setattr(sb, 'getpass', lambda prompt: 'secret')
+        monkeypatch.setenv('SSHPASS', '')
+        backup = remote_backup()
+
+        assert backup._ssh_connect() is client
+        assert client.connects[-1]['password'] == 'secret'
+        assert backup._password_auth
+        assert os.environ['SSHPASS'] == 'secret'
+
+    def test_wrong_password(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch) -> None:
+        client.outcomes = [ssh_error(), ssh_error(), ssh_error()]
+        monkeypatch.setattr(sb, 'getpass', lambda prompt: 'wrong')
+
+        assert remote_backup()._ssh_connect() is None
+        assert client.closed
+
+    def test_password_no_terminal(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch,
+                                  caplog: pytest.LogCaptureFixture) -> None:
+        def no_tty(prompt: str) -> str:
+            warnings.warn('Can not control echo on the terminal.', GetPassWarning)
+
+            return ''
+
+        client.outcomes = [ssh_error(), ssh_error()]
+        monkeypatch.setattr(sb, 'getpass', no_tty)
+
+        assert remote_backup()._ssh_connect() is None
+        assert 'Unable to get password' in caplog.text
+
+    @pytest.mark.parametrize('outcome, connected', [(None, True), ('error', False)])
+    def test_keyfile(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch,
+                     outcome: Optional[str], connected: bool) -> None:
+        pkey = object()
+        client.outcomes = [ssh_error(), ssh_error(), ssh_error() if outcome else None]
+        monkeypatch.setattr(sb.Backup, '_load_ssh_key', lambda self: pkey)
+
+        assert (remote_backup(ssh_keyfile='/keys/id')._ssh_connect() is client) == connected
+        assert client.connects[-1]['pkey'] is pkey
+
+    def test_keyfile_not_loaded(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch) -> None:
+        client.outcomes = [ssh_error(), ssh_error()]
+        monkeypatch.setattr(sb.Backup, '_load_ssh_key', lambda self: None)
+
+        assert remote_backup(ssh_keyfile='/keys/id')._ssh_connect() is None
+        assert len(client.connects) == 2
+        assert client.closed
+
+
+@pytest.fixture(scope='module')
+def keys(tmp_path_factory: pytest.TempPathFactory) -> Path:
+    """Real keys made by ssh-keygen; encrypted ones use the passphrase 'secret'"""
+
+    key_dir = tmp_path_factory.mktemp('keys')
+    key_types = {'ed25519': ['-t', 'ed25519'], 'ecdsa': ['-t', 'ecdsa'],
+                 'rsa': ['-t', 'rsa', '-b', '2048'], 'rsa_pem': ['-t', 'rsa', '-b', '2048', '-m', 'PEM']}
+
+    for name, args in key_types.items():
+        for suffix, passphrase in [('', ''), ('_enc', 'secret')]:
+            subprocess.run(['ssh-keygen', '-q', *args, '-N', passphrase, '-f', str(key_dir / f'{name}{suffix}')],
+                           check=True)
+
+    return key_dir
+
+
+@needs_paramiko
+@needs_ssh_keygen
+class TestLoadKey:
+    @pytest.fixture(autouse=True)
+    def prompts(self, monkeypatch: pytest.MonkeyPatch) -> list[str]:
+        asked: list[str] = []
+
+        def answer(prompt: str) -> str:
+            asked.append(prompt)
+
+            return self.passphrase
+
+        self.passphrase = 'secret'
+        monkeypatch.setattr(sb, 'getpass', answer)
+
+        return asked
+
+    @pytest.mark.parametrize('name', ['ed25519', 'ecdsa', 'rsa', 'rsa_pem'])
+    def test_unencrypted(self, keys: Path, prompts: list[str], name: str) -> None:
+        key = remote_backup(ssh_keyfile=str(keys / name))._load_ssh_key()
+
+        assert isinstance(key, sb.paramiko.PKey)
+        assert prompts == []
+
+    @pytest.mark.parametrize('name', ['ed25519_enc', 'ecdsa_enc', 'rsa_enc', 'rsa_pem_enc'])
+    def test_encrypted(self, keys: Path, prompts: list[str], name: str) -> None:
+        key = remote_backup(ssh_keyfile=str(keys / name))._load_ssh_key()
+
+        assert isinstance(key, sb.paramiko.PKey)
+        assert len(prompts) == 1
+
+    @pytest.mark.parametrize('name', ['ed25519_enc', 'rsa_pem_enc'])
+    def test_wrong_passphrase(self, keys: Path, caplog: pytest.LogCaptureFixture, name: str) -> None:
+        self.passphrase = 'wrong'
+
+        assert remote_backup(ssh_keyfile=str(keys / name))._load_ssh_key() is None
+        assert 'wrong passphrase?' in caplog.text
+
+    def test_no_terminal(self, keys: Path, monkeypatch: pytest.MonkeyPatch,
+                         caplog: pytest.LogCaptureFixture) -> None:
+        def no_tty(prompt: str) -> str:
+            warnings.warn('Can not control echo on the terminal.', GetPassWarning)
+
+            return ''
+
+        monkeypatch.setattr(sb, 'getpass', no_tty)
+
+        assert remote_backup(ssh_keyfile=str(keys / 'ed25519_enc'))._load_ssh_key() is None
+        assert 'Unable to get passphrase' in caplog.text
+
+    @pytest.mark.parametrize('content', [None, 'not a key\n'])
+    def test_bad_file(self, tmp_path: Path, caplog: pytest.LogCaptureFixture, content: Optional[str]) -> None:
+        keyfile = tmp_path / 'id'
+
+        if content is not None:
+            keyfile.write_text(content)
+
+        assert remote_backup(ssh_keyfile=str(keyfile))._load_ssh_key() is None
+        assert 'Unable to load SSH key' in caplog.text