Fuxino
8d08032b5d
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.
567 lines
20 KiB
Python
567 lines
20 KiB
Python
"""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
|