Compare commits

..
9 Commits
Author SHA1 Message Date
Fuxino c2ef47e4ea Bump version to 4.6.2 2026-11-02 20:56:33 +01:00
Fuxino ec27b8d77f Estimate progress while rsync scans from the last backup's file count
With incremental recursion rsync only knows the total near the end, so the bar showed
no ETA for most of the backup. Save the file count as a third line of the state file
and use it as an estimate until rsync reports the real total.
2026-11-02 20:56:12 +01:00
Fuxino 5e33781974 Update PKGBUILD 2026-11-02 20:26:10 +01:00
Fuxino fdb2fc960a Bump version to 4.6.1 2026-11-02 20:18:52 +01:00
Fuxino c5cfe49758 Show elapsed time as (h:)mm:ss and add it to the notification
Replace the timing decorator with a start time kept by Backup.run(), so the final
notification can report how long the backup took.
2026-11-02 19:54:11 +01:00
Fuxino 8d0fd46f44 Update PKGBUILD 2026-10-25 12:42:54 +01:00
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
9 changed files with 483 additions and 48 deletions
+2 -2
View File
@@ -4,7 +4,7 @@
pkgname=simple_backup pkgname=simple_backup
pkgdesc='Simple backup script that uses rsync to copy files' pkgdesc='Simple backup script that uses rsync to copy files'
pkgver=4.4.1 pkgver=4.6.1
pkgrel=1 pkgrel=1
url="https://git.shouldnt.work/fuxino/$pkgname" url="https://git.shouldnt.work/fuxino/$pkgname"
arch=('any') arch=('any')
@@ -24,7 +24,7 @@ checkdepends=('python-pytest'
conflicts=('simple_backup-git') conflicts=('simple_backup-git')
source=("git+$url?signed#tag=$pkgver") source=("git+$url?signed#tag=$pkgver")
validpgpkeys=('7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514') validpgpkeys=('7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514')
sha256sums=('cc5b378ac489f7cf3e3a38855dfd530a79f0651a1891ea83355b952ed3a20aee') sha256sums=('3646012b0d50ca2cc1adcd7904bb34389e8d33724534730b48ae30de6b28f994')
build() build()
{ {
+12
View File
@@ -62,6 +62,18 @@ 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. 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. 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`).
It also saves the number of files, so that the next run can show progress and an estimated time while rsync is still looking for files.
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 ## Remote backup
> **Warning** > **Warning**
> This feature is experimental > This feature is experimental
+21 -3
View File
@@ -1,4 +1,4 @@
.TH SIMPLE_BACKUP 1 2026-10-21 SIMPLE_BACKUP 4.5.0 .TH SIMPLE_BACKUP 1 2026-11-02 SIMPLE_BACKUP 4.6.2
.SH NAME .SH NAME
simple_backup \- Backup files and folders using rsync simple_backup \- Backup files and folders using rsync
.SH SYNOPSIS .SH SYNOPSIS
@@ -108,8 +108,9 @@ Don't use systemd journal for logging.
.B \-\-no\-progress .B \-\-no\-progress
Don't show a progress bar while rsync copies the files. The progress bar is only shown when 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. 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 While rsync is still looking for files, the percentage and estimated time (marked with
and estimated time are only approximate. \(aq~\(aq) are based on the number of files in the last backup, or the bar shows
\(aq(scanning)\(aq if that is unknown or already exceeded.
.TP .TP
.B \-\-rsync\-options OPTIONS [OPTION...] .B \-\-rsync\-options OPTIONS [OPTION...]
By default, the following rsync options are used: By default, the following rsync options are used:
@@ -145,6 +146,11 @@ needs to be allowed to run without password in the same way.
.TP .TP
.B \-\-numeric\-ids .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. 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 .SH CONFIGURATION
An example configuration file is provided at \(aq/usr/share/doc/simple_backup/simple_backup.conf\(aq. 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. Copy it to the default location ($HOME/.config/simple_backup) and edit it as needed.
@@ -216,6 +222,18 @@ SSH connection failed.
.TP .TP
.B 6 .B 6
Bad configuration file or options (for example, only one of SSH host and SSH user specified). 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), and the third line, if present, the number of files, used
for the progress bar estimate (only counted when the progress bar is shown, otherwise the previous
count is kept). 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 .SH SEE ALSO
.BR rsync (1) .BR rsync (1)
.SH AUTHORS .SH AUTHORS
+1 -1
View File
@@ -1,3 +1,3 @@
"""Init.""" """Init."""
__version__ = '4.5.0' __version__ = '4.6.2'
+187 -37
View File
@@ -25,9 +25,8 @@ import shlex
import sys import sys
import warnings import warnings
from collections import deque from collections import deque
from collections.abc import Callable, Iterator from collections.abc import Iterator
from datetime import datetime from datetime import datetime
from functools import wraps
from getpass import GetPassWarning, getpass from getpass import GetPassWarning, getpass
from glob import glob from glob import glob
from logging import StreamHandler from logging import StreamHandler
@@ -35,7 +34,7 @@ from shutil import get_terminal_size, rmtree, which
from subprocess import PIPE, STDOUT, Popen from subprocess import PIPE, STDOUT, Popen
from tempfile import mkstemp from tempfile import mkstemp
from timeit import default_timer from timeit import default_timer
from typing import Any, ParamSpec, TypedDict, TypeVar from typing import Any, TypedDict
try: try:
import paramiko import paramiko
@@ -69,9 +68,6 @@ if journal:
j_handler.setFormatter(j_format) j_handler.setFormatter(j_format)
logger.addHandler(j_handler) logger.addHandler(j_handler)
P = ParamSpec('P')
R = TypeVar('R')
# Seconds to wait for the TCP connection to the SSH server # Seconds to wait for the TCP connection to the SSH server
SSH_TIMEOUT = 15 SSH_TIMEOUT = 15
@@ -80,25 +76,15 @@ PROGRESS_RE = re.compile(r'\s*(\S+)\s+\d+%\s+\S+\s+\d+:\d\d:\d\d(?:\s+\(xfr#\d+,
PROGRESS_BAR_WIDTH = 30 PROGRESS_BAR_WIDTH = 30
def timing(func: Callable[P, R]) -> Callable[P, R]: def _format_duration(seconds: float) -> str:
"""Decorator to measure execution time of a function """Format seconds as mm:ss, or h:mm:ss from one hour"""
Parameters: s = round(seconds)
func: Function to decorate
"""
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
start = default_timer()
value = func(*args, **kwargs) if s >= 3600:
return f'{s // 3600}:{s // 60 % 60:02d}:{s % 60:02d}'
end = default_timer() return f'{s // 60:02d}:{s % 60:02d}'
logger.info('Elapsed time: %.3f seconds', end - start)
return value
return wrapper
class MyFormatter(argparse.RawTextHelpFormatter, argparse.ArgumentDefaultsHelpFormatter): class MyFormatter(argparse.RawTextHelpFormatter, argparse.ArgumentDefaultsHelpFormatter):
@@ -108,14 +94,27 @@ class MyFormatter(argparse.RawTextHelpFormatter, argparse.ArgumentDefaultsHelpFo
class _ProgressBar: class _ProgressBar:
"""Draws rsync --info=progress2 updates as a bar on stderr""" """Draws rsync --info=progress2 updates as a bar on stderr"""
def __init__(self) -> None: def __init__(self, estimate: int = 0) -> None:
self._start = default_timer() self._start = default_timer()
# rsync reports ir-chk while incremental recursion is still finding files # rsync reports ir-chk while incremental recursion is still finding files
self._scanning = True self._scanning = True
self._checked = 0 self._checked = 0
self._total = 0 self._total = 0
# Total files of the last backup, used until rsync knows the real total
self._estimate = estimate
self._shown = False self._shown = False
@property
def total(self) -> int | None:
"""Total number of files, once rsync has found them all"""
return None if self._scanning or self._total == 0 else self._total
def _eta(self, total: int) -> str:
eta = int((default_timer() - self._start) * (total - self._checked) / self._checked)
return f'{eta // 3600}:{eta // 60 % 60:02d}:{eta % 60:02d}'
def update(self, record: str) -> bool: def update(self, record: str) -> bool:
"""Draw the bar if record is a progress update, and return whether it was""" """Draw the bar if record is a progress update, and return whether it was"""
@@ -133,20 +132,29 @@ class _ProgressBar:
self._checked = self._total - int(m.group(3)) 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 # 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: percent: int | None
text = f'{self._checked:,} files checked {size} copied (scanning)'
percent = None if self.total is None:
# Estimate from the last backup, unless this one already has more files
if 0 < self._checked < self._estimate:
percent = self._checked * 100 // self._estimate
text = f'{f"~{percent}":>3}% {self._checked:,}/~{self._estimate:,} files {size} copied'
if percent > 0:
text += f' ETA ~{self._eta(self._estimate)}'
else:
text = f'{self._checked:,} files checked {size} copied (scanning)'
percent = None
else: else:
percent = self._checked * 100 // self._total percent = self._checked * 100 // self._total
text = f'{percent:3d}% {self._checked:,}/{self._total:,} files {size} copied' text = f'{percent:3d}% {self._checked:,}/{self._total:,} files {size} copied'
if 0 < percent < 100: if 0 < percent < 100:
eta = int((default_timer() - self._start) * (self._total - self._checked) / self._checked) text += f' ETA {self._eta(self._total)}'
text += f' ETA {eta // 3600}:{eta // 60 % 60:02d}:{eta % 60:02d}'
width = get_terminal_size().columns - 1 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 # Fixed width, so the bar doesn't jump when the text changes length; no bar without a total or estimate
if percent is not None and len(text) + PROGRESS_BAR_WIDTH + 4 <= width: if percent is not None and len(text) + PROGRESS_BAR_WIDTH + 4 <= width:
filled = PROGRESS_BAR_WIDTH * percent // 100 filled = PROGRESS_BAR_WIDTH * percent // 100
text = f'[{"#" * filled}{"-" * (PROGRESS_BAR_WIDTH - filled)}] {text}' text = f'[{"#" * filled}{"-" * (PROGRESS_BAR_WIDTH - filled)}] {text}'
@@ -209,7 +217,7 @@ class Backup:
keep: int | None, options: str, keep: int | None, options: str,
ssh_host: str | None = None, ssh_user: str | None = None, ssh_keyfile: str | None = None, 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, remote_sudo: bool = False, remove_before: bool = False, verbose: bool = False,
progress: bool = False) -> None: progress: bool = False, file_estimate: int = 0) -> None:
# Missing values are reported by check_params() # Missing values are reported by check_params()
self.inputs = inputs if inputs is not None else [] self.inputs = inputs if inputs is not None else []
self.output = output if output is not None else '' self.output = output if output is not None else ''
@@ -223,6 +231,9 @@ class Backup:
self._remove_before = remove_before self._remove_before = remove_before
self._verbose = verbose self._verbose = verbose
self._progress = progress self._progress = progress
self._file_estimate = file_estimate
# Files in the backup, known only with the progress bar
self.file_count: int | None = None
self._last_backup = '' self._last_backup = ''
self._server = '' self._server = ''
self._output_dir = '' self._output_dir = ''
@@ -233,6 +244,7 @@ class Backup:
# Passed to sshpass through rsync's environment only # Passed to sshpass through rsync's environment only
self._password: str | None = None self._password: str | None = None
self._removed_count = 0 self._removed_count = 0
self._start_time = 0.0
def check_params(self, homedir: str = '') -> int: def check_params(self, homedir: str = '') -> int:
"""Check if parameters for the backup are valid""" """Check if parameters for the backup are valid"""
@@ -315,6 +327,12 @@ class Backup:
if self._remote: if self._remote:
self._server = f'{self.ssh_user}@{self.ssh_host}:' 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: def remove_old_backups(self) -> None:
"""Remove old backups if there are more than indicated by 'keep'""" """Remove old backups if there are more than indicated by 'keep'"""
@@ -663,14 +681,19 @@ class Backup:
'Rsync error (return code %d) - Check rsync(1) for details', returncode) 'Rsync error (return code %d) - Check rsync(1) for details', returncode)
# Function to read configuration file # Function to read configuration file
@timing
def run(self) -> int: def run(self) -> int:
"""Perform the backup""" """Perform the backup"""
self._start_time = default_timer()
try: try:
return self._run() return self._run()
finally: finally:
self._remove_temp_files() self._remove_temp_files()
logger.info('Elapsed time: %s', self._elapsed())
def _elapsed(self) -> str:
return _format_duration(default_timer() - self._start_time)
def close(self) -> None: def close(self) -> None:
"""Close the SSH connection, if open""" """Close the SSH connection, if open"""
@@ -770,7 +793,7 @@ class Backup:
if self._progress: if self._progress:
args.append('--info=progress2') args.append('--info=progress2')
bar = _ProgressBar() bar = _ProgressBar(self._file_estimate)
with Popen(args, stdin=PIPE, stdout=PIPE, stderr=STDOUT, shell=False, env=env) 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) assert p.stdin is not None and isinstance(p.stdout, io.BufferedIOBase)
@@ -789,6 +812,9 @@ class Backup:
returncode = p.wait() returncode = p.wait()
if bar is not None:
self.file_count = bar.total
if returncode == 0: if returncode == 0:
# Without -v, only the summary, which may be fewer than two lines (e.g. with --rsync-options q) # Without -v, only the summary, which may be fewer than two lines (e.g. with --rsync-options q)
for r in output: for r in output:
@@ -825,14 +851,14 @@ class Backup:
logger.error( logger.error(
'Some errors occurred while performing the backup') 'Some errors occurred while performing the backup')
_notify( _notify(f'Some errors occurred while performing the backup (elapsed time {self._elapsed()}). '
'Some errors occurred while performing the backup. Check log for details') 'Check log for details')
return 4 return 4
logger.info('Backup completed') logger.info('Backup completed')
_notify('Backup completed') _notify(f'Backup completed in {self._elapsed()}')
return 0 return 0
@@ -902,6 +928,8 @@ def _parse_arguments() -> argparse.Namespace:
help='Run rsync on remote server with sudo if allowed') help='Run rsync on remote server with sudo if allowed')
parser.add_argument('--numeric-ids', action='store_true', parser.add_argument('--numeric-ids', action='store_true',
help='Use rsync \'--numeric-ids\' option (don\'t map uid/gid values by name)') 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() args = parser.parse_args()
@@ -1077,6 +1105,119 @@ def _notify(text: str) -> None:
os.seteuid(euid) 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 _last_file_count(user: str | None) -> int:
"""Files in the last backup (third line of the state file), or 0 if unknown"""
try:
with open(_state_file(_state_owner(user)), encoding='utf-8') as fp:
return int(fp.read().splitlines()[2])
except (OSError, ValueError, IndexError):
return 0
def _write_state(user: str | None, destination: str, file_count: int | None = None) -> None:
"""Record the time, directory and file count of the last successful backup (one per line)"""
owner = _state_owner(user)
path = _state_file(owner)
tmp = ''
# Without the progress bar the count is unknown, so keep the previous one
if file_count is None:
file_count = _last_file_count(user)
count_line = f'{file_count}\n' if file_count > 0 else ''
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{count_line}')
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: def simple_backup() -> int:
"""Main""" """Main"""
@@ -1084,6 +1225,9 @@ def simple_backup() -> int:
user: str | None = args.user or _detect_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: if not args.user and os.geteuid() != 0:
homedir = os.getenv('HOME', '') homedir = os.getenv('HOME', '')
elif user is not None: elif user is not None:
@@ -1160,15 +1304,21 @@ def simple_backup() -> int:
if args.numeric_ids or config_args['numeric_ids']: if args.numeric_ids or config_args['numeric_ids']:
rsync_options.append('--numeric-ids') rsync_options.append('--numeric-ids')
progress = not args.no_progress and sys.stderr.isatty()
backup = Backup(inputs, output, exclude, keep, ' '.join(rsync_options), ssh_host, ssh_user, ssh_keyfile, 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, remote_sudo, remove_before=args.remove_before_backup, verbose=args.verbose,
progress=not args.no_progress and sys.stderr.isatty()) progress=progress, file_estimate=_last_file_count(user) if progress else 0)
try: try:
return_code = backup.check_params(homedir) return_code = backup.check_params(homedir)
if return_code != 0:
return return_code
return_code = backup.run()
if return_code == 0: if return_code == 0:
return backup.run() _write_state(user, backup.destination, backup.file_count)
return return_code return return_code
finally: finally:
+4 -2
View File
@@ -3,6 +3,7 @@
import sys import sys
from collections.abc import Callable, Iterator from collections.abc import Callable, Iterator
from datetime import datetime, timedelta from datetime import datetime, timedelta
from pathlib import Path
import pytest import pytest
@@ -10,13 +11,14 @@ from simple_backup import simple_backup as sb
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def no_side_effects(monkeypatch: pytest.MonkeyPatch) -> None: def no_side_effects(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Keep tests out of the systemd journal and the desktop notifications""" """Keep tests out of the systemd journal, the desktop notifications and the user's state directory"""
if hasattr(sb, 'j_handler'): if hasattr(sb, 'j_handler'):
sb.logger.removeHandler(sb.j_handler) sb.logger.removeHandler(sb.j_handler)
monkeypatch.setattr(sb, '_notify', lambda text: None) monkeypatch.setattr(sb, '_notify', lambda text: None)
monkeypatch.setenv('XDG_STATE_HOME', str(tmp_path / 'state'))
@pytest.fixture @pytest.fixture
+35 -2
View File
@@ -26,11 +26,14 @@ class FakeBackup:
instances: list['FakeBackup'] = [] instances: list['FakeBackup'] = []
check_code = 0 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, def __init__(self, inputs: Any, output: Any, exclude: Any, keep: Any, options: str,
ssh_host: str | None = None, ssh_user: str | None = None, ssh_host: str | None = None, ssh_user: str | None = None,
ssh_keyfile: str | None = None, remote_sudo: bool = False, ssh_keyfile: str | None = None, remote_sudo: bool = False,
remove_before: bool = False, verbose: bool = False, progress: bool = False) -> None: remove_before: bool = False, verbose: bool = False, progress: bool = False,
file_estimate: int = 0) -> None:
self.inputs = inputs self.inputs = inputs
self.output = output self.output = output
self.exclude = exclude self.exclude = exclude
@@ -43,6 +46,8 @@ class FakeBackup:
self.remove_before = remove_before self.remove_before = remove_before
self.verbose = verbose self.verbose = verbose
self.progress = progress self.progress = progress
self.file_estimate = file_estimate
self.file_count: int | None = None
self.homedir: str | None = None self.homedir: str | None = None
self.ran = False self.ran = False
self.closed = False self.closed = False
@@ -56,7 +61,7 @@ class FakeBackup:
def run(self) -> int: def run(self) -> int:
self.ran = True self.ran = True
return 0 return FakeBackup.run_code
def close(self) -> None: def close(self) -> None:
self.closed = True self.closed = True
@@ -66,6 +71,7 @@ class FakeBackup:
def fake_backup(monkeypatch: pytest.MonkeyPatch) -> None: def fake_backup(monkeypatch: pytest.MonkeyPatch) -> None:
FakeBackup.instances = [] FakeBackup.instances = []
FakeBackup.check_code = 0 FakeBackup.check_code = 0
FakeBackup.run_code = 0
monkeypatch.setattr(sb, 'Backup', FakeBackup) monkeypatch.setattr(sb, 'Backup', FakeBackup)
@@ -83,6 +89,8 @@ def as_root(monkeypatch: pytest.MonkeyPatch) -> None:
"""Run as root, with no sudo/doas variables set""" """Run as root, with no sudo/doas variables set"""
monkeypatch.setattr(os, 'geteuid', lambda: 0) 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, file_count=None: None)
for var in ['SUDO_USER', 'SUDO_UID', 'DOAS_USER']: for var in ['SUDO_USER', 'SUDO_UID', 'DOAS_USER']:
monkeypatch.delenv(var, raising=False) monkeypatch.delenv(var, raising=False)
@@ -261,6 +269,17 @@ ssh_keyfile=/keys/id
assert run_cli('-c', config('backup_dir=/backups'), *argv) == 0 assert run_cli('-c', config('backup_dir=/backups'), *argv) == 0
assert last_backup().progress is progress assert last_backup().progress is progress
@pytest.mark.parametrize('tty, estimate', [(True, 1234), (False, 0)])
def test_file_estimate(self, run_cli: RunCli, config: Callable[[str], str], monkeypatch: pytest.MonkeyPatch,
tmp_path: Path, tty: bool, estimate: int) -> None:
monkeypatch.setattr(sb.sys.stderr, 'isatty', lambda: tty)
state = tmp_path / 'state' / 'simple_backup' / 'last_backup'
state.parent.mkdir(parents=True)
state.write_text('2026-10-25T10:00:00+02:00\n/backups/old\n1234\n')
assert run_cli('-c', config('backup_dir=/backups')) == 0
assert last_backup().file_estimate == estimate
def test_check_params_failure(self, run_cli: RunCli, config: Callable[[str], str]) -> None: def test_check_params_failure(self, run_cli: RunCli, config: Callable[[str], str]) -> None:
FakeBackup.check_code = 2 FakeBackup.check_code = 2
@@ -270,6 +289,20 @@ ssh_keyfile=/keys/id
assert not b.ran assert not b.ran
assert b.closed 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: class TestUser:
@pytest.mark.usefixtures('as_user') @pytest.mark.usefixtures('as_user')
+61 -1
View File
@@ -88,6 +88,18 @@ class TestRsync:
assert 'Backup completed' in caplog.text assert 'Backup completed' in caplog.text
assert 'rsync: sent' in caplog.text assert 'rsync: sent' in caplog.text
def test_elapsed_time(self, src: Path, out: Path, monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture) -> None:
notifications: list[str] = []
monkeypatch.setattr(sb, '_notify', notifications.append)
# Start, then the notification and the log line
monkeypatch.setattr(sb, 'default_timer', iter([0.0, 863.187, 863.2]).__next__)
assert run_backup(out, [str(src)]) == 0
assert notifications[-1] == 'Backup completed in 14:23'
assert 'Elapsed time: 14:23' in caplog.text
def test_hard_links(self, src: Path, out: Path) -> None: def test_hard_links(self, src: Path, out: Path) -> None:
assert run_backup(out, [str(src)]) == 0 assert run_backup(out, [str(src)]) == 0
(src / 'changed.txt').write_text('second') (src / 'changed.txt').write_text('second')
@@ -195,13 +207,30 @@ class TestRsync:
assert 'rsync: total size' in caplog.text assert 'rsync: total size' in caplog.text
assert '%' not in caplog.text assert '%' not in caplog.text
def test_file_count(self, src: Path, out: Path) -> None:
backup = make_backup(out, [str(src)], progress=True)
assert backup.run() == 0
# The source folder, its two files and the implied parent folders
assert backup.file_count is not None and backup.file_count >= 3
def test_no_file_count_without_progress(self, src: Path, out: Path) -> None:
backup = make_backup(out, [str(src)])
assert backup.run() == 0
assert backup.file_count is None
def test_cli(self, src: Path, out: Path, tmp_path: Path, run_cli: Callable[..., int]) -> None: def test_cli(self, src: Path, out: Path, tmp_path: Path, run_cli: Callable[..., int]) -> None:
config_file = str(tmp_path / 'missing.conf') config_file = str(tmp_path / 'missing.conf')
for _ in range(2): for _ in range(2):
assert run_cli('-c', config_file, '-i', str(src), '-o', str(out), '-k', '0') == 0 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): class ChunkStream(io.BufferedIOBase):
@@ -227,6 +256,13 @@ class TestReadRecords:
assert list(sb._read_records(stream)) == ['file', '', 'namé', '�'] assert list(sb._read_records(stream)) == ['file', '', 'namé', '�']
@pytest.mark.parametrize(('seconds', 'text'), [
(0.4, '00:00'), (59.6, '01:00'), (863.187, '14:23'), (3599.4, '59:59'), (3600, '1:00:00'), (37230, '10:20:30'),
])
def test_format_duration(seconds: float, text: str) -> None:
assert sb._format_duration(seconds) == text
class TestProgressBar: class TestProgressBar:
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def terminal(self, monkeypatch: pytest.MonkeyPatch) -> None: def terminal(self, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -257,6 +293,30 @@ class TestProgressBar:
assert bar.update(' 3.00M 2% 1.47MB/s 0:00:03') 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') assert capsys.readouterr().err.endswith('] 50% 2/4 files 3.00M copied ETA 0:00:30\033[K')
def test_estimate_while_scanning(self, monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str]) -> None:
monkeypatch.setattr(sb, 'get_terminal_size', lambda: os.terminal_size((100, 24)))
bar = sb._ProgressBar(estimate=4000)
assert bar.update(' 2.00M 25% 1.47MB/s 0:00:02 (xfr#1, ir-chk=1000/2000)')
assert bar.total is None
assert capsys.readouterr().err == \
'\r[#######-----------------------] ~25% 1,000/~4,000 files 2.00M copied ETA ~0:01:30\033[K'
def test_estimate_exceeded(self, capsys: pytest.CaptureFixture[str]) -> None:
bar = sb._ProgressBar(estimate=1000)
# More files than in the last backup: back to plain scanning
assert bar.update(' 2.00M 25% 1.47MB/s 0:00:02 (xfr#1, ir-chk=500/2000)')
assert capsys.readouterr().err == '\r1,500 files checked 2.00M copied (scanning)\033[K'
def test_real_total_replaces_estimate(self, capsys: pytest.CaptureFixture[str]) -> None:
bar = sb._ProgressBar(estimate=4000)
assert bar.update(' 4,000,000 1% 2.97MB/s 0:00:01 (xfr#2, to-chk=1/4)')
assert bar.total == 4
assert ' 75% 3/4 files' in capsys.readouterr().err
def test_not_progress(self, capsys: pytest.CaptureFixture[str]) -> None: def test_not_progress(self, capsys: pytest.CaptureFixture[str]) -> None:
bar = sb._ProgressBar() bar = sb._ProgressBar()
+160
View File
@@ -0,0 +1,160 @@
"""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_file_count(state: Path) -> None:
sb._write_state(None, DESTINATION, 1234)
assert state.read_text().splitlines()[2] == '1234'
assert sb._last_file_count(None) == 1234
def test_file_count_kept_when_unknown(state: Path) -> None:
write_state(state, '2026-10-01T10:00:00+02:00\n/old\n1234\n')
sb._write_state(None, DESTINATION)
assert state.read_text().splitlines()[1:] == [DESTINATION, '1234']
@pytest.mark.parametrize('text', [None, '2026-10-01T10:00:00+02:00\n/old\n', '2026-10-01T10:00:00+02:00\n/old\nx\n'])
def test_no_file_count(state: Path, text: str | None) -> None:
if text is not None:
write_state(state, text)
assert sb._last_file_count(None) == 0
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