diff --git a/tests/test_remote.py b/tests/test_remote.py new file mode 100644 index 0000000..9e3e7fb --- /dev/null +++ b/tests/test_remote.py @@ -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