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.1.7
-pkgrel=1
-url="https://git.shouldnt.work/fuxino/${pkgname}"
+pkgver=4.4.0
+pkgrel=2
+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=('011cd5546745e413ac7603b64531def2688c2bf6e4ea19876a67547a1db208ce')
+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}
-    python -m installer --destdir=${pkgdir} dist/*.whl
-    install -Dm644 ${pkgname}/${pkgname}.conf ${pkgdir}/usr/share/doc/${pkgname}/${pkgname}.conf
-    install -Dm644 man/${pkgname}.1 ${pkgdir}/usr/share/man/man1/${pkgname}.1
+    cd "$srcdir"/"$pkgname"
+    python -m installer --destdir="$pkgdir" dist/*.whl
+    install -Dm644 "$pkgname"/"$pkgname".conf "$pkgdir"/usr/share/doc/"$pkgname"/"$pkgname".conf
+    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:
-
-```bash
-pip install --upgrade build
-```
-
-Finally, run:
+Then build the package (this requires the Python `build` module, e.g. the python-build package on Arch Linux):
 
 ```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):
-- **simple_backup** for the release version
-- **simple_backup-git** for the git version
+Many distributions don't allow installing packages with pip into the system Python ("externally-managed-environment" error).
+In that case, install the wheel in a virtual environment, for example with pipx:
+
+```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.2.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
@@ -146,6 +152,14 @@ Copy it to the default location ($HOME/.config/simple_backup) and edit it as nee
 It is possible to choose a directory on a remote server as destination for the backup. The files
 are copied by rsync through SSH. Server hostname and username must be specified, either in the
 configuration file, or on the command line (\(aq\-\-ssh\-host\(aq and \(aq\-\-ssh\-user\(aq options).
+.P
+The output directory is a path on the server. A leading \(aq~\(aq and relative paths refer to the
+home directory of the user on the server. On the command line, quote \(aq~\(aq so that the local
+shell doesn\(aqt expand it, for example:
+.P
+.EX
+    simple_backup \-\-ssh\-host server \-\-ssh\-user user \-o \(aq~/backups\(aq
+.EE
 .SS AUTHENTICATION
 For authentication, it is possible to use SSH key or password.
 .P
@@ -180,7 +194,8 @@ for details. For this reason, use SSH key authentication if possible.
 .SH EXIT STATUS
 .TP
 .B 0
-The backup was completed without errors.
+The backup was completed without errors. This includes rsync return code 24 (some source files
+vanished during the transfer), which only logs a warning.
 .TP
 .B 1
 No valid inputs selected for backup.
@@ -192,13 +207,13 @@ Backup failed because output directory for storing the backup does not exist.
 Permission denied to access the output directory.
 .TP
 .B 4
-rsync error (rsync returned a non-zero value).
+rsync error (rsync returned a value other than 0 or 24).
 .TP
 .B 5
 SSH connection failed.
 .TP
 .B 6
-Bad configuration file.
+Bad configuration file or options (for example, only one of SSH host and SSH user specified).
 .SH SEE ALSO
 .BR rsync (1)
 .SH AUTHORS

pyproject.toml

@@ -33,7 +33,7 @@ classifiers = [
 [project.optional-dependencies]
 JOURNAL = ["systemd-python"]
 NOTIFICATIONS = ["dbus-python"]
-REMOTE = ["paramiko"]
+REMOTE = ["paramiko>=3.2"]
 
 [project.urls]
 Homepage = "https://git.shouldnt.work/fuxino/simple_backup"
@@ -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.2.0'
+__version__ = '4.4.1'

simple_backup/simple_backup.conf

@@ -4,7 +4,7 @@
 # Files and directories to backup. Multiple items can be separated using a comma (','). It is possible to use wildcards (i.e. '*' to match multiple characters and '~' for the user's home directory).
 inputs=/home/user
 
-# Output directory.
+# Output directory. For remote backups, this is a path on the server ('~' and relative paths refer to the home directory on the server).
 backup_dir=/media/Backup
 
 # Files, directories and patterns to exclude from the backup. Multiple items can be separated using a comma.
@@ -18,5 +18,5 @@ keep=-1
 # ssh_host=
 # ssh_user=
 # ssh_keyfile=
-# remote_sudo=
-# numeric_ids=
+# remote_sudo=false
+# numeric_ids=false

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
@@ -36,7 +38,6 @@ from glob import glob
 
 try:
     import paramiko
-    from paramiko import RSAKey, Ed25519Key, ECDSAKey, DSSKey
 except ImportError:
     pass
 
@@ -70,6 +71,13 @@ if journal:
 P = ParamSpec('P')
 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
@@ -96,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
 
@@ -120,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():
@@ -134,41 +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:
-        self.inputs = inputs
-        self.output = output
-        self.exclude = exclude
+                 remote_sudo: bool = False, remove_before: bool = False, verbose: bool = False,
+                 progress: bool = False) -> None:
+        # Missing values are reported by check_params()
+        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._password = None
         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')
 
@@ -178,33 +251,57 @@ class Backup:
             self._remote = True
 
         if self._remote:
-            self._ssh = self._ssh_connect(homedir)
+            try:
+                self._ssh = self._ssh_connect(homedir)
+            except OSError as e:
+                # Unknown hostname, connection refused, unreachable host or timeout
+                logger.critical('Can\'t connect to the server %s', self.ssh_host)
+                logger.critical(e)
+
+                return 5
 
             if self._ssh is None:
                 return 5
 
-            _, stdout, _ = self._ssh.exec_command(
-                f'if [ -d "{self.output}" ]; then echo "ok"; fi')
+            output = self._resolve_remote_path(self.output)
 
-            output = stdout.read().decode('utf-8').strip()
-
-            if output != 'ok':
+            if output is None:
                 logger.critical('Output path for backup does not exist')
 
                 return 2
+
+            self.output = output
         else:
             if not os.path.isdir(self.output):
                 logger.critical('Output path for backup does not exist')
 
                 return 2
 
-        self.output = os.path.abspath(self.output)
-
-        if self.keep is None:
-            self.keep = -1
+            self.output = os.path.abspath(self.output)
 
         return 0
 
+    def _resolve_remote_path(self, path: str) -> Optional[str]:
+        """Return the absolute path of a directory on the server, or None if it doesn't exist"""
+
+        assert self._ssh is not None
+
+        # Leave a leading '~' unquoted, so that the server's shell expands it to the remote home.
+        # Relative paths are resolved from the remote home, where the command starts
+        if path == '~' or path.startswith('~/'):
+            target = f'~/{shlex.quote(path[2:])}'
+        else:
+            target = shlex.quote(path)
+
+        _, stdout, _ = self._ssh.exec_command(f'cd -- {target} && pwd')
+
+        output = stdout.read().decode('utf-8').strip().split('\n')[-1]
+
+        if not output.startswith('/'):
+            return None
+
+        return output
+
     # Function to create the actual backup directory
     def define_backup_dir(self) -> None:
         """Define the actual backup dir"""
@@ -222,10 +319,23 @@ class Backup:
         if self._remote:
             assert self._ssh is not None
 
-            _, stdout, _ = self._ssh.exec_command(
-                f'ls {self.output}/simple_backup')
+            backup_dir = shlex.quote(self.output + '/simple_backup/')
 
-            dirs = stdout.read().decode('utf-8').strip().split('\n')
+            # List only directories, and nothing (without errors) if no backup was made yet
+            _, stdout, stderr = self._ssh.exec_command(
+                f'if [ -d {backup_dir} ]; then find {backup_dir} -mindepth 1 -maxdepth 1 -type d; fi')
+
+            output = stdout.read().decode('utf-8')
+            err = stderr.read().decode('utf-8').strip()
+
+            if err != '':
+                # Don't risk removing the wrong directories if the list may be incomplete
+                logger.error('Cannot list old backups on the server. Old backups will not be removed')
+                logger.error(err)
+
+                return
+
+            dirs = [os.path.basename(d) for d in output.split('\n') if d != '']
 
             n_backup = len(dirs)
 
@@ -238,27 +348,37 @@ 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 "{self.output}/simple_backup/{dirs[i]}"')
+                            f'sudo rm -r {shlex.quote(f"{self.output}/simple_backup/{dirs[i]}")}')
                     else:
                         _, _, stderr = self._ssh.exec_command(
-                            f'rm -r "{self.output}/simple_backup/{dirs[i]}"')
+                            f'rm -r {shlex.quote(f"{self.output}/simple_backup/{dirs[i]}")}')
 
                     err = stderr.read().decode('utf-8').strip().split('\n')[0]
 
                     if err != '':
                         logger.error(
-                            'Error while removing backup %s.', {dirs[i]})
+                            'Error while removing backup %s.', dirs[i])
                         logger.error(err)
                     else:
                         count += 1
         else:
             try:
-                dirs = os.listdir(f'{self.output}/simple_backup')
+                # Only count directories, so that stray files don't count as backups
+                dirs = [f.name for f in os.scandir(f'{self.output}/simple_backup')
+                        if f.is_dir(follow_symlinks=False)]
             except FileNotFoundError:
                 return
+            except PermissionError:
+                logger.error('Cannot list old backups. Permission denied. Old backups will not be removed')
+
+                return
 
             n_backup = len(dirs)
 
@@ -271,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
@@ -298,7 +422,7 @@ class Backup:
                 sys.exit(5)
 
             _, stdout, _ = self._ssh.exec_command(
-                f'find {self.output}/simple_backup/ -mindepth 1 -maxdepth 1 -type d | sort')
+                f'find {shlex.quote(self.output + "/simple_backup/")} -mindepth 1 -maxdepth 1 -type d | sort')
             output = stdout.read().decode('utf-8').strip().split('\n')
 
             if output[-1] != '':
@@ -334,6 +458,18 @@ class Backup:
 
             return None
 
+        connected = None
+
+        try:
+            connected = self._ssh_login(ssh, homedir)
+
+            return connected
+        finally:
+            # Don't leave a half-open connection (e.g. after failed authentication)
+            if connected is None:
+                ssh.close()
+
+    def _ssh_login(self, ssh: paramiko.client.SSHClient, homedir: str) -> Optional[paramiko.client.SSHClient]:
         try:
             ssh.load_host_keys(filename=f'{homedir}/.ssh/known_hosts')
         except FileNotFoundError:
@@ -345,16 +481,22 @@ class Backup:
             # WarningPolicy emits a UserWarning for unknown host keys
             with warnings.catch_warnings():
                 warnings.simplefilter('error', UserWarning)
-                ssh.connect(self.ssh_host, username=self.ssh_user)
+                ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT)
 
             return ssh
         except UserWarning:
-            k = input(
-                f'Unknown key for host {self.ssh_host}. Continue anyway? (Y/N) ')
+            try:
+                k = input(
+                    f'Unknown key for host {self.ssh_host}. Continue anyway? (Y/N) ')
+            except EOFError:
+                # No terminal available (e.g. cron or systemd)
+                k = ''
 
-            if k[0].upper() == 'Y':
+            if k.strip().upper().startswith('Y'):
                 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
             else:
+                logger.critical('Unknown host key for %s. Connection aborted', self.ssh_host)
+
                 return None
         except paramiko.BadHostKeyException as e:
             logger.critical('Can\'t connect to the server.')
@@ -365,7 +507,7 @@ class Backup:
             pass
 
         try:
-            ssh.connect(self.ssh_host, username=self.ssh_user)
+            ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT)
 
             return ssh
         except paramiko.SSHException:
@@ -379,7 +521,7 @@ class Backup:
                     password = getpass(
                         f'{self.ssh_user}@{self.ssh_host}\'s password: ')
 
-                ssh.connect(self.ssh_host, username=self.ssh_user,
+                ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT,
                             password=password)
 
                 self._password_auth = True
@@ -397,51 +539,13 @@ class Backup:
 
                 return None
 
-        pkey = None
+        pkey = self._load_ssh_key()
+
+        if pkey is None:
+            return None
 
         try:
-            pkey = RSAKey.from_private_key_file(self.ssh_keyfile)
-        except paramiko.PasswordRequiredException:
-            password = getpass(
-                f'Enter passwphrase for key \'{self.ssh_keyfile}\': ')
-
-            try:
-                pkey = RSAKey.from_private_key_file(self.ssh_keyfile, password)
-            except paramiko.SSHException:
-                pass
-
-        if pkey is None:
-            try:
-                pkey = Ed25519Key.from_private_key_file(self.ssh_keyfile)
-            except paramiko.PasswordRequiredException:
-                try:
-                    pkey = Ed25519Key.from_private_key_file(
-                        self.ssh_keyfile, password)
-                except paramiko.SSHException:
-                    pass
-
-        if pkey is None:
-            try:
-                pkey = ECDSAKey.from_private_key_file(self.ssh_keyfile)
-            except paramiko.PasswordRequiredException:
-                try:
-                    pkey = ECDSAKey.from_private_key_file(
-                        self.ssh_keyfile, password)
-                except paramiko.SSHException:
-                    pass
-
-        if pkey is None:
-            try:
-                pkey = DSSKey.from_private_key_file(self.ssh_keyfile)
-            except paramiko.PasswordRequiredException:
-                try:
-                    pkey = DSSKey.from_private_key_file(
-                        self.ssh_keyfile, password)
-                except paramiko.SSHException:
-                    pass
-
-        try:
-            ssh.connect(self.ssh_host, username=self.ssh_user, pkey=pkey)
+            ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT, pkey=pkey)
         except paramiko.SSHException:
             logger.critical('SSH connection to server failed')
 
@@ -449,6 +553,37 @@ class Backup:
 
         return ssh
 
+    def _load_ssh_key(self) -> Optional[paramiko.PKey]:
+        try:
+            return paramiko.PKey.from_path(self.ssh_keyfile)
+        except TypeError:
+            # Raised when the key is encrypted and no passphrase was given
+            pass
+        except (OSError, ValueError, paramiko.UnknownKeyType) as e:
+            logger.critical('Unable to load SSH key %s', self.ssh_keyfile)
+            logger.critical(e)
+
+            return None
+
+        try:
+            # Fail instead of reading the passphrase with echo if no terminal is available
+            with warnings.catch_warnings():
+                warnings.simplefilter('error', GetPassWarning)
+                passphrase = getpass(
+                    f'Enter passphrase for key \'{self.ssh_keyfile}\': ')
+        except GetPassWarning as e:
+            logger.critical('Unable to get passphrase')
+            logger.critical(e)
+
+            return None
+
+        try:
+            return paramiko.PKey.from_path(self.ssh_keyfile, passphrase.encode())
+        except (ValueError, paramiko.SSHException):
+            logger.critical('Unable to load SSH key %s, wrong passphrase?', self.ssh_keyfile)
+
+            return None
+
     def _returncode_log(self, returncode: int) -> None:
         match returncode:
             case 2:
@@ -480,7 +615,7 @@ class Backup:
                     'Rsync error (return code 23) - Partial transfer due to error')
             case 24:
                 logger.warning(
-                    'Rsync error (return code 24) - Partial transfer due to vanished source files')
+                    'Rsync warning (return code 24) - Some source files vanished during the transfer')
             case 30:
                 logger.error(
                     'Rsync error (return code 30) - Timeout in data send/receive')
@@ -501,6 +636,13 @@ class Backup:
         finally:
             self._remove_temp_files()
 
+    def close(self) -> None:
+        """Close the SSH connection, if open"""
+
+        if self._ssh is not None:
+            self._ssh.close()
+            self._ssh = None
+
     def _remove_temp_files(self) -> None:
         for path in [self._inputs_path, self._exclude_path]:
             if path != '':
@@ -540,10 +682,9 @@ 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')
+            for e in self.exclude:
+                fp.write(e)
+                fp.write('\n')
 
         if self.keep != -1 and self._remove_before:
             self.remove_old_backups()
@@ -553,46 +694,61 @@ class Backup:
 
         logger.info('Copying files. This may take a long time...')
 
-        if self._last_backup == '':
-            rsync = f'/usr/bin/rsync {self.options} --exclude-from={self._exclude_path} ' +\
-                f'--files-from={self._inputs_path} / "{self._server}{self._output_dir}"'
-        else:
-            rsync = f'/usr/bin/rsync {self.options} --link-dest="{self._last_backup}" --exclude-from=' +\
-                f'{self._exclude_path} --files-from={self._inputs_path} / "{self._server}{self._output_dir}"'
+        # Build the argument list directly, so paths with spaces or quotes are passed unchanged
+        args = ['/usr/bin/rsync', *self.options.split()]
 
-        euid = os.geteuid()
+        if self._last_backup != '':
+            args.append(f'--link-dest={self._last_backup}')
 
-        if euid == 0 and self.ssh_keyfile is not None:
-            rsync = f'{rsync} -e \'ssh -i {self.ssh_keyfile} -o StrictHostKeyChecking=no\''
-        elif self._password_auth and which('sshpass'):
-            rsync = f'{rsync} -e \'sshpass -e ssh -l {self.ssh_user} -o StrictHostKeyChecking=no\''
+        args.extend([f'--exclude-from={self._exclude_path}', f'--files-from={self._inputs_path}',
+                     '/', f'{self._server}{self._output_dir}'])
+
+        # 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 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:
-            rsync = f'{rsync} -e \'ssh -o StrictHostKeyChecking=no\''
+            ssh = 'ssh -o StrictHostKeyChecking=accept-new'
+
+        args.extend(['-e', ssh])
 
         if self._remote and self.remote_sudo:
-            rsync = f'{rsync} --rsync-path="sudo rsync"'
+            args.append('--rsync-path=sudo rsync')
 
-        args = shlex.split(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]]
-            output, _ = p.communicate()
+            assert p.stdin is not None and isinstance(p.stdout, io.BufferedIOBase)
+            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
-
-        output = output.decode("utf-8").split('\n')
+            returncode = p.wait()
 
         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)
 
@@ -620,40 +776,36 @@ class Backup:
             else:
                 logger.warning('Backup not completed successfully. Old backups will not be removed')
 
-        if self._remote:
-            assert self._ssh is not None
+        # Files vanishing during the transfer is normal for a live system (return code 24)
+        if returncode not in [0, 24]:
+            logger.error(
+                'Some errors occurred while performing the backup')
 
-            _, stdout, _ = self._ssh.exec_command(
-                f'if [ -d "{self._output_dir}" ]; then echo "ok"; fi')
+            _notify(
+                'Some errors occurred while performing the backup. Check log for details')
 
-            output = stdout.read().decode('utf-8').strip()
+            return 4
 
-            if output == 'ok':
-                logger.info('Backup completed')
+        logger.info('Backup completed')
 
-                _notify('Backup completed')
-            else:
-                logger.error('Backup failed')
+        _notify('Backup completed')
 
-                _notify('Backup failed (check log for details)')
+        return 0
 
-            if self._ssh:
-                self._ssh.close()
-        else:
-            if returncode != 0:
-                logger.error(
-                    'Some errors occurred while performing the backup')
 
-                _notify(
-                    'Some errors occurred while performing the backup. Check log for details')
+def _read_records(stream: io.BufferedIOBase) -> Iterator[str]:
+    """Yield rsync output as it arrives, split on '\\n' and '\\r' (which ends progress updates)"""
 
-                return 4
+    buffer = b''
 
-            logger.info('Backup completed')
+    while chunk := stream.read1(65536):
+        *records, buffer = re.split(rb'[\r\n]', buffer + chunk)
 
-            _notify('Backup completed')
+        for r in records:
+            yield r.decode('utf-8', errors='replace')
 
-            return 0
+    if buffer != b'':
+        yield buffer.decode('utf-8', errors='replace')
 
 
 def _detect_user() -> Optional[str]:
@@ -696,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'],
@@ -719,8 +873,8 @@ def _expand_home(path: str, user: str) -> str:
     return os.path.expanduser(path)
 
 
-def _expand_inputs(inputs, user: Optional[str] = None):
-    expanded_inputs = []
+def _expand_inputs(inputs: List[str], user: Optional[str] = None) -> List[str]:
+    expanded_inputs: List[str] = []
 
     for i in inputs:
         if i == '':
@@ -743,16 +897,38 @@ def _expand_inputs(inputs, user: Optional[str] = None):
     return expanded_inputs
 
 
-def _read_config(config_file, user: Optional[str] = None):
-    config_args = {'inputs': None,
-                   'output': None,
-                   'exclude': None,
-                   'keep': -1,
-                   'ssh_host': None,
-                   'ssh_user': None,
-                   'ssh_keyfile': None,
-                   'remote_sudo': False,
-                   'numeric_ids': False}
+class ConfigArgs(TypedDict):
+    """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,
+                               'ssh_host': None,
+                               'ssh_user': None,
+                               'ssh_keyfile': None,
+                               'remote_sudo': False,
+                               'numeric_ids': False}
 
     if not os.path.isfile(config_file):
         if user is not None:
@@ -776,66 +952,25 @@ def _read_config(config_file, user: Optional[str] = None):
         section = 'default'
         inputs = config.get(section, 'inputs')
 
-    inputs = [i.strip() for i in inputs.split(',')]
-    inputs = _expand_inputs(inputs, user)
-    inputs = list(set(inputs))
+    config_args['inputs'] = list(set(_expand_inputs([i.strip() for i in inputs.split(',')], user)))
 
-    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')
 
-    output = config.get(section, 'backup_dir')
+    exclude = config.get(section, 'exclude', fallback='')
+    config_args['exclude'] = [e.strip() for e in exclude.split(',') if e.strip() != '']
 
-    if user is not None:
-        output = _expand_home(output, user)
-    elif user is None and output.startswith('~'):
-        logger.warning('Cannot expand \'~\', no user specified')
+    config_args['keep'] = config.getint(section, 'keep', fallback=-1)
 
-    config_args['output'] = output
+    # Read host and user separately, so that a missing one is reported instead of silently
+    # disabling remote backup. Empty values count as not set
+    config_args['ssh_host'] = config.get('server', 'ssh_host', fallback='').strip() or None
+    config_args['ssh_user'] = config.get('server', 'ssh_user', fallback='').strip() or None
 
-    try:
-        exclude = config.get(section, 'exclude')
-        exclude = [e.strip() for e in exclude.split(',') if e.strip() != '']
-    except configparser.NoOptionError:
-        exclude = []
+    config_args['ssh_keyfile'] = config.get('server', 'ssh_keyfile', fallback='').strip() or None
 
-    config_args['exclude'] = exclude
-
-    try:
-        keep = config.getint(section, 'keep')
-    except configparser.NoOptionError:
-        keep = -1
-
-    config_args['keep'] = keep
-
-    try:
-        ssh_host = config.get('server', 'ssh_host')
-        ssh_user = config.get('server', 'ssh_user')
-    except (configparser.NoSectionError, configparser.NoOptionError):
-        ssh_host = None
-        ssh_user = None
-
-    config_args['ssh_host'] = ssh_host
-    config_args['ssh_user'] = ssh_user
-
-    try:
-        ssh_keyfile = config.get('server', 'ssh_keyfile')
-    except (configparser.NoSectionError, configparser.NoOptionError):
-        ssh_keyfile = None
-
-    config_args['ssh_keyfile'] = ssh_keyfile
-
-    try:
-        remote_sudo = config.getboolean('server', 'remote_sudo')
-    except (configparser.NoSectionError, configparser.NoOptionError):
-        remote_sudo = False
-
-    config_args['remote_sudo'] = remote_sudo
-
-    try:
-        numeric_ids = config.getboolean('server', 'numeric_ids')
-    except (configparser.NoSectionError, configparser.NoOptionError):
-        numeric_ids = False
-
-    config_args['numeric_ids'] = numeric_ids
+    config_args['remote_sudo'] = _get_bool(config, 'remote_sudo')
+    config_args['numeric_ids'] = _get_bool(config, 'numeric_ids')
 
     return config_args
 
@@ -942,11 +1077,30 @@ def simple_backup() -> int:
         return 6
 
     inputs = args.inputs if args.inputs is not None else config_args['inputs']
-    output = args.output if args.output is not None else config_args['output']
     exclude = args.exclude if args.exclude is not None else config_args['exclude']
     keep = args.keep if args.keep is not None else config_args['keep']
     ssh_host = args.ssh_host if args.ssh_host is not None else config_args['ssh_host']
     ssh_user = args.ssh_user if args.ssh_user is not None else config_args['ssh_user']
+
+    if (ssh_host is None) != (ssh_user is None):
+        logger.critical('Both SSH host and SSH user are needed for remote backup')
+
+        return 6
+
+    if args.output is not None:
+        output = args.output
+    else:
+        output = config_args['output']
+
+        # For remote backups, '~' is expanded on the server
+        remote = ssh_host is not None and ssh_user is not None
+
+        if output is not None and not remote:
+            if user is not None:
+                output = _expand_home(output, user)
+            elif output.startswith('~'):
+                logger.warning('Cannot expand \'~\', no user specified')
+
     ssh_keyfile = args.keyfile if args.keyfile is not None else config_args['ssh_keyfile']
     remote_sudo = args.remote_sudo or config_args['remote_sudo']
 
@@ -968,17 +1122,19 @@ 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,
+                    progress=not args.no_progress and sys.stderr.isatty())
 
-    backup = Backup(inputs, output, exclude, keep, rsync_options, ssh_host, ssh_user, ssh_keyfile,
-                    remote_sudo, remove_before=args.remove_before_backup, verbose=args.verbose)
+    try:
+        return_code = backup.check_params(homedir)
 
-    return_code = backup.check_params(homedir)
+        if return_code == 0:
+            return backup.run()
 
-    if return_code == 0:
-        return backup.run()
-
-    return return_code
+        return return_code
+    finally:
+        backup.close()
 
 
 if __name__ == '__main__':

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