Compare commits

..
21 Commits
Author SHA1 Message Date
Fuxino b65955f9f6 Bump version to 4.6.0 2026-10-25 12:42:22 +01:00
Fuxino 6d54019a18 Save the last successful backup and add --last-backup
Write the completion time and backup directory to
$XDG_STATE_HOME/simple_backup/last_backup after each successful backup
(as the target user when run as root), for status bar reminders.
--last-backup prints it.
2026-10-25 12:36:00 +01:00
Fuxino 7547db1ea8 Update PKGBUILD 2026-10-21 22:50:33 +02:00
Fuxino 37e87f31b5 Bump version to 4.5.0 2026-10-21 22:49:04 +02:00
Fuxino 4ac8b64085 Test whole remote backups through a fake ssh command
Real rsync runs over fake ssh and sshpass scripts that execute the remote side locally, covering --link-dest, removal, key file, password and --remote-sudo without a server.
2026-10-21 22:48:35 +02:00
Fuxino 34e3e41382 Share one removal loop between local and remote backups
remove_old_backups() repeated the counting and sorting in both branches (the keep < -1 bug was in both); listing and removing are now per-branch helpers.
2026-10-21 22:44:19 +02:00
Fuxino 0ffb5c47ac Reuse _detect_user() and _expand_home() in simple_backup()
The entry point repeated both, and its copy of the '~' expansion missed a bare '~'.
2026-10-21 22:42:49 +02:00
Fuxino 59a3f94ec8 Log verbose rsync output one line per message
With -v, the whole output was logged as a single Python list.
2026-10-21 22:41:29 +02:00
Fuxino 408beefc66 Keep only rsync's summary lines when not verbose
Every output line was kept in memory although only the last two were logged, and fewer than two lines (e.g. with --rsync-options q) raised an IndexError.
2026-10-21 22:40:17 +02:00
Fuxino b54f7e1817 Exit with simple_backup()'s return code when run as a script 2026-10-21 22:37:55 +02:00
Fuxino 9e6aadf7d3 Find rsync in PATH and pass the SSH password only to rsync
A missing /usr/bin/rsync ended in a traceback, possibly after --remove-before-backup had removed old backups, and left SSHPASS in the environment.
2026-10-21 22:36:03 +02:00
Fuxino ac59969d82 Catch all OS errors when removing old local backups
Errors other than a missing directory or denied permission (e.g. read-only filesystem, device busy) ended in a traceback after the backup.
2026-10-21 22:32:50 +02:00
Fuxino db4059ff36 Document ~/.ssh/config support for remote backups 2026-10-21 22:31:36 +02:00
Fuxino dc50ef1b07 Read ~/.ssh/config in the SSH pre-check
The paramiko pre-check ignored HostName, Port and IdentityFile, so hosts that rsync's ssh reached through ~/.ssh/config failed before the backup started.
2026-10-21 22:31:01 +02:00
Fuxino fa3ba2a6bd Make inputs and backup_dir optional in the config file
They can come from -i and -o; missing ones were rejected as a bad configuration file (exit code 6).
2026-10-21 22:26:55 +02:00
Fuxino e8b088f13e Use existing config inputs as is instead of globbing them
Paths containing glob characters, such as 'Photos [2020]', matched nothing and were skipped.
2026-10-21 22:23:22 +02:00
Fuxino 535f7a33fe Reject keep values below -1
A keep below -1 made remove_old_backups() delete every backup, including the new one, and then crash.
2026-10-21 22:16:47 +02:00
Fuxino 85bce54b05 Uniform installation instructions in README 2026-10-20 22:01:20 +02:00
Fuxino 125b77a839 Ignore CLAUDE.md 2026-10-18 16:29:34 +02:00
Fuxino 17c104c6c8 Add ruff linting
Configure ruff in pyproject.toml and apply its fixes: modern annotations
(X | None, builtin generics, collections.abc), sorted imports, explicit
stacklevel in test warnings.
2026-10-16 13:47:24 +02:00
Fuxino 7dbb1bcaee Update PKGBUILD
Update to 4.4.1.
2026-10-15 20:34:42 +02:00
13 changed files with 829 additions and 230 deletions
+1
View File
@@ -1,3 +1,4 @@
CLAUDE.md
.idea/
__pycache__/
test/
+3 -3
View File
@@ -4,8 +4,8 @@
pkgname=simple_backup
pkgdesc='Simple backup script that uses rsync to copy files'
pkgver=4.4.0
pkgrel=2
pkgver=4.5.0
pkgrel=1
url="https://git.shouldnt.work/fuxino/$pkgname"
arch=('any')
license=('GPL-3.0-or-later')
@@ -24,7 +24,7 @@ checkdepends=('python-pytest'
conflicts=('simple_backup-git')
source=("git+$url?signed#tag=$pkgver")
validpgpkeys=('7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514')
sha256sums=('cc5944558ed4bc30a0d1a00b49dfced371909dd6c4f39b0c1ccf8f39b445e907')
sha256sums=('c0f7c38d99208e1327c0a6af1d58d5d13797d1cef54f99e1b2fcb4f4a9480609')
build()
{
+17 -2
View File
@@ -22,12 +22,15 @@ sudo pacman -Syu rsync
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
## Installation
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:
Install one with your AUR helper, e.g. `aur sync simple_backup` with aurutils.
### Other distributions
First clone the repository:
```bash
git clone https://git.shouldnt.work/fuxino/simple_backup.git
@@ -59,6 +62,17 @@ The number of old backups to keep can be set with the --keep (or -k) argument (o
Old backups are only removed if rsync completes successfully, so that they are not deleted when the new backup may be incomplete.
The only rsync error that is ignored is return code 24 (source files vanished during the transfer), which is common when backing up a live system.
## Last backup
After each successful backup, simple_backup saves its time and directory to `~/.local/state/simple_backup/last_backup` (or `$XDG_STATE_HOME/simple_backup/last_backup`).
When run with sudo or doas, the file goes to the home directory of the user who ran the command.
Print it with:
```bash
simple_backup --last-backup
```
The file has two lines: the completion time in ISO 8601 format (e.g. `2026-10-25T14:03:12+02:00`) and the backup directory, so scripts (e.g. a status bar indicator) can easily read it to remind you when a backup is overdue.
## Remote backup
> **Warning**
> This feature is experimental
@@ -66,6 +80,7 @@ The only rsync error that is ignored is return code 24 (source files vanished du
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 host can be an alias from `~/.ssh/config`: its HostName, Port and IdentityFile are used (ProxyJump and ProxyCommand are not supported).
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:
+19 -2
View File
@@ -1,4 +1,4 @@
.TH SIMPLE_BACKUP 1 2026-10-15 SIMPLE_BACKUP 4.4.1
.TH SIMPLE_BACKUP 1 2026-10-25 SIMPLE_BACKUP 4.6.0
.SH NAME
simple_backup \- Backup files and folders using rsync
.SH SYNOPSIS
@@ -145,6 +145,11 @@ needs to be allowed to run without password in the same way.
.TP
.B \-\-numeric\-ids
Use rsync \-\-numeric\-ids option. This causes rsync to use numeric uid/gid instead of trying to map uid/gid names from the local machine to the server.
.TP
.B \-\-last\-backup
Print the date and directory of the last successful backup, then exit. The exit status is 0, or 1
if no backup has been recorded yet or the state file can\(aqt be read (see
.BR FILES ).
.SH CONFIGURATION
An example configuration file is provided at \(aq/usr/share/doc/simple_backup/simple_backup.conf\(aq.
Copy it to the default location ($HOME/.config/simple_backup) and edit it as needed.
@@ -152,6 +157,8 @@ 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).
The host can be an alias from \fI~/.ssh/config\fR: its HostName, Port and IdentityFile are used
(ProxyJump and ProxyCommand are not supported).
.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
@@ -207,13 +214,23 @@ 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 value other than 0 or 24).
rsync error (rsync returned a value other than 0 or 24, or rsync is not installed).
.TP
.B 5
SSH connection failed.
.TP
.B 6
Bad configuration file or options (for example, only one of SSH host and SSH user specified).
.SH FILES
.TP
.I $XDG_STATE_HOME/simple_backup/last_backup
Written after each successful backup (local or remote), replacing the previous one. The first line
is the completion time in ISO 8601 format with the UTC offset (for example
\(aq2026\-10\-25T14:03:12+02:00\(aq), the second line the backup directory (prefixed with
\(aquser@host:\(aq for remote backups). The default location is
\fI~/.local/state/simple_backup/last_backup\fR. When running as root with sudo, doas or \-\-user,
the file is written to that user\(aqs \fI~/.local/state\fR and owned by them. Useful for scripts
that remind you when a backup is overdue.
.SH SEE ALSO
.BR rsync (1)
.SH AUTHORS
+6
View File
@@ -57,6 +57,12 @@ warn_unused_ignores = true
module = ["paramiko", "systemd", "dbus"]
ignore_missing_imports = true
[tool.ruff]
line-length = 120
[tool.ruff.lint]
select = ["E", "F", "W", "I", "B", "UP"]
[tool.pytest.ini_options]
testpaths = ["tests"]
# Import the repo copy, not an installed one
+1 -1
View File
@@ -1,3 +1,3 @@
"""Init."""
__version__ = '4.4.1'
__version__ = '4.6.0'
+312 -161
View File
@@ -14,27 +14,28 @@ Classes:
# Avoid evaluating annotations at definition time, as they reference optional modules (paramiko)
from __future__ import annotations
# Import libraries
import sys
import os
import pwd
import io
import re
from typing import Callable, Iterator, List, Optional, ParamSpec, TypedDict, TypeVar
import warnings
from functools import wraps
from shutil import get_terminal_size, rmtree, which
import shlex
import argparse
import configparser
import io
import logging
from logging import StreamHandler
from timeit import default_timer
from subprocess import Popen, PIPE, STDOUT
import os
import pwd
import re
import shlex
import sys
import warnings
from collections import deque
from collections.abc import Callable, Iterator
from datetime import datetime
from tempfile import mkstemp
from functools import wraps
from getpass import GetPassWarning, getpass
from glob import glob
from logging import StreamHandler
from shutil import get_terminal_size, rmtree, which
from subprocess import PIPE, STDOUT, Popen
from tempfile import mkstemp
from timeit import default_timer
from typing import Any, ParamSpec, TypedDict, TypeVar
try:
import paramiko
@@ -204,9 +205,9 @@ class Backup:
Perform the backup
"""
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,
def __init__(self, inputs: list[str] | None, output: str | None, exclude: list[str] | None,
keep: int | None, options: str,
ssh_host: str | None = None, ssh_user: str | None = None, ssh_keyfile: str | None = None,
remote_sudo: bool = False, remove_before: bool = False, verbose: bool = False,
progress: bool = False) -> None:
# Missing values are reported by check_params()
@@ -228,8 +229,9 @@ class Backup:
self._inputs_path = ''
self._exclude_path = ''
self._remote = False
self._ssh: Optional[paramiko.client.SSHClient] = None
self._password_auth = False
self._ssh: paramiko.client.SSHClient | None = None
# Passed to sshpass through rsync's environment only
self._password: str | None = None
self._removed_count = 0
def check_params(self, homedir: str = '') -> int:
@@ -281,7 +283,7 @@ class Backup:
return 0
def _resolve_remote_path(self, path: str) -> Optional[str]:
def _resolve_remote_path(self, path: str) -> str | None:
"""Return the absolute path of a directory on the server, or None if it doesn't exist"""
assert self._ssh is not None
@@ -313,9 +315,50 @@ class Backup:
if self._remote:
self._server = f'{self.ssh_user}@{self.ssh_host}:'
@property
def destination(self) -> str:
"""The current backup directory, prefixed with 'user@host:' for remote backups"""
return f'{self._server}{self._output_dir}'
def remove_old_backups(self) -> None:
"""Remove old backups if there are more than indicated by 'keep'"""
dirs = self._list_backups()
if dirs is None:
return
n_backup = len(dirs)
# The new backup is already in the list, unless old ones are removed before it
if not self._remove_before:
n_backup -= 1
count = 0
if n_backup > self.keep:
logger.info('Removing old backups...')
dirs.sort()
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._remove_backup(dirs[i]):
count += 1
self._removed_count = count
if count == 1:
logger.info('Removed %d backup', count)
elif count > 1:
logger.info('Removed %d backups', count)
def _list_backups(self) -> list[str] | None:
"""Names of the backup directories, or None if they can't be listed (then nothing is removed)"""
if self._remote:
assert self._ssh is not None
@@ -333,85 +376,54 @@ class Backup:
logger.error('Cannot list old backups on the server. Old backups will not be removed')
logger.error(err)
return
return None
dirs = [os.path.basename(d) for d in output.split('\n') if d != '']
return [os.path.basename(d) for d in output.split('\n') if d != '']
n_backup = len(dirs)
try:
# Only count directories, so that stray files don't count as backups
return [f.name for f in os.scandir(f'{self.output}/simple_backup') if f.is_dir(follow_symlinks=False)]
except FileNotFoundError:
return None
except PermissionError:
logger.error('Cannot list old backups. Permission denied. Old backups will not be removed')
if not self._remove_before:
n_backup -= 1
return None
count = 0
def _remove_backup(self, name: str) -> bool:
"""Remove one backup directory; errors are logged"""
if n_backup > self.keep:
logger.info('Removing old backups...')
dirs.sort()
path = f'{self.output}/simple_backup/{name}'
n_remove = n_backup - self.keep
if self._remote:
assert self._ssh is not None
for i in range(n_remove):
logger.info('Removing backup %s (%d/%d)...', dirs[i], i + 1, n_remove)
sudo = 'sudo ' if self.remote_sudo else ''
_, _, stderr = self._ssh.exec_command(f'{sudo}rm -r {shlex.quote(path)}')
if self.remote_sudo:
_, _, stderr = self._ssh.exec_command(
f'sudo rm -r {shlex.quote(f"{self.output}/simple_backup/{dirs[i]}")}')
else:
_, _, stderr = self._ssh.exec_command(
f'rm -r {shlex.quote(f"{self.output}/simple_backup/{dirs[i]}")}')
err = stderr.read().decode('utf-8').strip().split('\n')[0]
err = stderr.read().decode('utf-8').strip().split('\n')[0]
if err != '':
logger.error('Error while removing backup %s.', name)
logger.error(err)
if err != '':
logger.error(
'Error while removing backup %s.', dirs[i])
logger.error(err)
else:
count += 1
else:
try:
# 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 False
return
return True
n_backup = len(dirs)
try:
rmtree(path)
if not self._remove_before:
n_backup -= 1
return True
except FileNotFoundError:
logger.error('Error while removing backup %s. Directory not found', name)
except PermissionError:
logger.error('Error while removing backup %s. Permission denied', name)
except OSError as e:
# E.g. read-only filesystem or device busy
logger.error('Error while removing backup %s. %s', name, e.strerror or e)
count = 0
if n_backup > self.keep:
logger.info('Removing old backups...')
dirs.sort()
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
except FileNotFoundError:
logger.error(
'Error while removing backup %s. Directory not found', dirs[i])
except PermissionError:
logger.error(
'Error while removing backup %s. Permission denied', dirs[i])
self._removed_count = count
if count == 1:
logger.info('Removed %d backup', count)
elif count > 1:
logger.info('Removed %d backups', count)
return False
def find_last_backup(self) -> None:
"""Get path of last backup (from last_backup symlink) for rsync --link-dest"""
@@ -450,7 +462,7 @@ class Backup:
except IndexError:
logger.info('No previous backups available')
def _ssh_connect(self, homedir: str = '') -> Optional[paramiko.client.SSHClient]:
def _ssh_connect(self, homedir: str = '') -> paramiko.client.SSHClient | None:
try:
ssh = paramiko.SSHClient()
except NameError:
@@ -469,7 +481,37 @@ class Backup:
if connected is None:
ssh.close()
def _ssh_login(self, ssh: paramiko.client.SSHClient, homedir: str) -> Optional[paramiko.client.SSHClient]:
def _ssh_config(self, homedir: str) -> dict[str, Any]:
"""connect() arguments for ssh_host from ~/.ssh/config (HostName, Port, IdentityFile), as rsync's ssh uses it"""
assert self.ssh_host is not None
params: dict[str, Any] = {'hostname': self.ssh_host}
try:
host = paramiko.SSHConfig.from_path(f'{homedir}/.ssh/config').lookup(self.ssh_host)
params['hostname'] = host['hostname']
if 'port' in host:
params['port'] = int(host['port'])
except FileNotFoundError:
return params
except (OSError, paramiko.SSHException, ImportError, ValueError) as e:
# ImportError: 'Match exec' needs the invoke module
logger.warning('Cannot read %s/.ssh/config: %s', homedir, e)
return {'hostname': self.ssh_host}
# Like ssh, skip identity files that don't exist
keys = [k for k in host.get('identityfile', []) if os.path.isfile(k)]
if keys:
params['key_filename'] = keys
return params
def _ssh_login(self, ssh: paramiko.client.SSHClient, homedir: str) -> paramiko.client.SSHClient | None:
try:
ssh.load_host_keys(filename=f'{homedir}/.ssh/known_hosts')
except FileNotFoundError:
@@ -477,11 +519,13 @@ class Backup:
ssh.set_missing_host_key_policy(paramiko.WarningPolicy())
host = self._ssh_config(homedir)
try:
# 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, timeout=SSH_TIMEOUT)
ssh.connect(**host, username=self.ssh_user, timeout=SSH_TIMEOUT)
return ssh
except UserWarning:
@@ -507,7 +551,7 @@ class Backup:
pass
try:
ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT)
ssh.connect(**host, username=self.ssh_user, timeout=SSH_TIMEOUT)
return ssh
except paramiko.SSHException:
@@ -521,11 +565,9 @@ class Backup:
password = getpass(
f'{self.ssh_user}@{self.ssh_host}\'s password: ')
ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT,
password=password)
ssh.connect(**host, username=self.ssh_user, timeout=SSH_TIMEOUT, password=password)
self._password_auth = True
os.environ['SSHPASS'] = password
self._password = password
return ssh
except GetPassWarning as e:
@@ -545,7 +587,7 @@ class Backup:
return None
try:
ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT, pkey=pkey)
ssh.connect(**host, username=self.ssh_user, timeout=SSH_TIMEOUT, pkey=pkey)
except paramiko.SSHException:
logger.critical('SSH connection to server failed')
@@ -553,7 +595,7 @@ class Backup:
return ssh
def _load_ssh_key(self) -> Optional[paramiko.PKey]:
def _load_ssh_key(self) -> paramiko.PKey | None:
try:
return paramiko.PKey.from_path(self.ssh_keyfile)
except TypeError:
@@ -656,6 +698,16 @@ class Backup:
_notify('Starting backup...')
# Check before anything is removed (--remove-before-backup)
rsync = which('rsync')
if rsync is None:
logger.critical('rsync not found. Install rsync to perform the backup')
_notify('Backup failed (check log for details)')
return 4
self.define_backup_dir()
fd, self._inputs_path = mkstemp(prefix='tmp_inputs', text=True)
@@ -695,7 +747,7 @@ class Backup:
logger.info('Copying files. This may take a long time...')
# Build the argument list directly, so paths with spaces or quotes are passed unchanged
args = ['/usr/bin/rsync', *self.options.split()]
args = [rsync, *self.options.split()]
if self._last_backup != '':
args.append(f'--link-dest={self._last_backup}')
@@ -703,11 +755,15 @@ class Backup:
args.extend([f'--exclude-from={self._exclude_path}', f'--files-from={self._inputs_path}',
'/', f'{self._server}{self._output_dir}'])
# None inherits our environment
env: dict[str, str] | None = None
# 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'):
elif self._password is not None 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'
env = {**os.environ, 'SSHPASS': self._password}
else:
ssh = 'ssh -o StrictHostKeyChecking=accept-new'
@@ -722,11 +778,12 @@ class Backup:
args.append('--info=progress2')
bar = _ProgressBar()
with Popen(args, stdin=PIPE, stdout=PIPE, stderr=STDOUT, shell=False) as p:
with Popen(args, stdin=PIPE, stdout=PIPE, stderr=STDOUT, shell=False, env=env) as p:
assert p.stdin is not None and isinstance(p.stdout, io.BufferedIOBase)
p.stdin.close()
output = []
# Without -v, only the summary (the last two lines) is logged
output: list[str] | deque[str] = [] if self._verbose else deque(maxlen=2)
try:
for r in _read_records(p.stdout):
@@ -736,27 +793,20 @@ class Backup:
if bar is not None:
bar.clear()
try:
del os.environ['SSHPASS']
except KeyError:
pass
returncode = p.wait()
if returncode == 0:
if self._verbose:
logger.info('rsync: %s', output)
else:
logger.info('rsync: %s', output[-2])
logger.info('rsync: %s', output[-1])
# Without -v, only the summary, which may be fewer than two lines (e.g. with --rsync-options q)
for r in output:
logger.info('rsync: %s', r)
else:
self._returncode_log(returncode)
if self._verbose:
if returncode in [23, 24]:
logger.warning(output)
else:
logger.error(output)
level = logging.WARNING if returncode in [23, 24] else logging.ERROR
for r in output:
logger.log(level, 'rsync: %s', r)
if self._removed_count > 0 and returncode not in [0, 24]:
# With --remove-before-backup, old backups are already gone if the new one failed
@@ -808,7 +858,7 @@ def _read_records(stream: io.BufferedIOBase) -> Iterator[str]:
yield buffer.decode('utf-8', errors='replace')
def _detect_user() -> Optional[str]:
def _detect_user() -> str | None:
if os.geteuid() == 0:
return os.getenv('SUDO_USER') or os.getenv('DOAS_USER')
@@ -858,6 +908,8 @@ def _parse_arguments() -> argparse.Namespace:
help='Run rsync on remote server with sudo if allowed')
parser.add_argument('--numeric-ids', action='store_true',
help='Use rsync \'--numeric-ids\' option (don\'t map uid/gid values by name)')
parser.add_argument('--last-backup', action='store_true',
help='Show when the last successful backup was made and exit')
args = parser.parse_args()
@@ -873,21 +925,24 @@ def _expand_home(path: str, user: str) -> str:
return os.path.expanduser(path)
def _expand_inputs(inputs: List[str], user: Optional[str] = None) -> List[str]:
expanded_inputs: List[str] = []
def _expand_inputs(inputs: list[str], user: str | None = None) -> list[str]:
expanded_inputs: list[str] = []
for i in inputs:
if i == '':
continue
if user is not None:
i_ex = glob(_expand_home(i, user))
path = _expand_home(i, user)
else:
i_ex = glob(i)
path = i
if i.startswith('~'):
logger.warning('Cannot expand \'~\'. No user specified')
# An existing path is used as is, so names with glob characters (e.g. 'Photos [2020]') still work
i_ex = [path] if os.path.lexists(path) else glob(path)
if len(i_ex) == 0:
logger.warning(
'No file or directory matching input %s. Skipping...', i)
@@ -900,13 +955,13 @@ def _expand_inputs(inputs: List[str], user: Optional[str] = None) -> List[str]:
class ConfigArgs(TypedDict):
"""Values read from the configuration file"""
inputs: Optional[List[str]]
output: Optional[str]
exclude: Optional[List[str]]
inputs: list[str] | None
output: str | None
exclude: list[str] | None
keep: int
ssh_host: Optional[str]
ssh_user: Optional[str]
ssh_keyfile: Optional[str]
ssh_host: str | None
ssh_user: str | None
ssh_keyfile: str | None
remote_sudo: bool
numeric_ids: bool
@@ -919,7 +974,7 @@ def _get_bool(config: configparser.ConfigParser, key: str) -> bool:
return config.getboolean('server', key)
def _read_config(config_file: str, user: Optional[str] = None) -> ConfigArgs:
def _read_config(config_file: str, user: str | None = None) -> ConfigArgs:
config_args: ConfigArgs = {'inputs': None,
'output': None,
'exclude': None,
@@ -943,19 +998,22 @@ def _read_config(config_file: str, user: Optional[str] = None) -> ConfigArgs:
config = configparser.ConfigParser(interpolation=None)
config.read(config_file)
section = 'backup'
# Allow compatibility with previous version of config file
try:
inputs = config.get(section, 'inputs')
except configparser.NoSectionError:
if config.has_section('backup'):
section = 'backup'
elif config.has_section('default'):
section = 'default'
inputs = config.get(section, 'inputs')
else:
raise configparser.NoSectionError('backup')
config_args['inputs'] = list(set(_expand_inputs([i.strip() for i in inputs.split(',')], user)))
# inputs and backup_dir are optional, since -i and -o can give them. Missing ones are reported by check_params()
inputs = config.get(section, 'inputs', fallback=None)
if inputs is not None:
config_args['inputs'] = list(set(_expand_inputs([i.strip() for i in inputs.split(',')], user)))
# '~' is expanded later, since for remote backups it refers to the home on the server
config_args['output'] = config.get(section, 'backup_dir')
config_args['output'] = config.get(section, 'backup_dir', fallback=None)
exclude = config.get(section, 'exclude', fallback='')
config_args['exclude'] = [e.strip() for e in exclude.split(',') if e.strip() != '']
@@ -1027,31 +1085,119 @@ def _notify(text: str) -> None:
os.seteuid(euid)
def _state_owner(user: str | None) -> pwd.struct_passwd | None:
"""The user whose state file is used when running as root on their behalf (sudo, doas, -u)"""
if os.geteuid() != 0 or user is None:
return None
try:
owner = pwd.getpwnam(user)
except KeyError:
return None
return owner if owner.pw_uid != 0 else None
def _state_file(owner: pwd.struct_passwd | None) -> str:
if owner is not None:
# XDG_STATE_HOME belongs to root here
base = os.path.join(owner.pw_dir, '.local', 'state')
else:
base = os.getenv('XDG_STATE_HOME', '')
# The XDG spec says to ignore relative paths
if not os.path.isabs(base):
base = os.path.expanduser('~/.local/state')
return os.path.join(base, 'simple_backup', 'last_backup')
def _write_state(user: str | None, destination: str) -> None:
"""Record the time and directory of the last successful backup (two lines)"""
owner = _state_owner(user)
path = _state_file(owner)
tmp = ''
euid, egid = os.geteuid(), os.getegid()
# Create the file (and any missing directories) as the user, not root
if owner is not None:
os.setegid(owner.pw_gid)
os.seteuid(owner.pw_uid)
try:
os.makedirs(os.path.dirname(path), exist_ok=True)
# Replace the file in one step, so readers (e.g. status bar scripts) never see it half written
fd, tmp = mkstemp(prefix='.last_backup', dir=os.path.dirname(path), text=True)
with os.fdopen(fd, 'w', encoding='utf-8') as fp:
fp.write(f'{datetime.now().astimezone().isoformat(timespec="seconds")}\n{destination}\n')
os.replace(tmp, path)
except OSError as e:
logger.warning('Cannot save the time of the last backup to %s: %s', path, e)
if tmp != '':
try:
os.remove(tmp)
except OSError:
pass
finally:
if owner is not None:
os.seteuid(euid)
os.setegid(egid)
def _show_last_backup(user: str | None) -> int:
path = _state_file(_state_owner(user))
try:
with open(path, encoding='utf-8') as fp:
time, destination = fp.read().splitlines()[:2]
completed = datetime.fromisoformat(time).astimezone()
except FileNotFoundError:
print('No backup recorded yet')
return 1
except (OSError, ValueError) as e:
print(f'Cannot read {path}: {e}', file=sys.stderr)
return 1
days = (datetime.now().astimezone().date() - completed.date()).days
if days == 0:
age = 'today'
elif days == 1:
age = 'yesterday'
else:
age = f'{days} days ago'
print(f'Last backup: {completed:%Y-%m-%d %H:%M} ({age})')
print(f'Directory: {destination}')
return 0
def simple_backup() -> int:
"""Main"""
args = _parse_arguments()
if args.user:
user = args.user
homedir = os.path.expanduser(f'~{user}')
user: str | None = args.user or _detect_user()
if args.last_backup:
return _show_last_backup(user)
if not args.user and os.geteuid() != 0:
homedir = os.getenv('HOME', '')
elif user is not None:
homedir = _expand_home('~', user)
else:
euid = os.geteuid()
if euid == 0:
user = _detect_user()
if user is not None:
homedir = os.path.expanduser(f'~{user}')
else:
logger.warning(
'Failed to detect user. You can use -u/--user parameter to manually specify it')
homedir = None
else:
user = os.getenv('USER')
homedir = os.getenv('HOME')
if homedir is None:
logger.warning('Failed to detect user. You can use -u/--user parameter to manually specify it')
homedir = ''
if args.no_syslog:
@@ -1060,13 +1206,8 @@ def simple_backup() -> int:
except NameError:
pass
config_file = args.config
# Expand '~' to the home of the user running the backup, not root's
if config_file.startswith('~/') and user is not None:
config_file = f'~{user}{config_file[1:]}'
config_file = os.path.expanduser(config_file)
config_file = _expand_home(args.config, user) if user is not None else os.path.expanduser(args.config)
try:
config_args = _read_config(config_file, user)
@@ -1082,6 +1223,11 @@ def simple_backup() -> int:
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 keep < -1:
logger.critical('Invalid number of backups to keep: %d (use -1 to keep all)', keep)
return 6
if (ssh_host is None) != (ssh_user is None):
logger.critical('Both SSH host and SSH user are needed for remote backup')
@@ -1129,8 +1275,13 @@ def simple_backup() -> int:
try:
return_code = backup.check_params(homedir)
if return_code != 0:
return return_code
return_code = backup.run()
if return_code == 0:
return backup.run()
_write_state(user, backup.destination)
return return_code
finally:
@@ -1138,4 +1289,4 @@ def simple_backup() -> int:
if __name__ == '__main__':
simple_backup()
sys.exit(simple_backup())
+27 -3
View File
@@ -1,7 +1,9 @@
"""Shared fixtures for the simple_backup tests"""
import sys
from typing import Callable
from collections.abc import Callable, Iterator
from datetime import datetime, timedelta
from pathlib import Path
import pytest
@@ -9,13 +11,35 @@ 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"""
def no_side_effects(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Keep tests out of the systemd journal, the desktop notifications and the user's state directory"""
if hasattr(sb, 'j_handler'):
sb.logger.removeHandler(sb.j_handler)
monkeypatch.setattr(sb, '_notify', lambda text: None)
monkeypatch.setenv('XDG_STATE_HOME', str(tmp_path / 'state'))
@pytest.fixture
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
+69 -12
View File
@@ -2,8 +2,11 @@
import os
import pwd
import subprocess
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Any, Callable, Optional
from typing import Any
import pytest
@@ -23,10 +26,12 @@ class FakeBackup:
instances: list['FakeBackup'] = []
check_code = 0
run_code = 0
destination = '/backups/simple_backup/2026-10-25_10-00-00'
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,
ssh_host: str | None = None, ssh_user: str | None = None,
ssh_keyfile: str | None = None, remote_sudo: bool = False,
remove_before: bool = False, verbose: bool = False, progress: bool = False) -> None:
self.inputs = inputs
self.output = output
@@ -40,7 +45,7 @@ class FakeBackup:
self.remove_before = remove_before
self.verbose = verbose
self.progress = progress
self.homedir: Optional[str] = None
self.homedir: str | None = None
self.ran = False
self.closed = False
FakeBackup.instances.append(self)
@@ -53,7 +58,7 @@ class FakeBackup:
def run(self) -> int:
self.ran = True
return 0
return FakeBackup.run_code
def close(self) -> None:
self.closed = True
@@ -63,6 +68,7 @@ class FakeBackup:
def fake_backup(monkeypatch: pytest.MonkeyPatch) -> None:
FakeBackup.instances = []
FakeBackup.check_code = 0
FakeBackup.run_code = 0
monkeypatch.setattr(sb, 'Backup', FakeBackup)
@@ -80,19 +86,21 @@ def as_root(monkeypatch: pytest.MonkeyPatch) -> None:
"""Run as root, with no sudo/doas variables set"""
monkeypatch.setattr(os, 'geteuid', lambda: 0)
# Writing the state file as root would call seteuid() (covered in test_state.py)
monkeypatch.setattr(sb, '_write_state', lambda user, destination: None)
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]]]:
def config_calls(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str | None]]:
"""Record the config file path and user instead of reading the file"""
calls: list[tuple[str, Optional[str]]] = []
calls: list[tuple[str, str | None]] = []
read_config = sb._read_config
def record(config_file: str, user: Optional[str] = None) -> sb.ConfigArgs:
def record(config_file: str, user: str | None = None) -> sb.ConfigArgs:
calls.append((config_file, user))
return read_config('/nonexistent/simple_backup.conf', user)
@@ -169,6 +177,15 @@ ssh_keyfile=/keys/id
assert b.keep == 0
assert (b.ssh_host, b.ssh_user, b.ssh_keyfile) == ('cli-host', 'cli-user', '/cli/key')
def test_inputs_and_output_only_from_cli(self, run_cli: RunCli, tmp_path: Path) -> None:
config_file = tmp_path / 'simple_backup.conf'
config_file.write_text('[backup]\nkeep=3\n')
assert run_cli('-c', str(config_file), '-i', '/cli/in', '-o', '/cli/out') == 0
b = last_backup()
assert (b.inputs, b.output, b.keep) == (['/cli/in'], '/cli/out', 3)
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')
@@ -190,6 +207,18 @@ ssh_keyfile=/keys/id
assert FakeBackup.instances == []
assert 'Both SSH host and SSH user are needed' in caplog.text
@pytest.mark.parametrize('config_keep, argv', [
('keep=-2', []),
('', ['-k', '-2']),
])
def test_keep_below_minus_one(self, run_cli: RunCli, config: Callable[[str], str],
caplog: pytest.LogCaptureFixture, config_keep: str, argv: list[str]) -> None:
config_file = config(f'backup_dir=/backups\n{config_keep}')
assert run_cli('-c', config_file, *argv) == 6
assert FakeBackup.instances == []
assert 'Invalid number of backups to keep: -2' 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')
@@ -246,23 +275,37 @@ ssh_keyfile=/keys/id
assert not b.ran
assert b.closed
@pytest.mark.parametrize('check_code, run_code, saved', [(0, 0, True), (2, 0, False), (0, 4, False)])
def test_state_saved_only_on_success(self, run_cli: RunCli, config: Callable[[str], str], tmp_path: Path,
check_code: int, run_code: int, saved: bool) -> None:
FakeBackup.check_code = check_code
FakeBackup.run_code = run_code
assert run_cli('-c', config('backup_dir=/backups')) == (check_code or run_code)
state = tmp_path / 'state' / 'simple_backup' / 'last_backup'
assert state.exists() is saved
if saved:
assert state.read_text().splitlines()[1] == FakeBackup.destination
class TestUser:
@pytest.mark.usefixtures('as_user')
def test_current_user(self, run_cli: RunCli, config_calls: list[tuple[str, Optional[str]]]) -> None:
def test_current_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]]) -> 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:
def test_explicit_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]]) -> 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]]],
def test_sudo_or_doas_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]],
monkeypatch: pytest.MonkeyPatch, var: str) -> None:
monkeypatch.setenv(var, USER)
@@ -271,7 +314,7 @@ class TestUser:
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]]],
def test_root_without_user(self, run_cli: RunCli, config_calls: list[tuple[str, str | None]],
caplog: pytest.LogCaptureFixture) -> None:
assert run_cli() == 0
# '~' is left to expanduser, i.e. $HOME
@@ -303,6 +346,20 @@ class TestDetectUser:
assert sb._detect_user() == USER
def test_run_as_script(tmp_path: Path) -> None:
"""Running the module directly passes simple_backup()'s return code to the shell"""
repo = Path(sb.__file__).resolve().parents[1]
env = {**os.environ, 'PYTHONPATH': str(repo), 'HOME': str(tmp_path)}
# No config file and no inputs: check_params() returns 1
result = subprocess.run([sys.executable, '-m', 'simple_backup.simple_backup', '--no-syslog',
'-c', str(tmp_path / 'missing.conf')], env=env, cwd=tmp_path,
capture_output=True, check=False)
assert result.returncode == 1
class TestTargetUid:
@pytest.mark.usefixtures('as_user')
def test_non_root(self) -> None:
+15 -2
View File
@@ -45,6 +45,13 @@ def test_expand_inputs_glob(tmp_path: Path) -> None:
assert sorted(result) == [f'{tmp_path}/a.txt', f'{tmp_path}/b.txt']
def test_expand_inputs_glob_characters_in_name(tmp_path: Path) -> None:
(tmp_path / 'Photos [2020]').mkdir()
(tmp_path / 'Photos 2').mkdir()
assert sb._expand_inputs([f'{tmp_path}/Photos [2020]'], USER) == [f'{tmp_path}/Photos [2020]']
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
@@ -217,9 +224,15 @@ backup_dir=/backups
sb._read_config(config_file, USER)
def test_inputs_and_backup_dir_optional(tmp_path: Path) -> None:
config_args = sb._read_config(write_config(tmp_path, '[backup]\nkeep=2\n'), USER)
assert config_args['inputs'] is None
assert config_args['output'] is None
assert config_args['keep'] == 2
@pytest.mark.parametrize('text', [
'[backup]\nbackup_dir=/backups\n',
'[backup]\ninputs=/tmp\n',
'[other]\ninputs=/tmp\n',
'inputs=/tmp\n',
])
+63 -27
View File
@@ -1,10 +1,12 @@
"""Tests for local backups, with real rsync runs and with a fake Popen"""
import errno
import io
import os
from datetime import datetime, timedelta
from collections.abc import Callable
from pathlib import Path
from typing import Any, Callable, Iterator, Optional
from shutil import which
from typing import Any
import pytest
@@ -13,25 +15,7 @@ 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)
pytestmark = pytest.mark.usefixtures('fake_now')
@pytest.fixture
@@ -52,10 +36,10 @@ def out(tmp_path: Path) -> Path:
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:
def make_backup(out: Path, inputs: list[str], keep: int = -1, exclude: list[str] | None = None,
remove_before: bool = False, progress: bool = False, verbose: bool = False) -> sb.Backup:
backup = sb.Backup(inputs, str(out), exclude or [], keep, OPTIONS, remove_before=remove_before,
progress=progress)
progress=progress, verbose=verbose)
assert backup.check_params() == 0
@@ -160,6 +144,19 @@ class TestRsync:
# Nothing to link against, as the old backup was removed first
assert copied(out, backup, src / 'same.txt').stat().st_nlink == 1
def test_removal_os_error(self, src: Path, out: Path, monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture) -> None:
assert run_backup(out, [str(src)]) == 0
def busy(path: str) -> None:
raise OSError(errno.EBUSY, os.strerror(errno.EBUSY), path)
monkeypatch.setattr(sb, 'rmtree', busy)
assert run_backup(out, [str(src)], keep=0) == 0
assert backups(out) == ['2026-10-14_10-00-00', '2026-10-14_11-00-00']
assert 'Error while removing backup 2026-10-14_10-00-00. Device or resource busy' in caplog.text
def test_stray_files_and_symlinks(self, src: Path, out: Path) -> None:
assert run_backup(out, [str(src)]) == 0
@@ -204,7 +201,11 @@ class TestRsync:
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']
# The fake clock also advances when the state file is written
assert backups(out) == ['2026-10-14_12-00-00']
state = (tmp_path / 'state' / 'simple_backup' / 'last_backup').read_text().splitlines()
assert state[0].startswith('2026-10-14T13:00:00')
assert state[1] == f'{out}/simple_backup/2026-10-14_12-00-00'
class ChunkStream(io.BufferedIOBase):
@@ -282,12 +283,13 @@ class FakePopen:
returncode = 0
calls: list[list[str]] = []
output = b'sending incremental file list\nsent 1 bytes received 2 bytes\ntotal size is 3\n'
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')
self.stdout = io.BytesIO(FakePopen.output)
def __enter__(self) -> 'FakePopen':
return self
@@ -323,10 +325,44 @@ class TestReturnCodes:
assert run_backup(out, [str(src)]) == 0
(args,) = fake_popen.calls
assert args[0] == '/usr/bin/rsync'
assert args[0] == which('rsync')
assert f'--link-dest={out}/simple_backup/{old_backups[-1]}' in args
assert args[args.index('/') + 1] == f'{out}/simple_backup/2026-10-14_10-00-00'
@pytest.mark.parametrize('lines, logged', [
([f'file{n}' for n in range(5)] + ['sent 1 bytes', 'total size is 3'], ['sent 1 bytes', 'total size is 3']),
([], []),
(['total size is 3'], ['total size is 3']),
])
def test_summary_logged(self, src: Path, out: Path, fake_popen: type[FakePopen], monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture, lines: list[str], logged: list[str]) -> None:
monkeypatch.setattr(fake_popen, 'output', ''.join(f'{line}\n' for line in lines).encode())
assert run_backup(out, [str(src)]) == 0
assert [m[7:] for m in caplog.messages if m.startswith('rsync: ')] == logged
@pytest.mark.parametrize('returncode, level', [(0, 'INFO'), (23, 'WARNING'), (1, 'ERROR')])
def test_verbose_one_line_per_message(self, src: Path, out: Path, fake_popen: type[FakePopen],
caplog: pytest.LogCaptureFixture, returncode: int, level: str) -> None:
fake_popen.returncode = returncode
run_backup(out, [str(src)], verbose=True)
rsync = [(r.levelname, r.getMessage()) for r in caplog.records if r.getMessage().startswith('rsync: ')]
lines = ['sending incremental file list', 'sent 1 bytes received 2 bytes', 'total size is 3']
assert rsync == [(level, f'rsync: {line}') for line in lines]
def test_rsync_missing(self, src: Path, out: Path, tmp_path: Path, fake_popen: type[FakePopen],
old_backups: list[str], monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture) -> None:
monkeypatch.setenv('PATH', str(tmp_path / 'empty'))
assert run_backup(out, [str(src)], keep=0, remove_before=True) == 4
assert fake_popen.calls == []
# Nothing was removed
assert backups(out) == old_backups
assert 'rsync not found' in caplog.text
@pytest.mark.parametrize('returncode, result, removed', [
(0, 0, True),
(24, 0, True),
+159 -17
View File
@@ -10,7 +10,7 @@ import warnings
from getpass import GetPassWarning
from pathlib import Path
from shutil import which
from typing import Any, Optional
from typing import Any
import pytest
@@ -30,7 +30,7 @@ class FakeSSH:
self.home = home
self.bin_dir = bin_dir
self.commands: list[str] = []
self.fail_on: Optional[str] = None
self.fail_on: str | None = None
self.closed = False
def exec_command(self, command: str) -> tuple[None, io.BytesIO, io.BytesIO]:
@@ -186,7 +186,7 @@ class TestRemoveOldBackups:
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:
def test_nothing_to_remove(self, remote_home: Path, names: list[str] | None) -> None:
if names is not None:
make_backups(remote_home, *names)
@@ -254,9 +254,11 @@ class RecordPopen:
returncode = 0
calls: list[list[str]] = []
envs: list[dict[str, str] | None] = []
def __init__(self, args: list[str], **kwargs: Any) -> None:
RecordPopen.calls.append(args)
RecordPopen.envs.append(kwargs.get('env'))
self.stdin = io.BytesIO()
self.stdout = io.BytesIO(b'sent 1 bytes received 2 bytes\ntotal size is 3\n')
@@ -276,6 +278,7 @@ class TestRsyncArgs:
def record_popen(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
RecordPopen.returncode = 0
RecordPopen.calls = []
RecordPopen.envs = []
monkeypatch.setattr(sb, 'Popen', RecordPopen)
# The input must exist locally
(tmp_path / 'input').mkdir()
@@ -322,19 +325,26 @@ class TestRsyncArgs:
def test_password_auth(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(sb, 'which', lambda cmd: f'/usr/bin/{cmd}')
monkeypatch.setenv('SSHPASS', 'secret')
monkeypatch.delenv('SSHPASS', raising=False)
backup = connected_backup()
backup.inputs = [str(tmp_path / 'input')]
backup._password_auth = True
backup._password = 'secret'
assert backup.run() == 0
(args,) = RecordPopen.calls
assert shlex.split(self.ssh_command(args)) == ['sshpass', '-e', 'ssh', '-l', 'user', '-o',
'StrictHostKeyChecking=accept-new']
# The password must not stay in the environment
# Only rsync gets the password, never our own environment
(env,) = RecordPopen.envs
assert env is not None and env['SSHPASS'] == 'secret'
assert 'SSHPASS' not in os.environ
def test_no_password_env(self, tmp_path: Path) -> None:
self.run(tmp_path)
assert RecordPopen.envs == [None]
def test_failure(self, tmp_path: Path, remote_home: Path) -> None:
base = make_backups(remote_home, '2026-10-01_10-00-00')
RecordPopen.returncode = 23
@@ -348,11 +358,11 @@ class TestRsyncArgs:
class FakeClient:
"""paramiko.SSHClient stand-in; connect() takes its outcomes from a list"""
def __init__(self, outcomes: list[Optional[BaseException]]) -> None:
def __init__(self, outcomes: list[BaseException | None]) -> None:
self.outcomes = outcomes
self.connects: list[dict[str, Any]] = []
self.policies: list[str] = []
self.known_hosts: Optional[str] = None
self.known_hosts: str | None = None
self.closed = False
def load_host_keys(self, filename: str) -> None:
@@ -365,12 +375,12 @@ class FakeClient:
self.policies.append(type(policy).__name__)
def connect(self, hostname: str, **kwargs: Any) -> None:
self.connects.append(kwargs)
self.connects.append({'hostname': hostname, **kwargs})
outcome = self.outcomes.pop(0)
if isinstance(outcome, UserWarning):
# Like WarningPolicy for an unknown host key
warnings.warn(outcome)
warnings.warn(outcome, stacklevel=2)
elif outcome is not None:
raise outcome
@@ -404,8 +414,35 @@ class TestLogin:
assert remote_backup()._ssh_connect(str(tmp_path)) is client
assert client.known_hosts == f'{tmp_path}/.ssh/known_hosts'
assert client.connects[0]['timeout'] == sb.SSH_TIMEOUT
assert client.connects[0]['hostname'] == 'server'
assert 'port' not in client.connects[0]
assert not client.closed
def test_ssh_config(self, client: FakeClient, tmp_path: Path) -> None:
(tmp_path / '.ssh').mkdir()
key = tmp_path / '.ssh' / 'id_backup'
key.touch()
(tmp_path / '.ssh' / 'config').write_text(
f'Host server\n HostName backup.example.org\n Port 2222\n'
f' IdentityFile {key}\n IdentityFile {tmp_path}/.ssh/missing\n')
client.outcomes = [ssh_error(), None]
assert remote_backup()._ssh_connect(str(tmp_path)) is client
for params in client.connects:
assert (params['hostname'], params['port']) == ('backup.example.org', 2222)
assert params['key_filename'] == [str(key)]
def test_bad_ssh_config(self, client: FakeClient, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
(tmp_path / '.ssh').mkdir()
(tmp_path / '.ssh' / 'config').write_text('Host server\n Port abc\n')
client.outcomes = [None]
assert remote_backup()._ssh_connect(str(tmp_path)) is client
assert client.connects[0]['hostname'] == 'server'
assert 'port' not in client.connects[0]
assert 'Cannot read' in caplog.text
@pytest.mark.parametrize('answer, connected', [('y', True), (' Yes', True), ('n', False), ('', False)])
def test_unknown_host_key(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch,
answer: str, connected: bool) -> None:
@@ -444,13 +481,13 @@ class TestLogin:
def test_password(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch) -> None:
client.outcomes = [ssh_error(), ssh_error(), None]
monkeypatch.setattr(sb, 'getpass', lambda prompt: 'secret')
monkeypatch.setenv('SSHPASS', '')
monkeypatch.delenv('SSHPASS', raising=False)
backup = remote_backup()
assert backup._ssh_connect() is client
assert client.connects[-1]['password'] == 'secret'
assert backup._password_auth
assert os.environ['SSHPASS'] == 'secret'
assert backup._password == 'secret'
assert 'SSHPASS' not in os.environ
def test_wrong_password(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch) -> None:
client.outcomes = [ssh_error(), ssh_error(), ssh_error()]
@@ -462,7 +499,7 @@ class TestLogin:
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)
warnings.warn('Can not control echo on the terminal.', GetPassWarning, stacklevel=2)
return ''
@@ -474,7 +511,7 @@ class TestLogin:
@pytest.mark.parametrize('outcome, connected', [(None, True), ('error', False)])
def test_keyfile(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch,
outcome: Optional[str], connected: bool) -> None:
outcome: str | None, connected: bool) -> None:
pkey = object()
client.outcomes = [ssh_error(), ssh_error(), ssh_error() if outcome else None]
monkeypatch.setattr(sb.Backup, '_load_ssh_key', lambda self: pkey)
@@ -548,7 +585,7 @@ class TestLoadKey:
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)
warnings.warn('Can not control echo on the terminal.', GetPassWarning, stacklevel=2)
return ''
@@ -558,7 +595,7 @@ class TestLoadKey:
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:
def test_bad_file(self, tmp_path: Path, caplog: pytest.LogCaptureFixture, content: str | None) -> None:
keyfile = tmp_path / 'id'
if content is not None:
@@ -566,3 +603,108 @@ class TestLoadKey:
assert remote_backup(ssh_keyfile=str(keyfile))._load_ssh_key() is None
assert 'Unable to load SSH key' in caplog.text
FAKE_SSH = """#!/bin/sh
# Record the arguments, then run the remote command locally in the fake remote home
for a in "$@"; do printf '%s\\n' "$a"; done >> "{log}"
echo --- >> "{log}"
while [ $# -gt 0 ]; do
case $1 in
-i|-l|-o|-p) shift 2 ;;
-*) shift ;;
*) break ;;
esac
done
# The host
shift
cd "{home}" && HOME="{home}" exec sh -c "$*"
"""
FAKE_SSHPASS = """#!/bin/sh
printf '%s' "${{SSHPASS-unset}}" > "{log}"
# -e
shift
exec "$@"
"""
@pytest.mark.usefixtures('connect', 'fake_now')
class TestRsyncOverSsh:
"""Whole remote backups: real rsync, with fake ssh and sshpass commands that run the remote side locally"""
@pytest.fixture(autouse=True)
def fake_ssh(self, ssh: FakeSSH, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
self.ssh_log = tmp_path / 'ssh.log'
self.sshpass_log = tmp_path / 'sshpass.log'
for name, script, log in [('ssh', FAKE_SSH, self.ssh_log), ('sshpass', FAKE_SSHPASS, self.sshpass_log)]:
path = ssh.bin_dir / name
path.write_text(script.format(log=log, home=ssh.home))
path.chmod(0o755)
monkeypatch.setenv('PATH', f'{ssh.bin_dir}:{os.environ["PATH"]}')
monkeypatch.delenv('SSHPASS', raising=False)
self.src = tmp_path / 'src dir'
self.src.mkdir()
(self.src / 'same.txt').write_text('unchanged')
(self.src / 'changed.txt').write_text('first')
def run(self, keep: int = -1, password: str | None = None, **kwargs: Any) -> int:
backup = connected_backup(keep=keep, **kwargs)
backup.inputs = [str(self.src)]
backup._password = password
return backup.run()
def ssh_calls(self) -> list[list[str]]:
return [c.strip('\n').split('\n') for c in self.ssh_log.read_text().split('---\n') if c != '']
def copied(self, remote_home: Path, backup: str, name: str) -> Path:
return remote_home / 'backups' / 'simple_backup' / backup / str(self.src).lstrip('/') / name
def test_backups(self, remote_home: Path) -> None:
assert self.run() == 0
(self.src / 'changed.txt').write_text('second')
assert self.run(keep=1) == 0
assert self.run(keep=1) == 0
old, new = '2026-10-14_11-00-00', '2026-10-14_12-00-00'
assert remote_dirs(remote_home / 'backups' / 'simple_backup') == [old, new]
assert self.copied(remote_home, new, 'changed.txt').read_text() == 'second'
# Unchanged files are hard-linked to the previous backup (--link-dest on the server)
assert self.copied(remote_home, old, 'same.txt').stat().st_ino == \
self.copied(remote_home, new, 'same.txt').stat().st_ino
for call in self.ssh_calls():
assert call[:5] == ['-o', 'StrictHostKeyChecking=accept-new', '-l', 'user', 'server']
# rsync passes the remote command as separate arguments
assert call[5:7] == ['rsync', '--server']
def test_keyfile(self, remote_home: Path) -> None:
assert self.run(ssh_keyfile='/keys/my key') == 0
(call,) = self.ssh_calls()
assert call[:4] == ['-i', '/keys/my key', '-o', 'StrictHostKeyChecking=accept-new']
def test_password(self, remote_home: Path) -> None:
assert self.run(password='secret') == 0
assert self.sshpass_log.read_text() == 'secret'
assert 'SSHPASS' not in os.environ
assert len(remote_dirs(remote_home / 'backups' / 'simple_backup')) == 1
def test_remote_sudo(self, remote_home: Path) -> None:
assert self.run() == 0
assert self.run(keep=0, remote_sudo=True) == 0
assert remote_dirs(remote_home / 'backups' / 'simple_backup') == ['2026-10-14_11-00-00']
assert self.ssh_calls()[-1][5:7] == ['sudo rsync', '--server']
def test_progress(self, remote_home: Path, capsys: pytest.CaptureFixture[str]) -> None:
assert self.run(progress=True) == 0
assert '100%' in capsys.readouterr().err
+137
View File
@@ -0,0 +1,137 @@
"""Tests for the last backup state file and --last-backup"""
import os
import pwd
from collections.abc import Callable
from datetime import datetime, timedelta
from pathlib import Path
import pytest
from simple_backup import simple_backup as sb
DESTINATION = '/backups/simple_backup/2026-10-25_10-00-00'
@pytest.fixture(autouse=True)
def as_user(monkeypatch: pytest.MonkeyPatch) -> None:
"""Run as a non-root user (the tests may run as root in a chroot)"""
monkeypatch.setattr(os, 'geteuid', lambda: 1000)
monkeypatch.setenv('USER', 'user')
@pytest.fixture
def state(tmp_path: Path) -> Path:
return tmp_path / 'state' / 'simple_backup' / 'last_backup'
def write_state(state: Path, text: str) -> None:
state.parent.mkdir(parents=True)
state.write_text(text)
def test_write(state: Path) -> None:
sb._write_state(None, DESTINATION)
time, destination = state.read_text().splitlines()
assert abs(datetime.now().astimezone() - datetime.fromisoformat(time)) < timedelta(minutes=1)
assert destination == DESTINATION
# No temporary file left behind
assert os.listdir(state.parent) == ['last_backup']
def test_overwrite(state: Path) -> None:
write_state(state, '2026-10-01T10:00:00+02:00\n/old\n')
sb._write_state(None, DESTINATION)
assert state.read_text().splitlines()[1] == DESTINATION
def test_relative_xdg_state_home_ignored(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setenv('XDG_STATE_HOME', 'relative')
monkeypatch.setenv('HOME', str(tmp_path / 'home'))
sb._write_state(None, DESTINATION)
assert (tmp_path / 'home' / '.local' / 'state' / 'simple_backup' / 'last_backup').exists()
def test_write_error(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
# A file where the state directory should be
(tmp_path / 'not a dir').write_text('')
monkeypatch.setenv('XDG_STATE_HOME', str(tmp_path / 'not a dir'))
sb._write_state(None, DESTINATION)
assert 'Cannot save the time of the last backup' in caplog.text
def test_root_writes_as_user(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
calls: list[tuple[str, int]] = []
egid = os.getegid()
owner = pwd.struct_passwd(('user', 'x', 1000, 1001, '', str(tmp_path / 'home'), '/bin/sh'))
monkeypatch.setattr(os, 'geteuid', lambda: 0)
monkeypatch.setattr(pwd, 'getpwnam', lambda name: owner)
monkeypatch.setattr(os, 'seteuid', lambda uid: calls.append(('euid', uid)))
monkeypatch.setattr(os, 'setegid', lambda gid: calls.append(('egid', gid)))
sb._write_state('user', DESTINATION)
# Root's XDG_STATE_HOME is not used
assert (tmp_path / 'home' / '.local' / 'state' / 'simple_backup' / 'last_backup').exists()
assert calls == [('egid', 1001), ('euid', 1000), ('euid', 0), ('egid', egid)]
class TestStateOwner:
def test_non_root(self) -> None:
assert sb._state_owner('user') is None
@pytest.mark.parametrize('name, expected', [('user', 'user'), ('root', None), ('unknown', None), (None, None)])
def test_root(self, monkeypatch: pytest.MonkeyPatch, name: str | None, expected: str | None) -> None:
users = {'user': pwd.struct_passwd(('user', 'x', 1000, 1000, '', '/home/user', '/bin/sh')),
'root': pwd.struct_passwd(('root', 'x', 0, 0, '', '/root', '/bin/sh'))}
def getpwnam(name: str) -> pwd.struct_passwd:
if name not in users:
raise KeyError(name)
return users[name]
monkeypatch.setattr(os, 'geteuid', lambda: 0)
monkeypatch.setattr(pwd, 'getpwnam', getpwnam)
owner = sb._state_owner(name)
assert (owner.pw_name if owner is not None else None) == expected
class TestShow:
@pytest.mark.parametrize('days, age', [(0, 'today'), (1, 'yesterday'), (3, '3 days ago')])
def test_age(self, run_cli: Callable[..., int], state: Path, capsys: pytest.CaptureFixture[str],
days: int, age: str) -> None:
completed = (datetime.now() - timedelta(days=days)).replace(hour=12, minute=30).astimezone()
write_state(state, f'{completed.isoformat(timespec="seconds")}\n{DESTINATION}\n')
assert run_cli('--last-backup') == 0
assert capsys.readouterr().out == \
f'Last backup: {completed:%Y-%m-%d} 12:30 ({age})\nDirectory: {DESTINATION}\n'
def test_after_write(self, run_cli: Callable[..., int], capsys: pytest.CaptureFixture[str]) -> None:
sb._write_state(None, DESTINATION)
assert run_cli('--last-backup') == 0
assert '(today)' in capsys.readouterr().out
def test_no_backup(self, run_cli: Callable[..., int], capsys: pytest.CaptureFixture[str]) -> None:
assert run_cli('--last-backup') == 1
assert capsys.readouterr().out == 'No backup recorded yet\n'
@pytest.mark.parametrize('text', ['', '2026-10-25T10:00:00+02:00\n', 'yesterday\n/backups\n'])
def test_bad_file(self, run_cli: Callable[..., int], state: Path, capsys: pytest.CaptureFixture[str],
text: str) -> None:
write_state(state, text)
assert run_cli('--last-backup') == 1
assert 'Cannot read' in capsys.readouterr().err