Find rsync in PATH and pass the SSH password only to rsync
A missing /usr/bin/rsync ended in a traceback, possibly after --remove-before-backup had removed old backups, and left SSHPASS in the environment.
This commit is contained in:
+13
-1
@@ -6,6 +6,7 @@ import os
|
||||
from collections.abc import Callable, Iterator
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from shutil import which
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
@@ -338,10 +339,21 @@ class TestReturnCodes:
|
||||
assert run_backup(out, [str(src)]) == 0
|
||||
|
||||
(args,) = fake_popen.calls
|
||||
assert args[0] == '/usr/bin/rsync'
|
||||
assert args[0] == which('rsync')
|
||||
assert f'--link-dest={out}/simple_backup/{old_backups[-1]}' in args
|
||||
assert args[args.index('/') + 1] == f'{out}/simple_backup/2026-10-14_10-00-00'
|
||||
|
||||
def test_rsync_missing(self, src: Path, out: Path, tmp_path: Path, fake_popen: type[FakePopen],
|
||||
old_backups: list[str], monkeypatch: pytest.MonkeyPatch,
|
||||
caplog: pytest.LogCaptureFixture) -> None:
|
||||
monkeypatch.setenv('PATH', str(tmp_path / 'empty'))
|
||||
|
||||
assert run_backup(out, [str(src)], keep=0, remove_before=True) == 4
|
||||
assert fake_popen.calls == []
|
||||
# Nothing was removed
|
||||
assert backups(out) == old_backups
|
||||
assert 'rsync not found' in caplog.text
|
||||
|
||||
@pytest.mark.parametrize('returncode, result, removed', [
|
||||
(0, 0, True),
|
||||
(24, 0, True),
|
||||
|
||||
+16
-6
@@ -254,9 +254,11 @@ class RecordPopen:
|
||||
|
||||
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')
|
||||
|
||||
@@ -276,6 +278,7 @@ class TestRsyncArgs:
|
||||
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()
|
||||
@@ -322,19 +325,26 @@ class TestRsyncArgs:
|
||||
|
||||
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')
|
||||
monkeypatch.delenv('SSHPASS', raising=False)
|
||||
backup = connected_backup()
|
||||
backup.inputs = [str(tmp_path / 'input')]
|
||||
backup._password_auth = True
|
||||
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']
|
||||
# The password must not stay in the environment
|
||||
# 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
|
||||
@@ -471,13 +481,13 @@ class TestLogin:
|
||||
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', '')
|
||||
monkeypatch.delenv('SSHPASS', raising=False)
|
||||
backup = remote_backup()
|
||||
|
||||
assert backup._ssh_connect() is client
|
||||
assert client.connects[-1]['password'] == 'secret'
|
||||
assert backup._password_auth
|
||||
assert os.environ['SSHPASS'] == '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()]
|
||||
|
||||
Reference in New Issue
Block a user