Bump version to 4.4.1

With --link-dest unchanged files are linked, not copied, so rsync's byte
percentage stayed near zero and the ETA was hours. Use the checked/total
file counts instead, and show no percentage or ETA while still scanning.

PKGBUILD

@@ -4,9 +4,9 @@
 
 pkgname=simple_backup
 pkgdesc='Simple backup script that uses rsync to copy files'
-pkgver=4.2.0
+pkgver=4.4.0
-pkgrel=1
+pkgrel=2
-url="https://git.shouldnt.work/fuxino/${pkgname}"
+url="https://git.shouldnt.work/fuxino/$pkgname"
 arch=('any')
 license=('GPL-3.0-or-later')
 makedepends=('git'
@@ -18,21 +18,31 @@ depends=('python>=3.10'
 optdepends=('python-systemd: use systemd log'
             'python-dbus: for desktop notifications'
             'python-paramiko: for remote backup through ssh')
+checkdepends=('python-pytest'
+              'python-paramiko'
+              'openssh')
 conflicts=('simple_backup-git')
-source=(git+${url}?signed#tag=${pkgver})
+source=("git+$url?signed#tag=$pkgver")
 validpgpkeys=('7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514')
-sha256sums=('6c674f7981e3e5222594bdb735d4fb02ca2fcc829de57ea8f92ce71e679853f9')
+sha256sums=('cc5944558ed4bc30a0d1a00b49dfced371909dd6c4f39b0c1ccf8f39b445e907')
 
 build()
 {
-    cd ${srcdir}/${pkgname}
+    cd "$srcdir"/"$pkgname"
     python -m build --wheel --no-isolation
 }
 
+check()
+{
+    cd "$srcdir"/"$pkgname"
+    python -m pytest
+}
+
 package()
 {
-    cd ${srcdir}/${pkgname}
+    cd "$srcdir"/"$pkgname"
-    python -m installer --destdir=${pkgdir} dist/*.whl
+    python -m installer --destdir="$pkgdir" dist/*.whl
-    install -Dm644 ${pkgname}/${pkgname}.conf ${pkgdir}/usr/share/doc/${pkgname}/${pkgname}.conf
+    install -Dm644 "$pkgname"/"$pkgname".conf "$pkgdir"/usr/share/doc/"$pkgname"/"$pkgname".conf
-    install -Dm644 man/${pkgname}.1 ${pkgdir}/usr/share/man/man1/${pkgname}.1
+    install -Dm644 README.md "$pkgdir"/usr/share/doc/"$pkgname"/README.md
+    install -Dm644 man/"$pkgname".1 "$pkgdir"/usr/share/man/man1/"$pkgname".1
 }

README.md

@@ -20,32 +20,39 @@ The script uses rsync to actually run the backup, so you will have to install it
 sudo pacman -Syu rsync
 ```
 
-Optional dependencies are systemd-python to enable using systemd journal for logging, dbus-python for desktop notifications, and paramiko for remote backup.
+Optional dependencies are systemd-python to enable using systemd journal for logging, dbus-python for desktop notifications, and paramiko (version 3.2 or later) for remote backup.
 
 ## Install
-To install the program, first clone the repository:
+For Arch Linux and Arch-based distros, two packages are available in the AUR (aur.archlinux.org):
+- **simple_backup** for the release version
+- **simple_backup-git** for the git version
+
+To install it from source, first clone the repository:
 
 ```bash
 git clone https://git.shouldnt.work/fuxino/simple_backup.git
 ```
 
-Then install the tools required to build the package:
+Then build the package (this requires the Python `build` module, e.g. the python-build package on Arch Linux):
-
-```bash
-pip install --upgrade build
-```
-
-Finally, run:
 
 ```bash
 cd simple_backup
 python -m build --wheel
-python -m pip install dist/*.whl
 ```
 
-For Arch Linux and Arch-based distros, two packages are available in the AUR (aur.archlinux.org):
+Many distributions don't allow installing packages with pip into the system Python ("externally-managed-environment" error).
-- **simple_backup** for the release version
+In that case, install the wheel in a virtual environment, for example with pipx:
-- **simple_backup-git** for the git version
+
+```bash
+pipx install dist/*.whl
+```
+
+Optional dependencies can be installed as extras, for example `pipx install 'dist/simple_backup-<version>-py3-none-any.whl[REMOTE]'` (available extras: JOURNAL, NOTIFICATIONS, REMOTE).
+
+## Running as root
+To back up files that your user can't read, run simple_backup with sudo or doas.
+In that case the user who ran the command is detected automatically, so that '~' (in the inputs and in the default configuration file path) refers to that user's home directory, not root's.
+When running as root in other ways (e.g. from a systemd service), specify the user with the --user (or -u) argument.
 
 ## Old backups
 The number of old backups to keep can be set with the --keep (or -k) argument (or in the configuration file).
@@ -57,8 +64,16 @@ The only rsync error that is ignored is return code 24 (source files vanished du
 > This feature is experimental
 
 It's possible to use a remote server as destination for the backup. Just use the --ssh-user and --ssh-host arguments (or set them in the configuration file).
+Both are required: if only one of them is set, simple_backup exits with an error.
 For this to work, rsync must be installed on the server too.
 
+The output directory is a path on the server. A leading '~' and relative paths refer to the user's home directory on the server.
+On the command line, quote '~' so that your local shell doesn't expand it:
+
+```bash
+simple_backup --ssh-host server --ssh-user user -o '~/backups'
+```
+
 ### Server authentication
 The best way to handle the authentication is to have an ssh agent running on your system, otherwise if a passphrase is necessary to unlock the ssh key, it will be necessary to enter it more than once.
 If needed, it's possible to specify the ssh key location with the --keyfile argument or in the configuration file.

man/simple_backup.1

@@ -1,4 +1,4 @@
-.TH SIMPLE_BACKUP 1 2026-10-08 SIMPLE_BACKUP 4.3.0
+.TH SIMPLE_BACKUP 1 2026-10-15 SIMPLE_BACKUP 4.4.1
 .SH NAME
 simple_backup \- Backup files and folders using rsync
 .SH SYNOPSIS
@@ -105,6 +105,12 @@ according to \-\-keep will be left, and the new backup may be incomplete. In par
 .B \-\-no\-syslog
 Don't use systemd journal for logging.
 .TP
+.B \-\-no\-progress
+Don't show a progress bar while rsync copies the files. The progress bar is only shown when
+the output is a terminal, so it never appears in the systemd journal or in redirected output.
+While rsync is still looking for files, the bar shows \(aq(scanning)\(aq, and the percentage
+and estimated time are only approximate.
+.TP
 .B \-\-rsync\-options OPTIONS [OPTION...]
 By default, the following rsync options are used:
 .RS

pyproject.toml

@@ -46,3 +46,18 @@ packages = ["simple_backup"]
 
 [tool.setuptools.dynamic]
 version = { attr = "simple_backup.__version__" }
+
+[tool.mypy]
+files = ["simple_backup", "tests"]
+disallow_untyped_defs = true
+warn_unused_ignores = true
+
+# No usable stubs (types-paramiko lags behind paramiko 5)
+[[tool.mypy.overrides]]
+module = ["paramiko", "systemd", "dbus"]
+ignore_missing_imports = true
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+# Import the repo copy, not an installed one
+pythonpath = ["."]

simple_backup/__init__.py

@@ -1,3 +1,3 @@
 """Init."""
 
-__version__ = '4.3.0'
+__version__ = '4.4.1'

simple_backup/simple_backup.py

@@ -18,10 +18,12 @@ from __future__ import annotations
 import sys
 import os
 import pwd
-from typing import Callable, List, Optional, ParamSpec, TypeVar, Union
+import io
+import re
+from typing import Callable, Iterator, List, Optional, ParamSpec, TypedDict, TypeVar
 import warnings
 from functools import wraps
-from shutil import rmtree, which
+from shutil import get_terminal_size, rmtree, which
 import shlex
 import argparse
 import configparser
@@ -72,6 +74,10 @@ R = TypeVar('R')
 # Seconds to wait for the TCP connection to the SSH server
 SSH_TIMEOUT = 15
 
+# An rsync --info=progress2 update: size, percent, speed, time, then optionally (xfr#N, ir-chk/to-chk=N/N)
+PROGRESS_RE = re.compile(r'\s*(\S+)\s+\d+%\s+\S+\s+\d+:\d\d:\d\d(?:\s+\(xfr#\d+, (ir|to)-chk=(\d+)/(\d+)\))?\s*')
+PROGRESS_BAR_WIDTH = 30
+
 
 def timing(func: Callable[P, R]) -> Callable[P, R]:
     """Decorator to measure execution time of a function
@@ -98,6 +104,66 @@ class MyFormatter(argparse.RawTextHelpFormatter, argparse.ArgumentDefaultsHelpFo
     """Custom format for argparse help text"""
 
 
+class _ProgressBar:
+    """Draws rsync --info=progress2 updates as a bar on stderr"""
+
+    def __init__(self) -> None:
+        self._start = default_timer()
+        # rsync reports ir-chk while incremental recursion is still finding files
+        self._scanning = True
+        self._checked = 0
+        self._total = 0
+        self._shown = False
+
+    def update(self, record: str) -> bool:
+        """Draw the bar if record is a progress update, and return whether it was"""
+
+        m = PROGRESS_RE.fullmatch(record)
+
+        if m is None:
+            return False
+
+        size, chk = m.group(1), m.group(2)
+
+        # Records in the middle of a file have no counts, so keep the last ones
+        if chk is not None:
+            self._scanning = chk == 'ir'
+            self._total = int(m.group(4))
+            self._checked = self._total - int(m.group(3))
+
+        # Count files, not bytes: with --link-dest unchanged files are linked, and rsync's percentage leaves them out
+        if self._scanning or self._total == 0:
+            text = f'{self._checked:,} files checked  {size} copied  (scanning)'
+            percent = None
+        else:
+            percent = self._checked * 100 // self._total
+            text = f'{percent:3d}%  {self._checked:,}/{self._total:,} files  {size} copied'
+
+            if 0 < percent < 100:
+                eta = int((default_timer() - self._start) * (self._total - self._checked) / self._checked)
+                text += f'  ETA {eta // 3600}:{eta // 60 % 60:02d}:{eta % 60:02d}'
+
+        width = get_terminal_size().columns - 1
+
+        # Fixed width, so the bar doesn't jump when the text changes length; no bar until the total is known
+        if percent is not None and len(text) + PROGRESS_BAR_WIDTH + 4 <= width:
+            filled = PROGRESS_BAR_WIDTH * percent // 100
+            text = f'[{"#" * filled}{"-" * (PROGRESS_BAR_WIDTH - filled)}]  {text}'
+
+        sys.stderr.write(f'\r{text[:width]}\033[K')
+        sys.stderr.flush()
+        self._shown = True
+
+        return True
+
+    def clear(self) -> None:
+        """Remove the bar, so that log messages start on a clean line"""
+
+        if self._shown:
+            sys.stderr.write('\r\033[K')
+            sys.stderr.flush()
+
+
 class Backup:
     """Main class defining parameters and functions for performing backup
 
@@ -122,6 +188,8 @@ class Backup:
             Run remote rsync with sudo
         remove_before: bool
             Indicate if removing old backups will be performed before copying files
+        progress: bool
+            Show a progress bar while rsync runs
 
     Methods:
         check_params():
@@ -136,40 +204,44 @@ class Backup:
             Perform the backup
     """
 
-    def __init__(self, inputs: List[str], output: str, exclude: List[str], keep: int, options: str,
+    def __init__(self, inputs: Optional[List[str]], output: Optional[str], exclude: Optional[List[str]],
+                 keep: Optional[int], options: str,
                  ssh_host: Optional[str] = None, ssh_user: Optional[str] = None, ssh_keyfile: Optional[str] = None,
-                 remote_sudo: bool = False, remove_before: bool = False, verbose: bool = False) -> None:
+                 remote_sudo: bool = False, remove_before: bool = False, verbose: bool = False,
-        self.inputs = inputs
+                 progress: bool = False) -> None:
-        self.output = output
+        # Missing values are reported by check_params()
-        self.exclude = exclude
+        self.inputs = inputs if inputs is not None else []
+        self.output = output if output is not None else ''
+        self.exclude = exclude if exclude is not None else []
         self.options = options
-        self.keep = keep
+        self.keep = keep if keep is not None else -1
         self.ssh_host = ssh_host
         self.ssh_user = ssh_user
         self.ssh_keyfile = ssh_keyfile
         self.remote_sudo = remote_sudo
         self._remove_before = remove_before
         self._verbose = verbose
+        self._progress = progress
         self._last_backup = ''
         self._server = ''
         self._output_dir = ''
         self._inputs_path = ''
         self._exclude_path = ''
         self._remote = False
-        self._ssh = None
+        self._ssh: Optional[paramiko.client.SSHClient] = None
         self._password_auth = False
         self._removed_count = 0
 
     def check_params(self, homedir: str = '') -> int:
         """Check if parameters for the backup are valid"""
 
-        if self.inputs is None or len(self.inputs) == 0:
+        if len(self.inputs) == 0:
             logger.info(
                 'No existing files or directories specified for backup. Nothing to do')
 
             return 1
 
-        if self.output is None:
+        if self.output == '':
             logger.critical(
                 'No output path specified. Use -o argument or specify output path in configuration file')
 
@@ -207,9 +279,6 @@ class Backup:
 
             self.output = os.path.abspath(self.output)
 
-        if self.keep is None:
-            self.keep = -1
-
         return 0
 
     def _resolve_remote_path(self, path: str) -> Optional[str]:
@@ -279,7 +348,11 @@ class Backup:
                 logger.info('Removing old backups...')
                 dirs.sort()
 
-                for i in range(n_backup - self.keep):
+                n_remove = n_backup - self.keep
+
+                for i in range(n_remove):
+                    logger.info('Removing backup %s (%d/%d)...', dirs[i], i + 1, n_remove)
+
                     if self.remote_sudo:
                         _, _, stderr = self._ssh.exec_command(
                             f'sudo rm -r {shlex.quote(f"{self.output}/simple_backup/{dirs[i]}")}')
@@ -318,7 +391,11 @@ class Backup:
                 logger.info('Removing old backups...')
                 dirs.sort()
 
-                for i in range(n_backup - self.keep):
+                n_remove = n_backup - self.keep
+
+                for i in range(n_remove):
+                    logger.info('Removing backup %s (%d/%d)...', dirs[i], i + 1, n_remove)
+
                     try:
                         rmtree(f'{self.output}/simple_backup/{dirs[i]}')
                         count += 1
@@ -605,7 +682,6 @@ class Backup:
         fd, self._exclude_path = mkstemp(prefix='tmp_exclude', text=True)
 
         with os.fdopen(fd, 'w', encoding='utf-8') as fp:
-            if self.exclude is not None:
             for e in self.exclude:
                 fp.write(e)
                 fp.write('\n')
@@ -630,7 +706,7 @@ class Backup:
         # rsync splits the -e command itself, so quote the parts that may contain spaces
         if self.ssh_keyfile is not None:
             ssh = f'ssh -i {shlex.quote(self.ssh_keyfile)} -o StrictHostKeyChecking=accept-new'
-        elif self._password_auth and which('sshpass'):
+        elif self._password_auth and self.ssh_user is not None and which('sshpass'):
             ssh = f'sshpass -e ssh -l {shlex.quote(self.ssh_user)} -o StrictHostKeyChecking=accept-new'
         else:
             ssh = 'ssh -o StrictHostKeyChecking=accept-new'
@@ -640,25 +716,39 @@ class Backup:
         if self._remote and self.remote_sudo:
             args.append('--rsync-path=sudo rsync')
 
+        bar = None
+
+        if self._progress:
+            args.append('--info=progress2')
+            bar = _ProgressBar()
+
         with Popen(args, stdin=PIPE, stdout=PIPE, stderr=STDOUT, shell=False) as p:
-            output: Union[bytes, List[str]]
+            assert p.stdin is not None and isinstance(p.stdout, io.BufferedIOBase)
-            output, _ = p.communicate()
+            p.stdin.close()
+
+            output = []
+
+            try:
+                for r in _read_records(p.stdout):
+                    if r != '' and (bar is None or not bar.update(r)):
+                        output.append(r)
+            finally:
+                if bar is not None:
+                    bar.clear()
 
             try:
                 del os.environ['SSHPASS']
             except KeyError:
                 pass
 
-            returncode = p.returncode
+            returncode = p.wait()
-
-        output = output.decode("utf-8").split('\n')
 
         if returncode == 0:
             if self._verbose:
                 logger.info('rsync: %s', output)
             else:
-                logger.info('rsync: %s', output[-3])
                 logger.info('rsync: %s', output[-2])
+                logger.info('rsync: %s', output[-1])
         else:
             self._returncode_log(returncode)
 
@@ -703,6 +793,21 @@ class Backup:
         return 0
 
 
+def _read_records(stream: io.BufferedIOBase) -> Iterator[str]:
+    """Yield rsync output as it arrives, split on '\\n' and '\\r' (which ends progress updates)"""
+
+    buffer = b''
+
+    while chunk := stream.read1(65536):
+        *records, buffer = re.split(rb'[\r\n]', buffer + chunk)
+
+        for r in records:
+            yield r.decode('utf-8', errors='replace')
+
+    if buffer != b'':
+        yield buffer.decode('utf-8', errors='replace')
+
+
 def _detect_user() -> Optional[str]:
     if os.geteuid() == 0:
         return os.getenv('SUDO_USER') or os.getenv('DOAS_USER')
@@ -743,6 +848,8 @@ def _parse_arguments() -> argparse.Namespace:
                         help='Remove old backups before executing the backup, instead of after')
     parser.add_argument('--no-syslog', action='store_true',
                         help='Disable systemd journal logging')
+    parser.add_argument('--no-progress', action='store_true',
+                        help='Don\'t show a progress bar (it is only shown on a terminal)')
     parser.add_argument('--rsync-options', nargs='+',
                         choices=['a', 'l', 'p', 't', 'g', 'o',
                                  'c', 'h', 's', 'D', 'H', 'X'],
@@ -766,8 +873,8 @@ def _expand_home(path: str, user: str) -> str:
     return os.path.expanduser(path)
 
 
-def _expand_inputs(inputs, user: Optional[str] = None):
+def _expand_inputs(inputs: List[str], user: Optional[str] = None) -> List[str]:
-    expanded_inputs = []
+    expanded_inputs: List[str] = []
 
     for i in inputs:
         if i == '':
@@ -790,8 +897,30 @@ def _expand_inputs(inputs, user: Optional[str] = None):
     return expanded_inputs
 
 
-def _read_config(config_file, user: Optional[str] = None):
+class ConfigArgs(TypedDict):
-    config_args = {'inputs': None,
+    """Values read from the configuration file"""
+
+    inputs: Optional[List[str]]
+    output: Optional[str]
+    exclude: Optional[List[str]]
+    keep: int
+    ssh_host: Optional[str]
+    ssh_user: Optional[str]
+    ssh_keyfile: Optional[str]
+    remote_sudo: bool
+    numeric_ids: bool
+
+
+def _get_bool(config: configparser.ConfigParser, key: str) -> bool:
+    # Empty values (as in the example config) count as false
+    if config.get('server', key, fallback='').strip() == '':
+        return False
+
+    return config.getboolean('server', key)
+
+
+def _read_config(config_file: str, user: Optional[str] = None) -> ConfigArgs:
+    config_args: ConfigArgs = {'inputs': None,
                                'output': None,
                                'exclude': None,
                                'keep': -1,
@@ -823,29 +952,15 @@ def _read_config(config_file, user: Optional[str] = None):
         section = 'default'
         inputs = config.get(section, 'inputs')
 
-    inputs = [i.strip() for i in inputs.split(',')]
+    config_args['inputs'] = list(set(_expand_inputs([i.strip() for i in inputs.split(',')], user)))
-    inputs = _expand_inputs(inputs, user)
-    inputs = list(set(inputs))
-
-    config_args['inputs'] = inputs
 
     # '~' is expanded later, since for remote backups it refers to the home on the server
     config_args['output'] = config.get(section, 'backup_dir')
 
-    try:
+    exclude = config.get(section, 'exclude', fallback='')
-        exclude = config.get(section, 'exclude')
+    config_args['exclude'] = [e.strip() for e in exclude.split(',') if e.strip() != '']
-        exclude = [e.strip() for e in exclude.split(',') if e.strip() != '']
-    except configparser.NoOptionError:
-        exclude = []
 
-    config_args['exclude'] = exclude
+    config_args['keep'] = config.getint(section, 'keep', fallback=-1)
-
-    try:
-        keep = config.getint(section, 'keep')
-    except configparser.NoOptionError:
-        keep = -1
-
-    config_args['keep'] = keep
 
     # Read host and user separately, so that a missing one is reported instead of silently
     # disabling remote backup. Empty values count as not set
@@ -854,12 +969,8 @@ def _read_config(config_file, user: Optional[str] = None):
 
     config_args['ssh_keyfile'] = config.get('server', 'ssh_keyfile', fallback='').strip() or None
 
-    # Empty values (as in the example config) count as false
+    config_args['remote_sudo'] = _get_bool(config, 'remote_sudo')
-    for key in ['remote_sudo', 'numeric_ids']:
+    config_args['numeric_ids'] = _get_bool(config, 'numeric_ids')
-        if config.get('server', key, fallback='').strip() == '':
-            config_args[key] = False
-        else:
-            config_args[key] = config.getboolean('server', key)
 
     return config_args
 
@@ -1011,10 +1122,9 @@ def simple_backup() -> int:
     if args.numeric_ids or config_args['numeric_ids']:
         rsync_options.append('--numeric-ids')
 
-    rsync_options = ' '.join(rsync_options)
+    backup = Backup(inputs, output, exclude, keep, ' '.join(rsync_options), ssh_host, ssh_user, ssh_keyfile,
-
+                    remote_sudo, remove_before=args.remove_before_backup, verbose=args.verbose,
-    backup = Backup(inputs, output, exclude, keep, rsync_options, ssh_host, ssh_user, ssh_keyfile,
+                    progress=not args.no_progress and sys.stderr.isatty())
-                    remote_sudo, remove_before=args.remove_before_backup, verbose=args.verbose)
 
     try:
         return_code = backup.check_params(homedir)

tests/conftest.py

@@ -0,0 +1,30 @@
+"""Shared fixtures for the simple_backup tests"""
+
+import sys
+from typing import Callable
+
+import pytest
+
+from simple_backup import simple_backup as sb
+
+
+@pytest.fixture(autouse=True)
+def no_side_effects(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Keep tests out of the systemd journal and the desktop notifications"""
+
+    if hasattr(sb, 'j_handler'):
+        sb.logger.removeHandler(sb.j_handler)
+
+    monkeypatch.setattr(sb, '_notify', lambda text: None)
+
+
+@pytest.fixture
+def run_cli(monkeypatch: pytest.MonkeyPatch) -> Callable[..., int]:
+    """Run simple_backup() with the given command line arguments"""
+
+    def run(*argv: str) -> int:
+        monkeypatch.setattr(sys, 'argv', ['simple_backup', *argv])
+
+        return sb.simple_backup()
+
+    return run

tests/test_cli.py

@@ -0,0 +1,331 @@
+"""Tests for combining command line and config values, and for detecting the user"""
+
+import os
+import pwd
+from pathlib import Path
+from typing import Any, Callable, Optional
+
+import pytest
+
+from simple_backup import simple_backup as sb
+
+USER = pwd.getpwuid(os.getuid()).pw_name
+HOME = pwd.getpwuid(os.getuid()).pw_dir
+UID = os.getuid()
+
+DEFAULT_OPTIONS = '-r -v --ignore-missing-args --mkpath -a -h -H -X -s'
+
+RunCli = Callable[..., int]
+
+
+class FakeBackup:
+    """Records the values simple_backup() passes to Backup"""
+
+    instances: list['FakeBackup'] = []
+    check_code = 0
+
+    def __init__(self, inputs: Any, output: Any, exclude: Any, keep: Any, options: str,
+                 ssh_host: Optional[str] = None, ssh_user: Optional[str] = None,
+                 ssh_keyfile: Optional[str] = None, remote_sudo: bool = False,
+                 remove_before: bool = False, verbose: bool = False, progress: bool = False) -> None:
+        self.inputs = inputs
+        self.output = output
+        self.exclude = exclude
+        self.keep = keep
+        self.options = options
+        self.ssh_host = ssh_host
+        self.ssh_user = ssh_user
+        self.ssh_keyfile = ssh_keyfile
+        self.remote_sudo = remote_sudo
+        self.remove_before = remove_before
+        self.verbose = verbose
+        self.progress = progress
+        self.homedir: Optional[str] = None
+        self.ran = False
+        self.closed = False
+        FakeBackup.instances.append(self)
+
+    def check_params(self, homedir: str = '') -> int:
+        self.homedir = homedir
+
+        return FakeBackup.check_code
+
+    def run(self) -> int:
+        self.ran = True
+
+        return 0
+
+    def close(self) -> None:
+        self.closed = True
+
+
+@pytest.fixture(autouse=True)
+def fake_backup(monkeypatch: pytest.MonkeyPatch) -> None:
+    FakeBackup.instances = []
+    FakeBackup.check_code = 0
+    monkeypatch.setattr(sb, 'Backup', FakeBackup)
+
+
+@pytest.fixture
+def as_user(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Run as the current (non-root) user"""
+
+    monkeypatch.setattr(os, 'geteuid', lambda: UID if UID != 0 else 1000)
+    monkeypatch.setenv('USER', USER)
+    monkeypatch.setenv('HOME', HOME)
+
+
+@pytest.fixture
+def as_root(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Run as root, with no sudo/doas variables set"""
+
+    monkeypatch.setattr(os, 'geteuid', lambda: 0)
+
+    for var in ['SUDO_USER', 'SUDO_UID', 'DOAS_USER']:
+        monkeypatch.delenv(var, raising=False)
+
+
+@pytest.fixture
+def config_calls(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, Optional[str]]]:
+    """Record the config file path and user instead of reading the file"""
+
+    calls: list[tuple[str, Optional[str]]] = []
+    read_config = sb._read_config
+
+    def record(config_file: str, user: Optional[str] = None) -> sb.ConfigArgs:
+        calls.append((config_file, user))
+
+        return read_config('/nonexistent/simple_backup.conf', user)
+
+    monkeypatch.setattr(sb, '_read_config', record)
+
+    return calls
+
+
+@pytest.fixture
+def config(tmp_path: Path) -> Callable[[str], str]:
+    """Write a config file with existing inputs, followed by the given text"""
+
+    (tmp_path / 'data').mkdir()
+
+    def write(text: str = '') -> str:
+        config_file = tmp_path / 'simple_backup.conf'
+        config_file.write_text(f'[backup]\ninputs={tmp_path}/data\n{text}\n')
+
+        return str(config_file)
+
+    return write
+
+
+def last_backup() -> FakeBackup:
+    assert len(FakeBackup.instances) == 1
+
+    return FakeBackup.instances[0]
+
+
+@pytest.mark.usefixtures('as_user')
+class TestValues:
+    def test_config_values(self, run_cli: RunCli, config: Callable[[str], str], tmp_path: Path) -> None:
+        config_file = config("""backup_dir=/backups
+exclude=*.bak
+keep=3
+[server]
+ssh_host=server
+ssh_user=user
+ssh_keyfile=/keys/id
+remote_sudo=true
+numeric_ids=true
+""")
+
+        assert run_cli('-c', config_file) == 0
+
+        b = last_backup()
+        assert b.inputs == [f'{tmp_path}/data']
+        assert b.output == '/backups'
+        assert b.exclude == ['*.bak']
+        assert b.keep == 3
+        assert (b.ssh_host, b.ssh_user, b.ssh_keyfile) == ('server', 'user', '/keys/id')
+        assert b.remote_sudo is True
+        assert b.options == f'{DEFAULT_OPTIONS} --numeric-ids'
+        assert b.ran and b.closed
+
+    def test_cli_overrides_config(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
+        config_file = config("""backup_dir=/backups
+exclude=*.bak
+keep=3
+[server]
+ssh_host=server
+ssh_user=user
+ssh_keyfile=/keys/id
+""")
+
+        assert run_cli('-c', config_file, '-i', '/cli/in', '-o', '/cli/out', '-e', '*.tmp', '-k', '0',
+                       '--ssh-host', 'cli-host', '--ssh-user', 'cli-user', '--keyfile', '/cli/key') == 0
+
+        b = last_backup()
+        assert b.inputs == ['/cli/in']
+        assert b.output == '/cli/out'
+        assert b.exclude == ['*.tmp']
+        assert b.keep == 0
+        assert (b.ssh_host, b.ssh_user, b.ssh_keyfile) == ('cli-host', 'cli-user', '/cli/key')
+
+    def test_ssh_host_and_user_from_cli_and_config(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
+        config_file = config('backup_dir=/backups\n[server]\nssh_user=user')
+
+        assert run_cli('-c', config_file, '--ssh-host', 'server') == 0
+        assert (last_backup().ssh_host, last_backup().ssh_user) == ('server', 'user')
+
+    @pytest.mark.parametrize('server, argv', [
+        ('ssh_host=server', []),
+        ('ssh_user=user', []),
+        ('', ['--ssh-host', 'server']),
+        ('', ['--ssh-user', 'user']),
+    ])
+    def test_only_one_of_ssh_host_and_user(self, run_cli: RunCli, config: Callable[[str], str],
+                                           caplog: pytest.LogCaptureFixture, server: str,
+                                           argv: list[str]) -> None:
+        config_file = config(f'backup_dir=/backups\n[server]\n{server}')
+
+        assert run_cli('-c', config_file, *argv) == 6
+        assert FakeBackup.instances == []
+        assert 'Both SSH host and SSH user are needed' in caplog.text
+
+    def test_bad_config(self, run_cli: RunCli, config: Callable[[str], str],
+                        caplog: pytest.LogCaptureFixture) -> None:
+        config_file = config('backup_dir=/backups\nkeep=abc')
+
+        assert run_cli('-c', config_file) == 6
+        assert FakeBackup.instances == []
+        assert 'Bad configuration file' in caplog.text
+
+    def test_local_output_expanded(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
+        assert run_cli('-c', config('backup_dir=~/backups')) == 0
+        assert last_backup().output == f'{HOME}/backups'
+
+    def test_remote_output_not_expanded(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
+        config_file = config('backup_dir=~/backups\n[server]\nssh_host=server\nssh_user=user')
+
+        assert run_cli('-c', config_file) == 0
+        # Expanded on the server
+        assert last_backup().output == '~/backups'
+
+    @pytest.mark.parametrize('argv, options', [
+        ([], DEFAULT_OPTIONS),
+        (['--rsync-options', 'a', 'l'], '-r -v --ignore-missing-args --mkpath -a -l'),
+        (['-s', '-z', '--numeric-ids'], f'{DEFAULT_OPTIONS} -c -z --numeric-ids'),
+    ])
+    def test_rsync_options(self, run_cli: RunCli, config: Callable[[str], str], argv: list[str],
+                           options: str) -> None:
+        assert run_cli('-c', config('backup_dir=/backups'), *argv) == 0
+        assert last_backup().options == options
+
+    def test_flags(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
+        assert run_cli('-c', config('backup_dir=/backups'), '--remote-sudo', '--remove-before-backup', '-v') == 0
+
+        b = last_backup()
+        assert b.remote_sudo and b.remove_before and b.verbose
+
+    @pytest.mark.parametrize('tty, argv, progress', [
+        (True, [], True),
+        (True, ['--no-progress'], False),
+        (False, [], False),
+    ])
+    def test_progress(self, run_cli: RunCli, config: Callable[[str], str], monkeypatch: pytest.MonkeyPatch,
+                      tty: bool, argv: list[str], progress: bool) -> None:
+        monkeypatch.setattr(sb.sys.stderr, 'isatty', lambda: tty)
+
+        assert run_cli('-c', config('backup_dir=/backups'), *argv) == 0
+        assert last_backup().progress is progress
+
+    def test_check_params_failure(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
+        FakeBackup.check_code = 2
+
+        assert run_cli('-c', config('backup_dir=/backups')) == 2
+
+        b = last_backup()
+        assert not b.ran
+        assert b.closed
+
+
+class TestUser:
+    @pytest.mark.usefixtures('as_user')
+    def test_current_user(self, run_cli: RunCli, config_calls: list[tuple[str, Optional[str]]]) -> None:
+        assert run_cli() == 0
+        assert config_calls == [(f'{HOME}/.config/simple_backup/simple_backup.conf', USER)]
+        assert last_backup().homedir == HOME
+
+    @pytest.mark.usefixtures('as_root')
+    def test_explicit_user(self, run_cli: RunCli, config_calls: list[tuple[str, Optional[str]]]) -> None:
+        assert run_cli('-u', USER, '-c', '~/other.conf') == 0
+        assert config_calls == [(f'{HOME}/other.conf', USER)]
+        assert last_backup().homedir == HOME
+
+    @pytest.mark.usefixtures('as_root')
+    @pytest.mark.parametrize('var', ['SUDO_USER', 'DOAS_USER'])
+    def test_sudo_or_doas_user(self, run_cli: RunCli, config_calls: list[tuple[str, Optional[str]]],
+                               monkeypatch: pytest.MonkeyPatch, var: str) -> None:
+        monkeypatch.setenv(var, USER)
+
+        assert run_cli() == 0
+        assert config_calls == [(f'{HOME}/.config/simple_backup/simple_backup.conf', USER)]
+        assert last_backup().homedir == HOME
+
+    @pytest.mark.usefixtures('as_root')
+    def test_root_without_user(self, run_cli: RunCli, config_calls: list[tuple[str, Optional[str]]],
+                               caplog: pytest.LogCaptureFixture) -> None:
+        assert run_cli() == 0
+        # '~' is left to expanduser, i.e. $HOME
+        assert config_calls[0][1] is None
+        assert last_backup().homedir == ''
+        assert 'Failed to detect user' in caplog.text
+
+
+class TestDetectUser:
+    @pytest.mark.usefixtures('as_root')
+    def test_root_prefers_sudo_user(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv('SUDO_USER', 'alice')
+        monkeypatch.setenv('DOAS_USER', 'bob')
+
+        assert sb._detect_user() == 'alice'
+
+    @pytest.mark.usefixtures('as_root')
+    def test_root_doas_user(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv('DOAS_USER', 'bob')
+
+        assert sb._detect_user() == 'bob'
+
+    @pytest.mark.usefixtures('as_root')
+    def test_root_no_user(self) -> None:
+        assert sb._detect_user() is None
+
+    @pytest.mark.usefixtures('as_user')
+    def test_non_root(self) -> None:
+        assert sb._detect_user() == USER
+
+
+class TestTargetUid:
+    @pytest.mark.usefixtures('as_user')
+    def test_non_root(self) -> None:
+        assert sb._target_uid() == os.geteuid()
+
+    @pytest.mark.usefixtures('as_root')
+    def test_sudo_uid(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv('SUDO_UID', '1234')
+
+        assert sb._target_uid() == 1234
+
+    @pytest.mark.usefixtures('as_root')
+    def test_doas_user(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv('DOAS_USER', USER)
+
+        assert sb._target_uid() == UID
+
+    @pytest.mark.usefixtures('as_root')
+    def test_unknown_doas_user(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv('DOAS_USER', 'no-such-user-simple-backup')
+
+        assert sb._target_uid() is None
+
+    @pytest.mark.usefixtures('as_root')
+    def test_no_user(self) -> None:
+        assert sb._target_uid() is None

tests/test_config.py

@@ -0,0 +1,230 @@
+"""Tests for reading the configuration file and expanding paths"""
+
+import configparser
+import os
+import pwd
+from pathlib import Path
+
+import pytest
+
+from simple_backup import simple_backup as sb
+
+USER = pwd.getpwuid(os.getuid()).pw_name
+HOME = pwd.getpwuid(os.getuid()).pw_dir
+
+
+def write_config(path: Path, text: str) -> str:
+    config_file = path / 'simple_backup.conf'
+    config_file.write_text(text)
+
+    return str(config_file)
+
+
+def test_repo_copy_imported() -> None:
+    repo = Path(__file__).resolve().parents[1]
+
+    assert Path(sb.__file__).resolve().is_relative_to(repo)
+
+
+@pytest.mark.parametrize('path, expected', [
+    ('~', HOME),
+    ('~/docs', f'{HOME}/docs'),
+    ('/abs/~/path', '/abs/~/path'),
+    ('relative', 'relative'),
+])
+def test_expand_home(path: str, expected: str) -> None:
+    assert sb._expand_home(path, USER) == expected
+
+
+def test_expand_inputs_glob(tmp_path: Path) -> None:
+    for name in ['a.txt', 'b.txt', 'c.log']:
+        (tmp_path / name).touch()
+
+    result = sb._expand_inputs([f'{tmp_path}/*.txt', ''], USER)
+
+    assert sorted(result) == [f'{tmp_path}/a.txt', f'{tmp_path}/b.txt']
+
+
+def test_expand_inputs_no_match(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
+    assert sb._expand_inputs([f'{tmp_path}/missing'], USER) == []
+    assert 'No file or directory matching input' in caplog.text
+
+
+def test_expand_inputs_tilde_without_user(caplog: pytest.LogCaptureFixture) -> None:
+    sb._expand_inputs(['~/nothing-here'], None)
+
+    assert 'Cannot expand' in caplog.text
+
+
+def test_missing_config(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
+    config_args = sb._read_config(str(tmp_path / 'missing.conf'), USER)
+
+    assert config_args['inputs'] is None
+    assert config_args['output'] is None
+    assert config_args['keep'] == -1
+    assert 'does not exist' in caplog.text
+
+
+def test_missing_config_without_user(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
+    sb._read_config(str(tmp_path / 'missing.conf'), None)
+
+    assert 'User not specified' in caplog.text
+
+
+def test_read_config(tmp_path: Path) -> None:
+    (tmp_path / 'one').mkdir()
+    (tmp_path / 'two').mkdir()
+
+    config_file = write_config(tmp_path, f"""
+[backup]
+inputs={tmp_path}/one, {tmp_path}/two, {tmp_path}/one
+backup_dir=~/backups
+exclude=*.bak, , *.tmp
+keep=3
+""")
+
+    config_args = sb._read_config(config_file, USER)
+
+    assert config_args['inputs'] is not None
+    assert sorted(config_args['inputs']) == [f'{tmp_path}/one', f'{tmp_path}/two']
+    # Expanded later, as it may be a path on the server
+    assert config_args['output'] == '~/backups'
+    assert config_args['exclude'] == ['*.bak', '*.tmp']
+    assert config_args['keep'] == 3
+    assert config_args['ssh_host'] is None
+    assert config_args['ssh_user'] is None
+    assert config_args['ssh_keyfile'] is None
+    assert config_args['remote_sudo'] is False
+    assert config_args['numeric_ids'] is False
+
+
+def test_optional_keys_missing(tmp_path: Path) -> None:
+    config_file = write_config(tmp_path, f"""
+[backup]
+inputs={tmp_path}
+backup_dir=/backups
+""")
+
+    config_args = sb._read_config(config_file, USER)
+
+    assert config_args['exclude'] == []
+    assert config_args['keep'] == -1
+
+
+def test_old_default_section(tmp_path: Path) -> None:
+    config_file = write_config(tmp_path, f"""
+[default]
+inputs={tmp_path}
+backup_dir=/backups
+""")
+
+    config_args = sb._read_config(config_file, USER)
+
+    assert config_args['inputs'] == [str(tmp_path)]
+    assert config_args['output'] == '/backups'
+
+
+def test_percent_in_values(tmp_path: Path) -> None:
+    config_file = write_config(tmp_path, f"""
+[backup]
+inputs={tmp_path}
+backup_dir=/backups/100%
+exclude=%temp%
+""")
+
+    config_args = sb._read_config(config_file, USER)
+
+    assert config_args['output'] == '/backups/100%'
+    assert config_args['exclude'] == ['%temp%']
+
+
+def test_server_section(tmp_path: Path) -> None:
+    config_file = write_config(tmp_path, f"""
+[backup]
+inputs={tmp_path}
+backup_dir=backups
+
+[server]
+ssh_host=server
+ssh_user=user
+ssh_keyfile=/keys/id_ed25519
+remote_sudo=yes
+numeric_ids=true
+""")
+
+    config_args = sb._read_config(config_file, USER)
+
+    assert config_args['ssh_host'] == 'server'
+    assert config_args['ssh_user'] == 'user'
+    assert config_args['ssh_keyfile'] == '/keys/id_ed25519'
+    assert config_args['remote_sudo'] is True
+    assert config_args['numeric_ids'] is True
+
+
+def test_empty_server_values(tmp_path: Path) -> None:
+    config_file = write_config(tmp_path, f"""
+[backup]
+inputs={tmp_path}
+backup_dir=/backups
+
+[server]
+ssh_host=
+ssh_user=
+ssh_keyfile=
+remote_sudo=
+numeric_ids=
+""")
+
+    config_args = sb._read_config(config_file, USER)
+
+    assert config_args['ssh_host'] is None
+    assert config_args['ssh_user'] is None
+    assert config_args['ssh_keyfile'] is None
+    assert config_args['remote_sudo'] is False
+    assert config_args['numeric_ids'] is False
+
+
+def test_only_ssh_host(tmp_path: Path) -> None:
+    config_file = write_config(tmp_path, f"""
+[backup]
+inputs={tmp_path}
+backup_dir=/backups
+
+[server]
+ssh_host=server
+""")
+
+    config_args = sb._read_config(config_file, USER)
+
+    # Kept, so that the missing user is reported instead of silently backing up locally
+    assert config_args['ssh_host'] == 'server'
+    assert config_args['ssh_user'] is None
+
+
+@pytest.mark.parametrize('line', ['keep=abc', 'remote_sudo=maybe'])
+def test_bad_value(tmp_path: Path, line: str) -> None:
+    section = '[server]' if line.startswith('remote_sudo') else ''
+
+    config_file = write_config(tmp_path, f"""
+[backup]
+inputs={tmp_path}
+backup_dir=/backups
+{section}
+{line}
+""")
+
+    with pytest.raises(ValueError):
+        sb._read_config(config_file, USER)
+
+
+@pytest.mark.parametrize('text', [
+    '[backup]\nbackup_dir=/backups\n',
+    '[backup]\ninputs=/tmp\n',
+    '[other]\ninputs=/tmp\n',
+    'inputs=/tmp\n',
+])
+def test_bad_config(tmp_path: Path, text: str) -> None:
+    config_file = write_config(tmp_path, text)
+
+    with pytest.raises(configparser.Error):
+        sb._read_config(config_file, USER)

tests/test_local.py

@@ -0,0 +1,360 @@
+"""Tests for local backups, with real rsync runs and with a fake Popen"""
+
+import io
+import os
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any, Callable, Iterator, Optional
+
+import pytest
+
+from simple_backup import simple_backup as sb
+
+OPTIONS = '-r -v --ignore-missing-args --mkpath -a -h -H -X -s'
+
+
+@pytest.fixture(autouse=True)
+def fake_now(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Give each backup a new timestamp, as real runs within the same second would collide"""
+
+    def times() -> Iterator[datetime]:
+        t = datetime(2026, 10, 14, 10, 0, 0)
+
+        while True:
+            yield t
+            t += timedelta(hours=1)
+
+    it = times()
+
+    class FakeDatetime:
+        @staticmethod
+        def now() -> datetime:
+            return next(it)
+
+    monkeypatch.setattr(sb, 'datetime', FakeDatetime)
+
+
+@pytest.fixture
+def src(tmp_path: Path) -> Path:
+    src = tmp_path / 'src dir'
+    src.mkdir()
+    (src / 'same.txt').write_text('unchanged')
+    (src / 'changed.txt').write_text('first')
+
+    return src
+
+
+@pytest.fixture
+def out(tmp_path: Path) -> Path:
+    out = tmp_path / 'out dir'
+    out.mkdir()
+
+    return out
+
+
+def make_backup(out: Path, inputs: list[str], keep: int = -1, exclude: Optional[list[str]] = None,
+                remove_before: bool = False, progress: bool = False) -> sb.Backup:
+    backup = sb.Backup(inputs, str(out), exclude or [], keep, OPTIONS, remove_before=remove_before,
+                       progress=progress)
+
+    assert backup.check_params() == 0
+
+    return backup
+
+
+def run_backup(out: Path, inputs: list[str], **kwargs: Any) -> int:
+    return make_backup(out, inputs, **kwargs).run()
+
+
+def backups(out: Path) -> list[str]:
+    return sorted(d.name for d in (out / 'simple_backup').iterdir() if d.is_dir() and not d.is_symlink())
+
+
+def copied(out: Path, backup: str, path: Path) -> Path:
+    """Path of a file inside a backup (rsync keeps the full source path)"""
+
+    return out / 'simple_backup' / backup / str(path).lstrip('/')
+
+
+class TestCheckParams:
+    def test_no_inputs(self, out: Path) -> None:
+        assert sb.Backup([], str(out), [], -1, OPTIONS).check_params() == 1
+
+    def test_no_output(self) -> None:
+        assert sb.Backup(['/x'], None, [], -1, OPTIONS).check_params() == 2
+
+    def test_missing_output(self, tmp_path: Path) -> None:
+        assert sb.Backup(['/x'], str(tmp_path / 'missing'), [], -1, OPTIONS).check_params() == 2
+
+    def test_relative_output(self, out: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.chdir(out.parent)
+        backup = sb.Backup(['/x'], out.name, [], None, OPTIONS)
+
+        assert backup.check_params() == 0
+        assert backup.output == str(out)
+        assert backup.keep == -1
+
+
+class TestRsync:
+    def test_first_backup(self, src: Path, out: Path, caplog: pytest.LogCaptureFixture) -> None:
+        assert run_backup(out, [str(src)]) == 0
+
+        assert backups(out) == ['2026-10-14_10-00-00']
+        assert copied(out, '2026-10-14_10-00-00', src / 'same.txt').read_text() == 'unchanged'
+        assert 'Backup completed' in caplog.text
+        assert 'rsync: sent' in caplog.text
+
+    def test_hard_links(self, src: Path, out: Path) -> None:
+        assert run_backup(out, [str(src)]) == 0
+        (src / 'changed.txt').write_text('second')
+        assert run_backup(out, [str(src)]) == 0
+
+        first, second = backups(out)
+
+        assert copied(out, first, src / 'same.txt').stat().st_ino == \
+            copied(out, second, src / 'same.txt').stat().st_ino
+        assert copied(out, first, src / 'changed.txt').stat().st_ino != \
+            copied(out, second, src / 'changed.txt').stat().st_ino
+        assert copied(out, second, src / 'changed.txt').read_text() == 'second'
+
+    def test_exclude_and_missing_input(self, src: Path, out: Path, caplog: pytest.LogCaptureFixture) -> None:
+        (src / 'skip.bak').touch()
+
+        assert run_backup(out, [str(src), str(src / 'missing')], exclude=['*.bak']) == 0
+
+        (backup,) = backups(out)
+        assert copied(out, backup, src / 'same.txt').exists()
+        assert not copied(out, backup, src / 'skip.bak').exists()
+        assert 'Input' in caplog.text and 'not found' in caplog.text
+
+    def test_all_inputs_missing(self, src: Path, out: Path) -> None:
+        assert run_backup(out, [str(src / 'missing')]) == 1
+        assert not (out / 'simple_backup').exists()
+
+    @pytest.mark.parametrize('keep, remaining', [
+        (-1, ['2026-10-14_10-00-00', '2026-10-14_11-00-00', '2026-10-14_12-00-00']),
+        (1, ['2026-10-14_11-00-00', '2026-10-14_12-00-00']),
+        (0, ['2026-10-14_12-00-00']),
+    ])
+    def test_keep(self, src: Path, out: Path, keep: int, remaining: list[str]) -> None:
+        for _ in range(3):
+            assert run_backup(out, [str(src)], keep=keep) == 0
+
+        assert backups(out) == remaining
+
+    def test_removal_logged(self, src: Path, out: Path, caplog: pytest.LogCaptureFixture) -> None:
+        for _ in range(3):
+            assert run_backup(out, [str(src)]) == 0
+
+        assert run_backup(out, [str(src)], keep=0) == 0
+        assert 'Removing backup 2026-10-14_10-00-00 (1/3)...' in caplog.text
+        assert 'Removing backup 2026-10-14_12-00-00 (3/3)...' in caplog.text
+        assert 'Removed 3 backups' in caplog.text
+
+    def test_remove_before(self, src: Path, out: Path) -> None:
+        assert run_backup(out, [str(src)]) == 0
+        assert run_backup(out, [str(src)], keep=0, remove_before=True) == 0
+
+        (backup,) = backups(out)
+        assert backup == '2026-10-14_11-00-00'
+        # Nothing to link against, as the old backup was removed first
+        assert copied(out, backup, src / 'same.txt').stat().st_nlink == 1
+
+    def test_stray_files_and_symlinks(self, src: Path, out: Path) -> None:
+        assert run_backup(out, [str(src)]) == 0
+
+        base = out / 'simple_backup'
+        (base / 'notes.txt').write_text('keep me')
+        # Left by old versions; must not count as a backup, nor be used for --link-dest
+        (base / 'last_backup').symlink_to(base / backups(out)[0])
+
+        assert run_backup(out, [str(src)], keep=0) == 0
+
+        assert backups(out) == ['2026-10-14_11-00-00']
+        assert (base / 'notes.txt').read_text() == 'keep me'
+        assert (base / 'last_backup').is_symlink()
+
+    def test_old_folder_name_format(self, src: Path, out: Path) -> None:
+        assert run_backup(out, [str(src)]) == 0
+
+        base = out / 'simple_backup'
+        (base / '2026-10-14_10-00-00').rename(base / '2026-10-03 20:15:25')
+
+        assert run_backup(out, [str(src)]) == 0
+        assert backups(out) == ['2026-10-03 20:15:25', '2026-10-14_11-00-00']
+        assert copied(out, '2026-10-03 20:15:25', src / 'same.txt').stat().st_ino == \
+            copied(out, '2026-10-14_11-00-00', src / 'same.txt').stat().st_ino
+
+        assert run_backup(out, [str(src)], keep=0) == 0
+        assert backups(out) == ['2026-10-14_12-00-00']
+
+    def test_progress(self, src: Path, out: Path, caplog: pytest.LogCaptureFixture,
+                      capsys: pytest.CaptureFixture[str]) -> None:
+        assert run_backup(out, [str(src)], progress=True) == 0
+
+        assert '100%' in capsys.readouterr().err
+        # Only the summary is logged, not the progress updates
+        assert 'rsync: sent' in caplog.text
+        assert 'rsync: total size' in caplog.text
+        assert '%' not in caplog.text
+
+    def test_cli(self, src: Path, out: Path, tmp_path: Path, run_cli: Callable[..., int]) -> None:
+        config_file = str(tmp_path / 'missing.conf')
+
+        for _ in range(2):
+            assert run_cli('-c', config_file, '-i', str(src), '-o', str(out), '-k', '0') == 0
+
+        assert backups(out) == ['2026-10-14_11-00-00']
+
+
+class ChunkStream(io.BufferedIOBase):
+    """Returns the given chunks one per read1() call, like a pipe"""
+
+    def __init__(self, chunks: list[bytes]) -> None:
+        self.chunks = chunks
+
+    def read1(self, size: int = -1) -> bytes:
+        return self.chunks.pop(0) if self.chunks else b''
+
+
+class TestReadRecords:
+    def test_split_on_newline_and_cr(self) -> None:
+        stream = ChunkStream([b'file\n\r  1.00M  50%\r  2.00M 100%\n\nsent 1 bytes\ntotal'])
+
+        assert list(sb._read_records(stream)) == ['file', '', '  1.00M  50%', '  2.00M 100%', '', 'sent 1 bytes',
+                                                  'total']
+
+    def test_records_across_chunks(self) -> None:
+        stream = ChunkStream([b'fi', b'le\r', b'\nna', 'mé\n'.encode(), b'\xff\n'])
+
+        assert list(sb._read_records(stream)) == ['file', '', 'namé', '�']
+
+
+class TestProgressBar:
+    @pytest.fixture(autouse=True)
+    def terminal(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(sb, 'get_terminal_size', lambda: os.terminal_size((81, 24)))
+        # 30 s elapsed at each update
+        monkeypatch.setattr(sb, 'default_timer', iter([0.0, 30.0]).__next__)
+
+    def test_scanning(self, capsys: pytest.CaptureFixture[str]) -> None:
+        bar = sb._ProgressBar()
+
+        assert bar.update('          2.00M  25%    1.47MB/s    0:00:02 (xfr#1, ir-chk=1009/2632)')
+        assert capsys.readouterr().err == '\r1,623 files checked  2.00M copied  (scanning)\033[K'
+
+    def test_update(self, capsys: pytest.CaptureFixture[str]) -> None:
+        bar = sb._ProgressBar()
+
+        # With --link-dest rsync's percentage leaves out linked files, so the files are counted instead
+        assert bar.update('      4,000,000   1%    2.97MB/s    0:00:01 (xfr#2, to-chk=1/4)')
+        assert capsys.readouterr().err == \
+            '\r[######################--------]   75%  3/4 files  4,000,000 copied  ETA 0:00:10\033[K'
+
+    def test_keeps_counts_within_file(self, monkeypatch: pytest.MonkeyPatch,
+                                      capsys: pytest.CaptureFixture[str]) -> None:
+        monkeypatch.setattr(sb, 'default_timer', iter([0.0, 30.0, 30.0]).__next__)
+        bar = sb._ProgressBar()
+
+        assert bar.update('          2.00M   1%    1.47MB/s    0:00:02 (xfr#1, to-chk=2/4)')
+        assert bar.update('          3.00M   2%    1.47MB/s    0:00:03')
+        assert capsys.readouterr().err.endswith(']   50%  2/4 files  3.00M copied  ETA 0:00:30\033[K')
+
+    def test_not_progress(self, capsys: pytest.CaptureFixture[str]) -> None:
+        bar = sb._ProgressBar()
+
+        for record in ['home/user/file 50% 1:00:00', 'sent 1 bytes  received 2 bytes', 'total size is 3']:
+            assert not bar.update(record)
+
+        bar.clear()
+        assert capsys.readouterr().err == ''
+
+    def test_narrow_terminal(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
+        monkeypatch.setattr(sb, 'get_terminal_size', lambda: os.terminal_size((20, 24)))
+
+        assert sb._ProgressBar().update('  6.00M 100%    1.48MB/s    0:00:03 (xfr#3, to-chk=0/4)')
+        # No room for the bar
+        assert capsys.readouterr().err == '\r100%  4/4 files  6.\033[K'
+
+
+class FakePopen:
+    """Stands in for rsync: creates the destination and returns a chosen return code"""
+
+    returncode = 0
+    calls: list[list[str]] = []
+
+    def __init__(self, args: list[str], **kwargs: Any) -> None:
+        FakePopen.calls.append(args)
+        os.makedirs(args[args.index('/') + 1])
+        self.stdin = io.BytesIO()
+        self.stdout = io.BytesIO(b'sending incremental file list\nsent 1 bytes  received 2 bytes\ntotal size is 3\n')
+
+    def __enter__(self) -> 'FakePopen':
+        return self
+
+    def __exit__(self, *args: Any) -> None:
+        pass
+
+    def wait(self) -> int:
+        return FakePopen.returncode
+
+
+@pytest.fixture
+def fake_popen(monkeypatch: pytest.MonkeyPatch) -> type[FakePopen]:
+    FakePopen.returncode = 0
+    FakePopen.calls = []
+    monkeypatch.setattr(sb, 'Popen', FakePopen)
+
+    return FakePopen
+
+
+@pytest.fixture
+def old_backups(out: Path) -> list[str]:
+    names = ['2026-10-01_10-00-00', '2026-10-02_10-00-00']
+
+    for name in names:
+        (out / 'simple_backup' / name).mkdir(parents=True)
+
+    return names
+
+
+class TestReturnCodes:
+    def test_args(self, src: Path, out: Path, fake_popen: type[FakePopen], old_backups: list[str]) -> None:
+        assert run_backup(out, [str(src)]) == 0
+
+        (args,) = fake_popen.calls
+        assert args[0] == '/usr/bin/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'
+
+    @pytest.mark.parametrize('returncode, result, removed', [
+        (0, 0, True),
+        (24, 0, True),
+        (23, 4, False),
+        (2, 4, False),
+    ])
+    def test_old_backups_removed_only_on_success(self, src: Path, out: Path, fake_popen: type[FakePopen],
+                                                 old_backups: list[str], caplog: pytest.LogCaptureFixture,
+                                                 returncode: int, result: int, removed: bool) -> None:
+        fake_popen.returncode = returncode
+
+        assert run_backup(out, [str(src)], keep=0) == result
+
+        if removed:
+            assert backups(out) == ['2026-10-14_10-00-00']
+        else:
+            assert backups(out) == [*old_backups, '2026-10-14_10-00-00']
+            assert 'Old backups will not be removed' in caplog.text
+            assert f'return code {returncode}' in caplog.text
+
+    @pytest.mark.parametrize('keep, message', [
+        (0, 'No complete backup is available'),
+        (1, 'Old backups were removed before the backup'),
+    ])
+    def test_failure_after_remove_before(self, src: Path, out: Path, fake_popen: type[FakePopen],
+                                         old_backups: list[str], caplog: pytest.LogCaptureFixture,
+                                         keep: int, message: str) -> None:
+        fake_popen.returncode = 23
+
+        assert run_backup(out, [str(src)], keep=keep, remove_before=True) == 4
+        assert message in caplog.text

tests/test_remote.py

@@ -0,0 +1,568 @@
+"""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)
+        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 = []
+        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