simple_backup/tests/test_remote.py

"""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

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: str | None = 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: list[str] | None) -> 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]] = []
    envs: list[dict[str, str] | None] = []

    def __init__(self, args: list[str], **kwargs: Any) -> None:
        RecordPopen.calls.append(args)
        RecordPopen.envs.append(kwargs.get('env'))
        self.stdin = io.BytesIO()
        self.stdout = io.BytesIO(b'sent 1 bytes  received 2 bytes\ntotal size is 3\n')

    def __enter__(self) -> 'RecordPopen':
        return self

    def __exit__(self, *args: Any) -> None:
        pass

    def wait(self) -> int:
        return RecordPopen.returncode


@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 = []
        RecordPopen.envs = []
        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.delenv('SSHPASS', raising=False)
        backup = connected_backup()
        backup.inputs = [str(tmp_path / 'input')]
        backup._password = 'secret'

        assert backup.run() == 0

        (args,) = RecordPopen.calls
        assert shlex.split(self.ssh_command(args)) == ['sshpass', '-e', 'ssh', '-l', 'user', '-o',
                                                       'StrictHostKeyChecking=accept-new']
        # Only rsync gets the password, never our own environment
        (env,) = RecordPopen.envs
        assert env is not None and env['SSHPASS'] == 'secret'
        assert 'SSHPASS' not in os.environ

    def test_no_password_env(self, tmp_path: Path) -> None:
        self.run(tmp_path)

        assert RecordPopen.envs == [None]

    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[BaseException | None]) -> None:
        self.outcomes = outcomes
        self.connects: list[dict[str, Any]] = []
        self.policies: list[str] = []
        self.known_hosts: str | None = 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({'hostname': hostname, **kwargs})
        outcome = self.outcomes.pop(0)

        if isinstance(outcome, UserWarning):
            # Like WarningPolicy for an unknown host key
            warnings.warn(outcome, stacklevel=2)
        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 client.connects[0]['hostname'] == 'server'
        assert 'port' not in client.connects[0]
        assert not client.closed

    def test_ssh_config(self, client: FakeClient, tmp_path: Path) -> None:
        (tmp_path / '.ssh').mkdir()
        key = tmp_path / '.ssh' / 'id_backup'
        key.touch()
        (tmp_path / '.ssh' / 'config').write_text(
            f'Host server\n  HostName backup.example.org\n  Port 2222\n'
            f'  IdentityFile {key}\n  IdentityFile {tmp_path}/.ssh/missing\n')
        client.outcomes = [ssh_error(), None]

        assert remote_backup()._ssh_connect(str(tmp_path)) is client

        for params in client.connects:
            assert (params['hostname'], params['port']) == ('backup.example.org', 2222)
            assert params['key_filename'] == [str(key)]

    def test_bad_ssh_config(self, client: FakeClient, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
        (tmp_path / '.ssh').mkdir()
        (tmp_path / '.ssh' / 'config').write_text('Host server\n  Port abc\n')
        client.outcomes = [None]

        assert remote_backup()._ssh_connect(str(tmp_path)) is client
        assert client.connects[0]['hostname'] == 'server'
        assert 'port' not in client.connects[0]
        assert 'Cannot read' in caplog.text

    @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.delenv('SSHPASS', raising=False)
        backup = remote_backup()

        assert backup._ssh_connect() is client
        assert client.connects[-1]['password'] == 'secret'
        assert backup._password == 'secret'
        assert 'SSHPASS' not in os.environ

    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, stacklevel=2)

            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: str | None, 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, stacklevel=2)

            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: str | None) -> 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